From daf7463e7a35c5dc596edea5a0610b8710775a97 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Thu, 27 Aug 2026 12:07:49 +0200 Subject: [PATCH] refactor(ai-chat): lock the composer in watching tabs and own run position in one module Co-Authored-By: Claude Opus 5 --- .../copilot/chat/AIChatDisplay.svelte | 18 +- .../copilot/chat/AIChatInput.svelte | 7 + .../copilot/chat/AIChatManager.svelte.ts | 54 +++-- .../copilot/chat/AIChatManager.test.ts | 26 +- .../sessions/sessionRunOwner.svelte.ts | 228 ++++++++++++++++++ .../sessions/sessionRuntime.svelte.ts | 127 ++++------ .../components/sessions/sessionSync.svelte.ts | 173 ++----------- 7 files changed, 372 insertions(+), 261 deletions(-) create mode 100644 frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts diff --git a/frontend/src/lib/components/copilot/chat/AIChatDisplay.svelte b/frontend/src/lib/components/copilot/chat/AIChatDisplay.svelte index cbfc6c9e6e..ce192356c6 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatDisplay.svelte +++ b/frontend/src/lib/components/copilot/chat/AIChatDisplay.svelte @@ -499,9 +499,8 @@ const availableAutonomyModeOptions = $derived( autonomyModeOptions.filter((option) => option.isAvailable(autonomyAvailability)) ) - // Only when this tab could hold the posture itself: the label is a promise that - // the workspace stays read-only, and the takeover in `runGuard` keeps it by - // entering plan mode here — which an unavailable mode would silently refuse. + // Only when this tab could hold the posture itself: a mode its own selector + // does not offer would read as a mode the user could switch away from here. const showsMirroredPlan = $derived( aiChatManager.mirroredPlanMode && aiChatManager.planModeAvailable ) @@ -536,6 +535,17 @@ return pending?.action === 'question' ? pending.toolCallId : undefined }) + // The composer is locked while another tab's run is on screen: this tab pairs + // a mirrored transcript with the history it held before that run, and a turn + // sent from that pair would reach the model as a conversation the driver has + // already moved past. It unlocks on its own once the re-read that follows the + // turn lands. A run parked on a question is the exception — answering is the + // one thing a watching tab is there to do, and the answer travels to the + // driver instead of starting a turn here. + const composerLocked = $derived( + aiChatManager.mirroringRemoteRun && pendingQuestionToolCallId === undefined + ) + // Get app context for display when in APP mode const appContext = $derived.by((): SelectedContext | undefined => { if (aiChatManager.mode !== AIMode.APP || !aiChatManager.appAiChatHelpers) { @@ -819,7 +829,7 @@ the panel, or the Escape-to-stop focus check would wrongly reject them. --> {initialInstructions} {onDraftChange} showContext={aiChatManager.mode !== AIMode.GLOBAL} - {disabled} + disabled={disabled || composerLocked} {pendingQuestionToolCallId} isFirstMessage={messages.length === 0} /> diff --git a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte index 14182d555f..a39d003998 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte +++ b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte @@ -133,6 +133,13 @@ return 'Answer the question above' } + // Ahead of 'Ask followup': a mirrored run is almost always mid-conversation, + // so a check below that one would never be reached, and the locked composer + // would sit there inviting a followup it will not accept. + if (aiChatManager.mirroringRemoteRun) { + return 'Running in another tab' + } + if (!isFirstMessage) { return 'Ask followup' } diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index d7e0c44f2b..b870cf9b79 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -61,6 +61,7 @@ import { SvelteMap, SvelteSet } from 'svelte/reactivity' import { createLongHash } from '$lib/editorLangUtils' import type { AIProvider, UserDraftItemKind } from '$lib/gen' import { maskKey } from '$lib/components/sessions/modifiedItemsMask' +import { isMirroring, runPosition } from '$lib/components/sessions/sessionRunOwner.svelte' import { getStringError } from './utils' import { type PasteAttachment } from './pasteTokens' import { @@ -694,32 +695,35 @@ export class AIChatManager { remoteToolConfirmation: ((toolId: string, confirmed: boolean) => boolean) | undefined = undefined remoteQuestionAnswer: ((toolId: string, choices: string[]) => boolean) | undefined = undefined - /** True while this manager is rendering a run another tab is driving. Its - * transcript then comes from mirror frames while `messages` still holds the - * pre-run history, so the pair is mismatched and must not be written to the - * shared record — the driving tab owns it until the run ends. Set by the - * session runtime. */ - mirroringRemoteRun = $state(false) + /** True while this manager is showing a run another tab is driving, and on + * through the re-read that follows it. Its transcript then comes from mirror + * frames while `messages` still holds the pre-run history, so the pair is + * mismatched and must not be written to the shared record — the driving tab + * owns it until the re-read lands. + * + * Derived rather than stored: the position is what the runtime moves, and a + * second copy of it here could only ever be a copy that disagrees. A chat + * with no session (the docked copilot) is never mirroring. */ + get mirroringRemoteRun(): boolean { + return isMirroring(this.sessionId) + } - /** Whether the tab driving this session was last seen in plan mode. + /** Whether the run on screen is one another tab is running in plan mode. * - * Plan mode is the one autonomy state that is deliberately never persisted - * (see `persistAutonomyMode`): a model entered it for this session, so it - * lives only in the memory of the tab running the turn. Every other mode is - * a stored preference each tab is entitled to its own copy of. - * - * Kept after the turn ends rather than cleared with `mirroringRemoteRun`, - * because the driver stays in plan mode between turns and stops sending - * frames that could say so. The next turn's first frame corrects it; this - * tab driving one of its own clears it outright. */ - mirroredPlanMode = $state(false) + * Plan mode is the one autonomy state deliberately never persisted (see + * `persistAutonomyMode`): a model entered it for this session, so it lives + * only in the memory of the tab running the turn. It is read from the run's + * position for the same reason it is not stored there — the posture belongs + * to that run, and must not outlive it into a turn this tab drives. */ + get mirroredPlanMode(): boolean { + const position = runPosition(this.sessionId) + return position.state === 'watching' && position.planMode + } /** Whether the turn that just finished is one a follow-up should be sent * after: it committed, or the user deliberately stopped it. False through a * provider error, an empty-response rollback, or a programmatic cancel — the - * states this manager deliberately keeps its own queued message through. The - * session runtime reports it to the other tabs, whose queues follow the same - * rule. */ + * states this manager deliberately keeps its own queued message through. */ lastTurnAcceptsFollowUp = false // Workspace items the CURRENT chat modified via AI tool calls, as @@ -2880,13 +2884,13 @@ export class AIChatManager { sendUserToast('This action needs the AI chat. Start an AI session to continue.', true) return } - // Still holding a mirrored transcript against the pre-turn history: the - // other tab's run has ended but this one has not finished reading what it - // left behind. Sending now would put that stale history to the model and - // persist it over the driver's completed turn. + // Still pairing a mirrored transcript with the pre-run history: sending now + // would put a conversation the driving tab has already moved past to the + // model, and persist it over what that tab saved. The composer is locked + // for exactly this window, so nothing the user typed reaches here — this + // catches the sends that start without one (tool handlers, auto-resume). if (this.mirroringRemoteRun) { this.restoreRefusedSend(options) - sendUserToast('Catching up on the turn that just finished. Try again in a moment.', true) return false } this.#sendsInFlight++ diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index 1fc4dc697c..74d22391bc 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts @@ -11,6 +11,10 @@ import { chatState } from './sharedChatState.svelte' import { PLAN_MODE_MESSAGES } from './planModeMessages' import { runChatLoop } from './chatLoop' import { clearWorkspaceRoleCache } from '$lib/user' +import { + noteDriverAlive, + noteRemoteTurnEnded +} from '$lib/components/sessions/sessionRunOwner.svelte' // This suite forces esm-env BROWSER=true (below). That makes @sveltejs/kit's // client runtime (pulled transitively via $lib/navigation) evaluate browser-only @@ -295,8 +299,9 @@ describe('AIChatManager cross-tab run guard', () => { it('does not persist while rendering a run another tab owns', async () => { const manager = new AIChatManager() manager.isSessionChat = true + manager.sessionId = 'session-watching' const saveChat = vi.spyOn(manager.historyManager, 'saveChat').mockResolvedValue(undefined) - manager.mirroringRemoteRun = true + noteDriverAlive('session-watching', false) await manager.saveAndClear() await manager.compactManually() @@ -304,6 +309,21 @@ describe('AIChatManager cross-tab run guard', () => { expect(saveChat).not.toHaveBeenCalled() }) + // The posture describes the run it arrived with, and the driving tab stops + // saying anything about it once that run is over. Outliving it is how a turn + // driven from this tab ends up promising a plan mode it is not in. + it('drops the mirrored plan posture when the remote run ends', () => { + const manager = new AIChatManager() + manager.isSessionChat = true + manager.sessionId = 'session-planning' + + noteDriverAlive('session-planning', true) + expect(manager.mirroredPlanMode).toBe(true) + + noteRemoteTurnEnded('session-planning') + expect(manager.mirroredPlanMode).toBe(false) + }) + // The other tab's run has ended but this one is still reading what it left // behind, so its transcript is mirrored while `messages` is the pre-turn // history. A send in that window would put the stale history to the model and @@ -311,7 +331,9 @@ describe('AIChatManager cross-tab run guard', () => { it('refuses to send while still catching up on a finished remote turn', async () => { const manager = new AIChatManager() manager.isSessionChat = true - manager.mirroringRemoteRun = true + manager.sessionId = 'session-catching-up' + noteDriverAlive('session-catching-up', false) + noteRemoteTurnEnded('session-catching-up') const restoreInstructions = vi.fn(() => true) manager.setAiChatInput({ restoreInstructions } as any) diff --git a/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts new file mode 100644 index 0000000000..04ca8e77ea --- /dev/null +++ b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts @@ -0,0 +1,228 @@ +import { BROWSER } from 'esm-env' +import { SvelteMap } from 'svelte/reactivity' + +// Who holds a session's run, for every tab open on it. +// +// Two tabs sending on one session append to the same chat id, and whichever +// saveChat lands last silently drops the other's turn — after its tool calls +// have already run against the workspace. So one tab drives and the rest watch. +// +// Every gate in the cross-tab paths asks a variant of one question: may I send, +// is this frame mine to adopt, is what I am showing still the other tab's. This +// module answers all of them from a single position per session, so they cannot +// disagree with each other. + +/** Whether this context can hold a real lock on a run. Web Locks is + * secure-context only, so a self-hosted instance served over plain HTTP has + * none — watching still works there, on the weaker footing described at + * {@link withSessionRunLock} and {@link runLockHeld}. */ +const EXCLUSIVE_OWNERSHIP = BROWSER && !!globalThis.navigator?.locks?.query + +/** A driver with no frame for this long is presumed gone, pending the lock + * query that confirms it. Deliberately many times the frame cadence: a busy + * tab can be starved of ticks for a while without having gone anywhere. */ +const DRIVER_SILENCE_MS = 10_000 + +/** Where this tab stands in one session's run. + * + * `catchingUp` is its own position rather than a corner of `watching`: the + * driver's turn is over, but this tab still pairs a mirrored transcript with + * the history it held before that turn, and sending from that pair would put a + * conversation the driver has already moved past to the model. */ +export type RunPosition = + | { state: 'idle' } + | { state: 'driving' } + | { + state: 'watching' + lastHeardAt: number + /** The driver's plan-mode posture, carried so a watching tab can show the + * mode the run is actually under. It lives here rather than on the + * manager so it cannot outlive the run it describes: the position moves + * on at turn end and the posture goes with it. */ + planMode: boolean + } + | { state: 'catchingUp' } + +const IDLE: RunPosition = { state: 'idle' } + +/** Reactive because the composer's locked state is derived from it. */ +const positions = new SvelteMap() + +/** Sessions with no entry are idle, which is also the answer for a chat that + * has no session at all (the docked side-panel copilot). */ +export function runPosition(sessionId: string | undefined): RunPosition { + if (!sessionId) return IDLE + return positions.get(sessionId) ?? IDLE +} + +/** This tab is running the turn. */ +export function isDriving(sessionId: string): boolean { + return runPosition(sessionId).state === 'driving' +} + +/** Another tab is running the turn and this one is rendering its frames. */ +export function isWatching(sessionId: string): boolean { + return runPosition(sessionId).state === 'watching' +} + +/** True from the first frame adopted until the re-read that follows the + * driver's turn completes. The save and send paths gate on it: until it + * clears, this tab's transcript and its model history are not the same + * conversation. */ +export function isMirroring(sessionId: string | undefined): boolean { + const state = runPosition(sessionId).state + return state === 'watching' || state === 'catchingUp' +} + +/** A frame arrived, which is both the transcript and the sign of life. */ +export function noteDriverAlive(sessionId: string, planMode: boolean): void { + // A tab mid-turn is the authority on its own session; a frame reaching it can + // only be an echo of the run it is itself driving. + if (isDriving(sessionId)) return + positions.set(sessionId, { state: 'watching', lastHeardAt: Date.now(), planMode }) + ensureReaper() +} + +/** The driver says its turn is over. The re-read that follows is what actually + * frees this tab, so the position moves to `catchingUp` rather than to idle. */ +export function noteRemoteTurnEnded(sessionId: string): void { + if (runPosition(sessionId).state !== 'watching') return + positions.set(sessionId, { state: 'catchingUp' }) +} + +/** The re-read finished: this tab's transcript and history are one conversation + * again, and it may drive the next turn. */ +export function noteCaughtUp(sessionId: string): void { + if (runPosition(sessionId).state !== 'catchingUp') return + positions.delete(sessionId) +} + +/** Drop everything held for a session whose runtime is going away, so a torn + * down tab can't leave a position behind for a session nothing is watching. */ +export function clearRunPosition(sessionId: string): void { + positions.delete(sessionId) +} + +function lockName(sessionId: string): string { + return `wm-session-run:${sessionId}` +} + +/** Run `body` as the session's sole driver, or return 'busy' without running it + * when another tab already holds the run. `body` runs at most once either way. + * + * Web Locks is what makes this safe across a crash: the lock is held by the tab, + * not by a record someone has to clean up, so a driver that dies mid-turn + * releases it and the next send succeeds. Where it is missing or unusable, + * {@link bestEffort} takes over. */ +export async function withSessionRunLock( + sessionId: string, + body: () => Promise +): Promise { + if (!EXCLUSIVE_OWNERSHIP) return await bestEffort(sessionId, body) + // Set before the body runs, so a turn that throws is told apart from a lock + // that could not be taken. Without it a failing turn would look like a failed + // arbitration and be run a second time by the fallback below. + let entered = false + try { + return (await navigator.locks.request( + lockName(sessionId), + { mode: 'exclusive', ifAvailable: true }, + async (lock) => { + // `ifAvailable` hands back a null lock instead of queueing when another + // tab holds it, which is exactly the "refuse, don't stack up turns" + // behavior we want. + if (!lock) return 'busy' as const + entered = true + return await drive(sessionId, body) + } + )) as T | 'busy' + } catch (e) { + if (entered) throw e + // The lock API is present but refused to arbitrate. Degrade to the footing + // a context without it already runs on rather than failing the turn: a + // session the user cannot send in is a worse outcome than one whose + // exclusion is best-effort for this send. + console.error('sessionRunOwner: run lock unavailable, excluding best-effort instead', e) + return await bestEffort(sessionId, body) + } +} + +/** Exclusion without a lock: refuse only when another tab is visibly mid-run. + * That covers the case that actually happens (one tab already going) and leaves + * a genuine simultaneous start racing, which is what a session already did + * before any of this. */ +async function bestEffort(sessionId: string, body: () => Promise): Promise { + if (isWatching(sessionId)) return 'busy' + return await drive(sessionId, body) +} + +async function drive(sessionId: string, body: () => Promise): Promise { + positions.set(sessionId, { state: 'driving' }) + try { + return await body() + } finally { + // Straight to idle: this tab wrote the turn it just ran, so there is + // nothing of anyone else's to catch up on. + positions.delete(sessionId) + } +} + +/** Whether any tab currently holds the run lock for this session. Used to + * settle a driver that stopped sending frames: a released lock proves the tab + * is gone, where silence alone only suggests it. */ +async function runLockHeld(sessionId: string): Promise { + // Nothing to consult without the lock API, so silence is the only evidence — + // and the caller only asks after a driver has gone quiet for the whole + // silence window. Reaping a driver that was merely starved costs a re-read + // the next frame would have triggered anyway, and cannot cost a turn: this + // tab still has to take the lock before it can send. + if (!EXCLUSIVE_OWNERSHIP) return false + try { + const state = await navigator.locks.query() + const name = lockName(sessionId) + return !!state.held?.some((l) => l.name === name) + } catch { + return true + } +} + +let driverLost: ((sessionId: string) => void) | undefined + +/** Registered by sessionRuntime at module load, so this module stays free of + * its imports — the two would otherwise sit in a cycle. */ +export function onDriverLost(fn: (sessionId: string) => void): void { + driverLost = fn +} + +// Runs only while some session is being driven elsewhere, and stops itself once +// none is — a browser with a single tab open never arms it at all. +let reaperTimer: ReturnType | undefined + +function ensureReaper(): void { + if (reaperTimer) return + reaperTimer = setInterval(() => { + if (![...positions.values()].some((p) => p.state === 'watching')) { + clearInterval(reaperTimer) + reaperTimer = undefined + return + } + void reapDeadDrivers() + }, DRIVER_SILENCE_MS) +} + +/** Release watchers whose driver went silent and no longer holds the lock, so a + * closed tab can't leave a session showing "generating" forever. */ +async function reapDeadDrivers(): Promise { + const now = Date.now() + const stale = [...positions.entries()] + .filter(([, p]) => p.state === 'watching' && now - p.lastHeardAt > DRIVER_SILENCE_MS) + .map(([id]) => id) + for (const id of stale) { + // Re-checked after the await: a frame may have landed while the query was + // in flight, and reaping then would tear down a run that is visibly alive. + if (await runLockHeld(id)) continue + if (!isWatching(id)) continue + noteRemoteTurnEnded(id) + driverLost?.(id) + } +} diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index 9add7b4438..cb0c9c1b27 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -1,11 +1,7 @@ import { SvelteMap } from 'svelte/reactivity' import { get } from 'svelte/store' import { base } from '$lib/base' -import { - AIAutonomyMode, - AIChatManager, - AIMode -} from '$lib/components/copilot/chat/AIChatManager.svelte' +import { AIChatManager, AIMode } from '$lib/components/copilot/chat/AIChatManager.svelte' import { PipelineEditorState } from '$lib/components/assets/AssetGraph/pipelineEditorState.svelte' import { initFlow } from '$lib/components/flows/flowStore.svelte' import { @@ -101,18 +97,23 @@ import { canSpliceFrame, mirrorFrameStart, withoutHeavyPayloads } from './sessio import { broadcastMirror, broadcastTurnEnd, - isLocallyDriven, - isRemotelyDriven, MIRROR_THROTTLE_MS, registerSyncHandlers, requestCancel, requestResync, sendQuestionAnswer, sendToolConfirmation, - withSessionRunLock, type MirrorMsg, type MirrorSnapshot } from './sessionSync.svelte' +import { + clearRunPosition, + isDriving, + isWatching, + noteCaughtUp, + onDriverLost, + withSessionRunLock +} from './sessionRunOwner.svelte' // Per-kind load state for a session's editor target. Pure state container the // load methods write into; the editor-target gate reads it to decide between @@ -940,16 +941,6 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { // turns interleave into one chat id. manager.runGuard = async (body) => { const outcome = await withSessionRunLock(session.id, async () => { - // The picker has read Plan since the last driver left the session in it, - // and this send is the user acting on what it says. Take the posture on - // rather than drop it: running under this tab's own autonomy instead - // would unblock the very workspace tools the label promised were held - // back. Once it is genuinely this manager's mode, the mirrored copy has - // nothing left to say. - if (manager.mirroredPlanMode) { - if (manager.planModeAvailable) manager.setAutonomyMode(AIAutonomyMode.PLAN) - manager.mirroredPlanMode = false - } // The first frame doubles as the "a run started here" signal: it is // posted immediately and carries the chat id the watchers need. startMirroring(session.id) @@ -959,14 +950,7 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { stopMirroring(session.id) // The chat id is re-read here, not reused from above: the turn may // have rotated it, and the listeners key their IndexedDB re-read on it. - broadcastTurnEnd( - session.id, - manager.historyManager.getCurrentChatId(), - // Not "did the send return truthy": that means the input was - // consumed, and stays true through provider errors and rollbacks. - // The manager reports the rule it applies to its own queue. - manager.lastTurnAcceptsFollowUp - ) + broadcastTurnEnd(session.id, manager.historyManager.getCurrentChatId()) } }) if (outcome === 'busy') { @@ -978,7 +962,7 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { // Stop is available wherever the run is visible, so from a watching tab it // has to travel to the one holding the turn. manager.remoteCancel = () => { - if (isLocallyDriven(session.id) || !isRemotelyDriven(session.id)) return false + if (!isWatching(session.id)) return false requestCancel(session.id) return true } @@ -986,12 +970,12 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { // Same for a run parked on the user: the tab showing the prompt is not // necessarily the tab whose loop is awaiting the answer. manager.remoteToolConfirmation = (toolId, confirmed) => { - if (isLocallyDriven(session.id) || !isRemotelyDriven(session.id)) return false + if (!isWatching(session.id)) return false sendToolConfirmation(session.id, toolId, confirmed) return true } manager.remoteQuestionAnswer = (toolId, choices) => { - if (isLocallyDriven(session.id) || !isRemotelyDriven(session.id)) return false + if (!isWatching(session.id)) return false sendQuestionAnswer(session.id, toolId, choices) return true } @@ -1114,7 +1098,7 @@ function stopMirroring(sessionId: string): void { function applyMirror(msg: MirrorMsg): void { // A tab mid-turn is the authority on its own session; a frame can only be // an echo of a run this tab is itself driving. - if (isLocallyDriven(msg.sessionId)) return + if (isDriving(msg.sessionId)) return const runtime = runtimes.get(msg.sessionId) if (!runtime) return const m = runtime.manager @@ -1145,23 +1129,17 @@ function applyMirror(msg: MirrorMsg): void { // holds complete from the store. m.displayMessages = msg.baseIndex === 0 ? msg.tail : [...m.displayMessages.slice(0, msg.baseIndex), ...msg.tail] - // The frame carries the rendered transcript but not the API-format history, - // so this manager is now holding a mismatched pair. Flag it: the save paths - // that run outside a turn would otherwise write that pair over the record the - // driving tab is still appending to. - m.mirroringRemoteRun = true m.loading = msg.loading m.currentReply = msg.currentReply m.currentReasoning = msg.currentReasoning m.currentReasoningActive = msg.currentReasoningActive m.loadingLabel = msg.loadingLabel m.compacting = msg.compacting - m.mirroredPlanMode = msg.planModeActive } /** The driver answers a resync with its whole transcript. */ function answerResync(sessionId: string): void { - if (!isLocallyDriven(sessionId)) return + if (!isDriving(sessionId)) return postMirror(sessionId, { full: true }) } @@ -1169,27 +1147,28 @@ function answerResync(sessionId: string): void { * only, so re-read the record the driver just saved for everything else — the * API-format history, context usage, the edits mask, background jobs — leaving * this tab able to take the conversation over. */ -async function applyTurnEnd(sessionId: string, chatId: string, committed: boolean): Promise { - if (isLocallyDriven(sessionId)) return - const runtime = runtimes.get(sessionId) - if (!runtime) return - const m = runtime.manager - // `loading` has to go first: loadPastChat refuses to run while the manager - // looks busy, and this one is the mirrored driver's, not a turn of our own. - m.loading = false - m.currentReply = '' - m.currentReasoning = '' - m.currentReasoningActive = false - m.loadingLabel = undefined - m.compacting = false - const id = chatId || m.historyManager.getCurrentChatId() +async function applyTurnEnd(sessionId: string, chatId: string): Promise { + if (isDriving(sessionId)) return try { + const runtime = runtimes.get(sessionId) + if (!runtime) return + const m = runtime.manager + // `loading` has to go first: loadPastChat refuses to run while the manager + // looks busy, and this one is the mirrored driver's, not a turn of our own. + m.loading = false + m.currentReply = '' + m.currentReasoning = '' + m.currentReasoningActive = false + m.loadingLabel = undefined + m.compacting = false + const id = chatId || m.historyManager.getCurrentChatId() if (!id) return const found = await m.historyManager.reloadChat(id) if (found === 'loaded') { - // `refresh`: the same conversation caught up from the store, so a message - // queued here while the other tab held the session is still meant for it. - // A plain load would drop it instead of sending it below. + // `refresh`: the same conversation caught up from the store, so whatever + // this manager still holds for it — a queued message, the background-job + // tray — belongs to the chat being reloaded rather than to a chat being + // left, and a plain load would drop it. await m.loadPastChat(id, { refresh: true }) } else if (found === 'missing') { // The driver rotated to a chat that holds nothing: it ran "/clear", or its @@ -1202,30 +1181,16 @@ async function applyTurnEnd(sessionId: string, chatId: string, committed: boolea // 'unavailable' leaves everything as it is: the store is unreadable right // now, which says nothing about the conversation. } finally { - // Cleared only once the catch-up is done. Until then this manager still - // pairs a mirrored transcript with the pre-turn history, and the send and - // save paths gate on this flag to stay off that pair. - m.mirroringRemoteRun = false + // Unconditional, including the paths that never reached the store: leaving + // the position at `catchingUp` would lock this tab's composer for a run + // that is already over, with nothing left to arrive and free it. + noteCaughtUp(sessionId) } - // Anything typed here while the other tab held the session was queued rather - // than sent. Send it only after a turn that landed, which is the rule a turn - // follows locally: firing it into a failed turn, or into the gap left by a tab - // that vanished, is how a follow-up ends up answering nothing. - if (!committed) return - if (m.mirroredPlanMode) { - // Plan mode belongs to the tab that entered it, and a turn sent from here - // would run under this tab's own autonomy instead — unblocking the very - // workspace tools the posture exists to hold back. Leave the message where - // the user put it and say why, rather than quietly sending it out of mode. - sendUserToast('This session is planning in another tab. Your message stays queued.') - return - } - await m.flushQueuedMessage() } /** A Stop pressed in a watching tab reaches the run here. */ function applyCancelRequest(sessionId: string): void { - if (!isLocallyDriven(sessionId)) return + if (!isDriving(sessionId)) return // No reason: this IS the user's Stop, just pressed elsewhere, and the // queued-message and rollback paths key off that. runtimes.get(sessionId)?.manager.cancel() @@ -1235,12 +1200,12 @@ function applyCancelRequest(sessionId: string): void { // most once — the driver drops the callback as it resolves it — so two tabs // answering the same prompt is a race the first click simply wins. function applyToolConfirmation(sessionId: string, toolId: string, confirmed: boolean): void { - if (!isLocallyDriven(sessionId)) return + if (!isDriving(sessionId)) return runtimes.get(sessionId)?.manager.handleToolConfirmation(toolId, confirmed) } function applyQuestionAnswer(sessionId: string, toolId: string, choices: string[]): void { - if (!isLocallyDriven(sessionId)) return + if (!isDriving(sessionId)) return runtimes.get(sessionId)?.manager.handleUserQuestionAnswer(toolId, choices) } @@ -1250,7 +1215,7 @@ registerSyncHandlers({ onCancelRequest: applyCancelRequest, onToolConfirmation: applyToolConfirmation, onQuestionAnswer: applyQuestionAnswer, - onTurnEnd: (sessionId, chatId, committed) => void applyTurnEnd(sessionId, chatId, committed), + onTurnEnd: (sessionId, chatId) => void applyTurnEnd(sessionId, chatId), // A session deleted in another tab takes its runtime with it, so an open // chat for it stops streaming and releases its editors. onSessionDelete: (id) => disposeRuntime(id), @@ -1261,6 +1226,11 @@ registerSyncHandlers({ void runtimes.get(sessionId)?.manager.artifacts.applyRemoteArtifact(artifactId) }) +// A driving tab that closed mid-turn sends no turn-end. Its watchers land here +// instead, and take the same path: stop showing the run, re-read whatever the +// store holds, and become able to send again. +onDriverLost((sessionId) => void applyTurnEnd(sessionId, '')) + export function getOrCreateRuntime(session: Session): SessionRuntime { let runtime = runtimes.get(session.id) if (!runtime) { @@ -1277,6 +1247,11 @@ export function disposeRuntime(sessionId: string) { runtime.manager.cancel('runtime disposed') runtime.manager.historyManager.close() runtimes.delete(sessionId) + // The per-run bookkeeping outlives the manager it describes otherwise: a + // frame timer with no runtime to read keeps ticking, and a leftover position + // would answer for a session this tab no longer holds. + stopMirroring(sessionId) + clearRunPosition(sessionId) } export function listRuntimes(): SessionRuntime[] { diff --git a/frontend/src/lib/components/sessions/sessionSync.svelte.ts b/frontend/src/lib/components/sessions/sessionSync.svelte.ts index 582bfa3ecf..057a97b0f6 100644 --- a/frontend/src/lib/components/sessions/sessionSync.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionSync.svelte.ts @@ -1,38 +1,24 @@ import { BROWSER } from 'esm-env' -import { SvelteMap, SvelteSet } from 'svelte/reactivity' import { onUserChange, scopedKey } from '$lib/userScopedStorage' import type { DisplayMessage } from '$lib/components/copilot/chat/shared' +import { noteDriverAlive, noteRemoteTurnEnded } from './sessionRunOwner.svelte' -// Cross-tab coordination for AI sessions. Everything a session is made of — -// the record list, the chat transcript, the run itself — lives in the tab, so -// two tabs on the same session are two independent copies of it. This module -// is the one channel between them: it mirrors record writes, elects a single -// driving tab per run, and streams the driver's live transcript to the others. -// -// Why a run needs an owner at all: two tabs sending on one session append to -// the same chat id, and whichever saveChat lands last silently drops the other's -// turn — after its tool calls have already run against the workspace. +// The channel between tabs open on the same AI session. Everything a session is +// made of — the record list, the chat transcript, the run itself — lives in the +// tab, so two tabs on one session are two independent copies of it. This module +// carries messages between those copies: record writes, and the driving tab's +// live transcript. Who is entitled to drive is sessionRunOwner's question; this +// module only tells it what arrived. // // The channel is per-user (same email scoping as the IndexedDB stores), so a // browser shared by two accounts never crosses them. const CHANNEL_BASE = 'windmill-sessions-sync' -/** Whether this context can hold a real lock on a run. Web Locks is - * secure-context only, so a self-hosted instance served over plain HTTP has - * none — mirroring still runs there, on the weaker footing described at - * {@link withSessionRunLock} and {@link runLockHeld}. */ -const EXCLUSIVE_OWNERSHIP = BROWSER && !!globalThis.navigator?.locks?.query - /** Mirror ticks are throttled to this while a turn streams. Also the heartbeat * interval: an unchanged run still ticks, so silence means the driver is gone. */ export const MIRROR_THROTTLE_MS = 250 -/** A driver with no mirror for this long is presumed dead, pending the lock - * query that confirms it. Generous next to the heartbeat: a busy tab can be - * starved of frames for a while without actually having gone away. */ -const MIRROR_SILENCE_MS = 10_000 - /** Carries only the id on purpose. Broadcast delivery order and IndexedDB * commit order are independent, so shipping a copy of the record lets an older * write land on top of a newer one; re-reading converges on what the shared @@ -44,10 +30,12 @@ type SessionDeleteMsg = { kind: 'session-delete'; id: string } * removal arrives. `sessionId` is here so a receiver can route to the right * store without a database round-trip for artifacts it does not hold. */ type SessionArtifactMsg = { kind: 'session-artifact'; sessionId: string; artifactId: string } -/** `committed` distinguishes a turn that landed from one that errored, was - * rolled back, or belonged to a tab that vanished. Watchers auto-send what the - * user queued only on the first, matching the rule a turn follows locally. */ -type TurnEndMsg = { kind: 'turn-end'; sessionId: string; chatId: string; committed: boolean } +/** Names the chat the driver ended on, which is not necessarily the one it + * started on: a "/clear" rotates it mid-turn, and the watcher's re-read has to + * follow. Whether the turn landed or errored is not carried, because the + * watcher does the same thing either way — re-read the record and stop + * showing the run. */ +type TurnEndMsg = { kind: 'turn-end'; sessionId: string; chatId: string } type MirrorMsg = { kind: 'mirror' sessionId: string @@ -108,7 +96,7 @@ type Handlers = { onSessionPut: (id: string) => void onSessionDelete: (id: string) => void onSessionArtifact: (sessionId: string, artifactId: string) => void - onTurnEnd: (sessionId: string, chatId: string, committed: boolean) => void + onTurnEnd: (sessionId: string, chatId: string) => void onMirror: (msg: MirrorMsg) => void onResyncRequest: (sessionId: string) => void onCancelRequest: (sessionId: string) => void @@ -181,11 +169,11 @@ function receive(msg: SyncMsg): void { emit('onSessionArtifact', msg.sessionId, msg.artifactId) break case 'turn-end': - remoteDriven.delete(msg.sessionId) - emit('onTurnEnd', msg.sessionId, msg.chatId, msg.committed) + noteRemoteTurnEnded(msg.sessionId) + emit('onTurnEnd', msg.sessionId, msg.chatId) break case 'mirror': - noteDriverAlive(msg.sessionId) + noteDriverAlive(msg.sessionId, msg.planModeActive) emit('onMirror', msg) break case 'resync-request': @@ -229,135 +217,12 @@ export function broadcastSessionArtifact(sessionId: string, artifactId: string): post({ kind: 'session-artifact', sessionId, artifactId }) } -// --------------------------------------------------------------------------- -// Ownership -// --------------------------------------------------------------------------- - -/** Sessions currently being driven by another tab, with the time of the last - * sign of life. Reactive so a tab that starts or stops driving re-renders the - * gates that read it. */ -const remoteDriven = new SvelteMap() - -function noteDriverAlive(sessionId: string): void { - remoteDriven.set(sessionId, { lastAt: Date.now() }) - ensureReaper() -} - -// Runs only while some session is being driven elsewhere, and stops itself once -// none is — a browser with a single tab open never arms it at all. -let reaperTimer: ReturnType | undefined - -function ensureReaper(): void { - if (reaperTimer) return - reaperTimer = setInterval(() => { - if (remoteDriven.size === 0) { - clearInterval(reaperTimer) - reaperTimer = undefined - return - } - void reapDeadDrivers() - }, MIRROR_SILENCE_MS) -} - -export function isRemotelyDriven(sessionId: string): boolean { - return remoteDriven.has(sessionId) -} - -/** Sessions this tab is currently driving. Reactive so the gates that read it - * re-render when a run starts or ends. */ -const locallyDriven = new SvelteSet() - -export function isLocallyDriven(sessionId: string): boolean { - return locallyDriven.has(sessionId) -} - -function lockName(sessionId: string): string { - return `wm-session-run:${sessionId}` -} - -/** Run `body` as the session's sole driver, or return 'busy' without running it - * when another tab already holds the run. - * - * Web Locks is what makes this safe across a crash: the lock is held by the tab, - * not by a record someone has to clean up, so a driver that dies mid-turn - * releases it and the next send succeeds. */ -export async function withSessionRunLock( - sessionId: string, - body: () => Promise -): Promise { - if (!EXCLUSIVE_OWNERSHIP) { - // No lock to take, so exclusion is best-effort: refuse only when another - // tab is visibly mid-run. That covers the case that actually happens (one - // tab already going) and leaves a genuine simultaneous start racing, which - // is what a session already did before any of this. Marking the run ours - // is what keeps the two tabs from adopting each other's frames. - if (isRemotelyDriven(sessionId)) return 'busy' - locallyDriven.add(sessionId) - try { - return await body() - } finally { - locallyDriven.delete(sessionId) - } - } - return (await navigator.locks.request( - lockName(sessionId), - { mode: 'exclusive', ifAvailable: true }, - async (lock) => { - // `ifAvailable` hands back a null lock instead of queueing when another - // tab holds it, which is exactly the "refuse, don't stack up turns" - // behavior we want. - if (!lock) return 'busy' as const - locallyDriven.add(sessionId) - try { - return await body() - } finally { - locallyDriven.delete(sessionId) - } - } - )) as T | 'busy' -} - -/** Whether any tab currently holds the run lock for this session. Used to - * settle a driver that stopped mirroring: a released lock proves the tab is - * gone, where silence alone only suggests it. */ -async function runLockHeld(sessionId: string): Promise { - // Nothing to consult without the lock API, so silence is the only evidence — - // and the caller only asks after a driver has gone quiet for the whole - // silence window. Reaping a driver that was merely starved is self-correcting: - // the watcher stops showing a run that is not visibly progressing and re-reads - // the record, which the next frame or the turn-end would have done anyway. - if (!EXCLUSIVE_OWNERSHIP) return false - try { - const state = await navigator.locks.query() - const name = lockName(sessionId) - return !!state.held?.some((l) => l.name === name) - } catch { - return true - } -} - -/** Drop drivers that have gone silent and whose lock is no longer held, so a - * closed tab can't leave a session showing "generating" forever. */ -async function reapDeadDrivers(): Promise { - const now = Date.now() - const stale = [...remoteDriven.entries()] - .filter(([, v]) => now - v.lastAt > MIRROR_SILENCE_MS) - .map(([id]) => id) - for (const id of stale) { - if (!(await runLockHeld(id))) { - remoteDriven.delete(id) - // A driver that disappeared mid-turn committed nothing. - emit('onTurnEnd', id, '', false) - } - } -} - // --------------------------------------------------------------------------- // Live mirroring // --------------------------------------------------------------------------- -export function broadcastTurnEnd(sessionId: string, chatId: string, committed: boolean): void { - post({ kind: 'turn-end', sessionId, chatId, committed }) +export function broadcastTurnEnd(sessionId: string, chatId: string): void { + post({ kind: 'turn-end', sessionId, chatId }) } export function requestResync(sessionId: string): void {