diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index 5e56500cff..2c688c7d6a 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -102,11 +102,7 @@ import type AIChatInput from './AIChatInput.svelte' import { prepareApiSystemMessage, prepareApiUserMessage } from './api/core' import { closeInterruptedToolBatch, runChatLoop, truncateToToolPairedPrefix } from './chatLoop' import { sanitizeToolCallArguments } from './toolCallArguments' -import { - billedTokens, - normalizeContextUsage, - type ChatTokenUsage -} from './tokenUsage' +import { billedTokens, normalizeContextUsage, type ChatTokenUsage } from './tokenUsage' import { logAiUsage } from '$lib/utils/aiUsageReporter' import type { ReviewChangesOpts } from './monaco-adapter' import { @@ -683,6 +679,30 @@ export class AIChatManager { // chat's items. Set here (not imported) to avoid a copilot→sessions cycle. onChatRotated: ((chatId: string) => void) | undefined = undefined + // Wraps a whole turn so an external coordinator can bracket it or refuse it + // outright by returning 'busy' without running the body. Session runtimes + // wire this to the cross-tab run lock: two tabs sending on one session append + // to the same chat id, and the loser's turn is dropped by the next saveChat + // after its tool calls have already hit the workspace. Undefined everywhere + // else, where a turn has no rival. Set here (not imported) to avoid a + // copilot→sessions cycle. + runGuard: ((body: () => Promise) => Promise) | undefined = undefined + + // Routes a Stop to whatever is actually running the turn. A tab mirroring + // another tab's run has no abortController to abort, so its Stop button would + // otherwise do nothing at all. Returns true once the request is on its way, + // and the local cancel is skipped. Set here (not imported) to avoid a + // copilot→sessions cycle. + remoteCancel: (() => boolean) | undefined = undefined + + // Route a blocked run's answer to the tab holding the resolver. A run parked + // on a tool confirmation or a question is exactly when the user is most likely + // to be looking at another tab, and the promise waiting on the answer lives + // only in the driver. Each returns true once the answer is on its way. Set + // here (not imported) to avoid a copilot→sessions cycle. + remoteToolConfirmation: ((toolId: string, confirmed: boolean) => boolean) | undefined = undefined + remoteQuestionAnswer: ((toolId: string, choices: string[]) => boolean) | undefined = undefined + // Workspace items the CURRENT chat modified via AI tool calls, as // `${UserDraftItemKind}:${storagePath}` keys (see modifiedItemsMask.ts). // undefined = untracked: only the global side-panel chat (never initialised), @@ -1595,7 +1615,12 @@ export class AIChatManager { if (confirmationCallback) { confirmationCallback.resolve(confirmed) this.confirmationCallbacks.delete(toolId) + return } + // No local resolver. Either another tab is running this turn and holds it, + // or this is a card restored from history whose resolver died with the old + // page — the hook answers only in the first case. + this.remoteToolConfirmation?.(toolId, confirmed) } private acceptPendingToolConfirmations = () => { @@ -1691,7 +1716,11 @@ export class AIChatManager { handleUserQuestionAnswer = (toolId: string, choices: string[]): boolean => { const callback = this.userQuestionCallbacks.get(toolId) if (!callback) { - return false + // Another tab is running this turn and holds the resolver: hand the + // answer over, and report it delivered so the composer clears as it + // would locally. A card restored from history reaches the same branch + // with nobody driving, and still reports undelivered. + return this.remoteQuestionAnswer?.(toolId, choices) ?? false } // Display-only readback for the collapsed tool-header: a compact comma list. @@ -1852,6 +1881,27 @@ export class AIChatManager { void this.sendRequest({ instructions: text }) } + /** Send the queued message, if there is one, as its own turn. Also the path a + * tab takes when the turn it was watching ends in another tab: the queue was + * filled here while that run held the session, and it is owed the same + * auto-send it would have got had this tab been the one running. */ + async flushQueuedMessage(): Promise { + if (!this.#hasQueuedMessage()) return + const next = this.#takeQueue() + const accepted = await this.sendRequest({ + instructions: next.draft.text, + images: next.draft.images, + files: next.draft.files, + contextOverride: next.context, + queued: true + }) + if (accepted === false) { + // The auto-send bailed before becoming a turn (e.g. beforeSend failed, or + // another tab took the session first); keep it queued rather than lose it. + this.#restoreQueue(next) + } + } + /** Remove the queued message and put it back into the input, images included. */ dequeueMessage() { if (!this.#hasQueuedMessage()) { @@ -2778,7 +2828,18 @@ export class AIChatManager { } this.#sendsInFlight++ try { - return await this.sendRequestImpl(options) + // Depth 1 only. A turn recursively flushes queued messages through this + // same method, and the guard's lock is not reentrant — running it on the + // nested send would refuse the queued message as if a rival tab held it. + const guard = this.#sendsInFlight === 1 ? this.runGuard : undefined + if (!guard) return await this.sendRequestImpl(options) + const outcome = await guard(() => this.sendRequestImpl(options)) + // Refused: the body never ran, so the composer still holds the message + // and the user can retry here or carry on in the driving tab. Reported + // as a plain failed send so an auto-sent queued message is put back on + // the queue rather than dropped. + if (outcome === 'busy') return false + return outcome } finally { this.#sendsInFlight-- } @@ -3720,20 +3781,8 @@ export class AIChatManager { // empty-response rollback, or a programmatic cancel (panel teardown, // save-and-clear) leaves it in place as a card so it isn't fired into a // failed or torn-down turn. - if ((turnCommittedCleanly || this.wasCancelledByUser()) && this.#hasQueuedMessage()) { - const next = this.#takeQueue() - const accepted = await this.sendRequest({ - instructions: next.draft.text, - images: next.draft.images, - files: next.draft.files, - contextOverride: next.context, - queued: true - }) - if (accepted === false) { - // The auto-send bailed before becoming a turn (e.g. beforeSend - // failed); keep it as the queued message instead of losing it. - this.#restoreQueue(next) - } + if (turnCommittedCleanly || this.wasCancelledByUser()) { + await this.flushQueuedMessage() } // A background job may have finished mid-turn: its note missed this turn's // preamble (captured at the start) and the poller skipped auto-resume while @@ -3752,6 +3801,11 @@ export class AIChatManager { } cancel = (reason?: string) => { + // Only the user's own Stop travels: it alone arrives without a reason. + // Every internal cancel names one (teardown, save-and-clear, a disposed + // runtime), and those must stay local — a passive tab closing a session + // would otherwise kill a turn still running in the tab that owns it. + if (reason === undefined && this.remoteCancel?.()) return for (const { resolve } of this.confirmationCallbacks.values()) { resolve(false) } diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index f788844135..5af8afdc79 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts @@ -224,6 +224,85 @@ describe('AIChatManager unmounted-chat guard', () => { }) }) +describe('AIChatManager cross-tab run guard', () => { + // Session runtimes wire `runGuard` to a lock held by the one tab driving the + // session. Refusal has to read as a failed send: a queued message auto-sends + // through this same path and is put back on the queue only on `false`. + it('reports a refused turn as a failed send and leaves the queue intact', async () => { + const manager = new AIChatManager() + manager.isSessionChat = true + manager.runGuard = async () => 'busy' + manager.queueMessage('follow up') + + await manager.flushQueuedMessage() + + expect(mocks.runChatLoop).not.toHaveBeenCalled() + expect(manager.queuedMessage).toBe('follow up') + }) + + // A turn flushes its queued message by re-entering sendRequest, and the lock + // behind the guard is not reentrant: applying it to the nested send would + // refuse the queued message as though a rival tab held the session. + it('applies the guard to the outermost send only', async () => { + const manager = new AIChatManager() + manager.isSessionChat = true + const runGuard = vi.fn(async (body: () => Promise) => body()) + manager.runGuard = runGuard as typeof manager.runGuard + mocks.runChatLoop.mockImplementation(async (config: any) => { + const message = { role: 'assistant' as const, content: 'done' } + config.addedMessages?.push(message) + return { + addedMessages: [message], + tokenUsage: { prompt: 0, completion: 0, total: 0 }, + hitMaxIterations: false + } + }) + manager.queueMessage('follow up') + + await manager.sendRequest({ instructions: 'first' }) + + expect(mocks.runChatLoop).toHaveBeenCalledTimes(2) + expect(runGuard).toHaveBeenCalledTimes(1) + }) + + // A run parked on the user is answered from whichever tab the user is in, but + // the resolver exists only in the tab running the loop. Routing has to be the + // fallback, never the first move: the driving tab holds its own resolvers and + // must answer them directly. + it('routes a confirmation onward only when no local resolver holds it', async () => { + const manager = new AIChatManager() + const remoteToolConfirmation = vi.fn(() => true) + manager.remoteToolConfirmation = remoteToolConfirmation + + const local = manager.requestConfirmation('tool-local', 'delete_workspace_item') + manager.handleToolConfirmation('tool-local', true) + await expect(local).resolves.toBe(true) + expect(remoteToolConfirmation).not.toHaveBeenCalled() + + manager.handleToolConfirmation('tool-elsewhere', false) + expect(remoteToolConfirmation).toHaveBeenCalledWith('tool-elsewhere', false) + }) + + // Only the user's own Stop travels to the driving tab. Every internal cancel + // names a reason, and routing one would let a passive tab's teardown kill a + // turn still running in the tab that owns it. + it('routes only an unattributed cancel to the driving tab', () => { + const manager = new AIChatManager() + const remoteCancel = vi.fn(() => true) + manager.remoteCancel = remoteCancel + const abort = vi.fn() + ;(manager as any).abortController = { abort, signal: { aborted: false } } + + manager.cancel() + expect(remoteCancel).toHaveBeenCalledTimes(1) + expect(abort).not.toHaveBeenCalled() + + manager.cancel('runtime disposed') + expect(remoteCancel).toHaveBeenCalledTimes(1) + expect(abort).toHaveBeenCalledWith('runtime disposed') + }) +}) + describe('AIChatManager.sendOrQueue', () => { // The programmatic senders (an editor's "AI Fix", an arriving hand-off) have no // composer to enforce the composer's rule for them: a second loop on one manager diff --git a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts index 24d7284b95..ef310230d7 100644 --- a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts @@ -599,6 +599,23 @@ export default class HistoryManager { }).catch((err) => console.error('Could not delete chat', err)) } + /** Re-read one chat from the store into the in-memory mirror, for a record + * another tab wrote after this manager last read it. `init()` is the wrong + * tool: it re-reads the user's entire history to pick up a single chat. */ + async reloadChat(id: string): Promise { + const db = await this.dbh.whenReady() + if (!db) return false + try { + const chat = await db.get('chats', id) + if (!chat) return false + this.savedChats = { ...this.savedChats, [id]: chat } + return true + } catch (err) { + console.error('Could not reload chat', err) + return false + } + } + async loadPastChat(id: string) { const chat = this.savedChats[id] if (!chat) return diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index fc0d291190..ee8a4fde26 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -90,8 +90,25 @@ import type { RawAppDomResult } from '$lib/components/raw_apps/rawAppDom' import { getNonStreamingMetadataCompletion } from '$lib/components/copilot/lib' +import { sendUserToast } from '$lib/toast' import { pendingUserAction, type DisplayMessage } from '$lib/components/copilot/chat/shared' import type { ChatCompletionMessageParam } from 'openai/resources/index.mjs' +import { + broadcastMirror, + broadcastTurnEnd, + broadcastTurnStart, + isLocallyDriven, + isRemotelyDriven, + MIRROR_THROTTLE_MS, + registerSyncHandlers, + requestCancel, + requestResync, + sendQuestionAnswer, + sendToolConfirmation, + withSessionRunLock, + type MirrorMsg, + type MirrorSnapshot +} from './sessionSync.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 @@ -914,6 +931,49 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { // preselect the wrong chat's items. manager.onChatRotated = (chatId) => setSessionChatId(session.id, chatId) + // One tab drives a session at a time. The lock brackets the turn so the other + // tabs can mirror it, and refuses a second driver rather than letting two + // turns interleave into one chat id. + manager.runGuard = async (body) => { + const outcome = await withSessionRunLock(session.id, async () => { + broadcastTurnStart(session.id, manager.historyManager.getCurrentChatId()) + startMirroring(session.id) + try { + return await body() + } finally { + 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()) + } + }) + if (outcome === 'busy') { + sendUserToast('This session is running in another tab. Your message was kept.', true) + } + return outcome + } + + // 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 (!isRemotelyDriven(session.id)) return false + requestCancel(session.id) + return true + } + + // 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 (!isRemotelyDriven(session.id)) return false + sendToolConfirmation(session.id, toolId, confirmed) + return true + } + manager.remoteQuestionAnswer = (toolId, choices) => { + if (!isRemotelyDriven(session.id)) return false + sendQuestionAnswer(session.id, toolId, choices) + return true + } + if (session.chatId) { manager.historyManager.setCurrentChatId(session.chatId) await manager.historyManager.tagChatWithSession(session.chatId, session.id) @@ -932,6 +992,154 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { } } +// --------------------------------------------------------------------------- +// Cross-tab mirroring +// --------------------------------------------------------------------------- + +// Driving tabs post a frame on this cadence for as long as their turn runs. +// A plain interval rather than an $effect on the manager's fields: it also +// serves as the liveness heartbeat (a run that stalls in a long tool call +// still ticks), and it picks up in-place edits to already-rendered tool cards +// that reactive tracking of the array root would miss. +const mirrorTimers = new Map>() + +function mirrorSnapshotOf(sessionId: string): MirrorSnapshot | undefined { + const runtime = runtimes.get(sessionId) + if (!runtime) return undefined + const m = runtime.manager + return { + sessionId, + chatId: m.historyManager.getCurrentChatId(), + displayMessages: $state.snapshot(m.displayMessages) as DisplayMessage[], + loading: m.loading, + currentReply: m.currentReply, + currentReasoning: m.currentReasoning, + currentReasoningActive: m.currentReasoningActive, + loadingLabel: m.loadingLabel, + compacting: m.compacting + } +} + +function postMirror(sessionId: string, opts?: { full?: boolean }): void { + const snap = mirrorSnapshotOf(sessionId) + if (snap) broadcastMirror(snap, opts) +} + +function startMirroring(sessionId: string): void { + stopMirroring(sessionId) + postMirror(sessionId) + mirrorTimers.set( + sessionId, + setInterval(() => postMirror(sessionId), MIRROR_THROTTLE_MS) + ) +} + +function stopMirroring(sessionId: string): void { + const timer = mirrorTimers.get(sessionId) + if (!timer) return + clearInterval(timer) + mirrorTimers.delete(sessionId) + // One last frame: the closing tokens of a turn usually land between ticks, + // and this is what the passive tabs render until their re-read completes. + postMirror(sessionId) +} + +/** Adopt a frame from the tab driving this session. */ +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 + const runtime = runtimes.get(msg.sessionId) + if (!runtime) return + const m = runtime.manager + const onSameChat = m.historyManager.getCurrentChatId() === msg.chatId + if (msg.baseIndex > 0 && !(onSameChat && m.displayMessages.length >= msg.baseIndex)) { + // Nothing here can host this tail: this tab joined mid-run, or the driver + // rotated to a chat it isn't on. Ask for the whole transcript instead of + // splicing onto a prefix that isn't the same conversation. + requestResync(msg.sessionId) + return + } + if (!onSameChat) { + // A full snapshot names the conversation it came from, so follow the + // driver onto it rather than rendering it under the wrong chat id. + m.historyManager.setCurrentChatId(msg.chatId) + setSessionChatId(msg.sessionId, msg.chatId) + } + m.displayMessages = + msg.baseIndex === 0 ? msg.tail : [...m.displayMessages.slice(0, msg.baseIndex), ...msg.tail] + m.loading = msg.loading + m.currentReply = msg.currentReply + m.currentReasoning = msg.currentReasoning + m.currentReasoningActive = msg.currentReasoningActive + m.loadingLabel = msg.loadingLabel + m.compacting = msg.compacting +} + +/** The driver answers a resync with its whole transcript. */ +function answerResync(sessionId: string): void { + if (!isLocallyDriven(sessionId)) return + postMirror(sessionId, { full: true }) +} + +/** A run finished in another tab. The mirror carried the rendered transcript + * 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): Promise { + if (isLocallyDriven(sessionId)) return + const runtime = runtimes.get(sessionId) + if (!runtime) return + const m = runtime.manager + // Cleared before the re-read, not after: loadPastChat refuses to run while + // the manager looks busy, and `loading` here is the mirrored driver's. + m.loading = false + m.currentReply = '' + m.currentReasoning = '' + m.currentReasoningActive = false + m.loadingLabel = undefined + m.compacting = false + const id = chatId || m.historyManager.getCurrentChatId() + if (id && (await m.historyManager.reloadChat(id))) await m.loadPastChat(id) + // Anything typed here while the other tab held the session was queued rather + // than sent. The session is free now, so it goes out — the same auto-send it + // would have had if this tab had been the one running the turn. + await m.flushQueuedMessage() +} + +/** A Stop pressed in a watching tab reaches the run here. */ +function applyCancelRequest(sessionId: string): void { + if (!isLocallyDriven(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() +} + +// An answer from a watching tab reaches the parked loop here. Both resolve at +// 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 + runtimes.get(sessionId)?.manager.handleToolConfirmation(toolId, confirmed) +} + +function applyQuestionAnswer(sessionId: string, toolId: string, choices: string[]): void { + if (!isLocallyDriven(sessionId)) return + runtimes.get(sessionId)?.manager.handleUserQuestionAnswer(toolId, choices) +} + +registerSyncHandlers({ + onMirror: applyMirror, + onResyncRequest: answerResync, + onCancelRequest: applyCancelRequest, + onToolConfirmation: applyToolConfirmation, + onQuestionAnswer: applyQuestionAnswer, + 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) +}) + export function getOrCreateRuntime(session: Session): SessionRuntime { let runtime = runtimes.get(session.id) if (!runtime) { diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index d8ef56b273..965fd5d072 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -25,6 +25,11 @@ import { type DBSchema, type IDBPDatabase } from 'idb' import { userScopedDb } from '$lib/userScopedDb' import { deleteItemsForSession } from '../copilot/chat/files/attachedFilesDB' import { deleteArtifactsForSession } from '../copilot/chat/artifacts/artifactsDB' +import { + broadcastSessionDelete, + broadcastSessionPut, + registerSyncHandlers +} from './sessionSync.svelte' // Switch the global workspace iff the target differs from the active one // and is non-empty. Centralises the "session needs its workspace in focus" @@ -471,7 +476,11 @@ export async function putSession(s: Session): Promise { const db = await sessionsDb.whenReady() if (!db) return try { - await putSessionRow(db, $state.snapshot(s)) + const row = $state.snapshot(s) + await putSessionRow(db, row) + // Only after the row lands: the other tabs are being told what IndexedDB + // now holds, so a failed write must not announce itself. + broadcastSessionPut(row) } catch (e) { console.error('Failed to persist session', e) } @@ -483,11 +492,46 @@ export async function deleteSessionRecord(id: string): Promise { if (!db) return try { await deleteSessionRow(db, id) + broadcastSessionDelete(id) } catch (e) { console.error('Failed to delete session record', e) } } +// Apply a record another tab just wrote. In-memory only: the row is already in +// the shared IndexedDB, so re-persisting it here would echo back out through +// putSession. The incoming copy wins wholesale rather than field-wise — it is +// what the store now holds, and a merge would invent a third version that +// matches neither tab. +function applyRemoteSessionPut(session: Session): void { + if (deletedSessionIds.has(session.id)) return + const i = sessionState.sessions.findIndex((s) => s.id === session.id) + if (i >= 0) { + sessionState.sessions[i] = session + return + } + // New elsewhere: slot it in by createdAt, after any local transient drafts, + // reproducing the newest-first order hydrateSessions maintains. + const at = sessionState.sessions.findIndex((s) => !s.transient && s.createdAt < session.createdAt) + if (at < 0) sessionState.sessions.push(session) + else sessionState.sessions.splice(at, 0, session) +} + +// Mirror of the local delete path: tombstone so this tab's own pending writes +// (unread watermark, preview-tab flush) can't resurrect a record another tab +// removed, then drop it from the list. +function applyRemoteSessionDelete(id: string): void { + deletedSessionIds.add(id) + const i = sessionState.sessions.findIndex((s) => s.id === id) + if (i >= 0) sessionState.sessions.splice(i, 1) + if (sessionState.currentSessionId === id) sessionState.currentSessionId = undefined +} + +registerSyncHandlers({ + onSessionPut: applyRemoteSessionPut, + onSessionDelete: applyRemoteSessionDelete +}) + // Load the in-memory list from the user's DB. getAll() returns records in key // (id) order; sort by createdAt descending to reproduce the newest-first order // createSession() maintains (it prepends). whenReady() reopens for the current diff --git a/frontend/src/lib/components/sessions/sessionSync.svelte.ts b/frontend/src/lib/components/sessions/sessionSync.svelte.ts new file mode 100644 index 0000000000..bcf3ddf269 --- /dev/null +++ b/frontend/src/lib/components/sessions/sessionSync.svelte.ts @@ -0,0 +1,375 @@ +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 type { Session } from './sessionState.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. +// +// 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' + +/** Tail length of the transcript sent on each mirror tick. In-place edits to + * already-rendered messages (tool cards settling) land within a few messages + * of the end, so a short tail carries them while keeping the payload bounded + * on long conversations. Anything older is corrected by the IndexedDB re-read + * at turn end. */ +const MIRROR_TAIL = 10 + +/** 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 + +type SessionPutMsg = { kind: 'session-put'; session: Session } +type SessionDeleteMsg = { kind: 'session-delete'; id: string } +type TurnStartMsg = { kind: 'turn-start'; sessionId: string; chatId: string } +type TurnEndMsg = { kind: 'turn-end'; sessionId: string; chatId: string } +type MirrorMsg = { + kind: 'mirror' + sessionId: string + chatId: string + /** Index the tail starts at; the receiver keeps its own prefix below it, and + * asks for a full snapshot when it has no prefix that reaches this far. */ + baseIndex: number + tail: DisplayMessage[] + loading: boolean + currentReply: string + currentReasoning: string + currentReasoningActive: boolean + loadingLabel: string | undefined + compacting: boolean +} +/** Sent by a tab whose local prefix can't host the tail it just received (it + * joined mid-run, or is on a different chat). The driver answers with a full + * snapshot. */ +type ResyncRequestMsg = { kind: 'resync-request'; sessionId: string } +/** A Stop pressed in a tab that is only watching the run. */ +type CancelRequestMsg = { kind: 'cancel-request'; sessionId: string } +/** A run blocked on the user is unblocked from whichever tab the user is in; + * the resolver waiting on the answer only exists in the driving tab. */ +type ToolConfirmationMsg = { + kind: 'tool-confirmation' + sessionId: string + toolId: string + confirmed: boolean +} +type QuestionAnswerMsg = { + kind: 'question-answer' + sessionId: string + toolId: string + choices: string[] +} + +type SyncMsg = + | SessionPutMsg + | SessionDeleteMsg + | TurnStartMsg + | TurnEndMsg + | MirrorMsg + | ResyncRequestMsg + | CancelRequestMsg + | ToolConfirmationMsg + | QuestionAnswerMsg + +type Handlers = { + onSessionPut: (session: Session) => void + onSessionDelete: (id: string) => void + onTurnStart: (sessionId: string, chatId: string) => void + onTurnEnd: (sessionId: string, chatId: string) => void + onMirror: (msg: MirrorMsg) => void + onResyncRequest: (sessionId: string) => void + onCancelRequest: (sessionId: string) => void + onToolConfirmation: (sessionId: string, toolId: string, confirmed: boolean) => void + onQuestionAnswer: (sessionId: string, toolId: string, choices: string[]) => void +} + +const subscribers: Partial[] = [] + +/** Registered by sessionState (records) and sessionRuntime (runs) at module + * load. Split so this module stays free of both their imports — it would + * otherwise sit in an import cycle with each. Several subscribers can claim + * the same event: a remote delete has to reach the record list AND tear down + * the runtime, and neither module can call the other. */ +export function registerSyncHandlers(h: Partial): void { + subscribers.push(h) +} + +function emit(event: K, ...args: Parameters): void { + for (const sub of subscribers) { + const fn = sub[event] as ((...a: Parameters) => void) | undefined + if (!fn) continue + try { + fn(...args) + } catch (e) { + // One bad subscriber must not strand the others mid-broadcast. + console.error(`sessionSync: ${event} handler failed`, e) + } + } +} + +let channel: BroadcastChannel | undefined +let channelName: string | undefined + +function openChannel(): void { + const name = scopedKey(CHANNEL_BASE) + if (name === channelName) return + channel?.close() + channel = undefined + channelName = name + if (!name) return + try { + const ch = new BroadcastChannel(name) + ch.onmessage = (ev: MessageEvent) => receive(ev.data) + channel = ch + } catch (e) { + // No BroadcastChannel (or blocked): every tab simply stays independent, + // which is the pre-sync behavior rather than a broken one. + console.error('sessionSync: could not open channel', e) + } +} + +if (BROWSER) { + onUserChange(() => { + // A user switch rescopes the channel name, so the previous identity's + // channel is closed before the next one opens. + openChannel() + }) +} + +function receive(msg: SyncMsg): void { + switch (msg.kind) { + case 'session-put': + emit('onSessionPut', msg.session) + break + case 'session-delete': + emit('onSessionDelete', msg.id) + break + case 'turn-start': + noteDriverAlive(msg.sessionId) + emit('onTurnStart', msg.sessionId, msg.chatId) + break + case 'turn-end': + remoteDriven.delete(msg.sessionId) + emit('onTurnEnd', msg.sessionId, msg.chatId) + break + case 'mirror': + noteDriverAlive(msg.sessionId) + emit('onMirror', msg) + break + case 'resync-request': + emit('onResyncRequest', msg.sessionId) + break + case 'cancel-request': + emit('onCancelRequest', msg.sessionId) + break + case 'tool-confirmation': + emit('onToolConfirmation', msg.sessionId, msg.toolId, msg.confirmed) + break + case 'question-answer': + emit('onQuestionAnswer', msg.sessionId, msg.toolId, msg.choices) + break + } +} + +function post(msg: SyncMsg): void { + if (!channel) return + try { + channel.postMessage(msg) + } catch (e) { + // A non-cloneable payload must never take the turn down with it. + console.error('sessionSync: could not post message', e) + } +} + +// --------------------------------------------------------------------------- +// Record sync +// --------------------------------------------------------------------------- + +export function broadcastSessionPut(session: Session): void { + post({ kind: 'session-put', session }) +} + +export function broadcastSessionDelete(id: string): void { + post({ kind: 'session-delete', id }) +} + +// --------------------------------------------------------------------------- +// Ownership +// --------------------------------------------------------------------------- + +/** Sessions currently being driven by another tab, with the time of the last + * sign of life. Reactive: the picker's status dot and the composer's disabled + * state read it directly. */ +export 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 per-runtime mirror + * effect starts and stops with the run it feeds. */ +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. + * + * The Web Locks API 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. Without a lock, two tabs + * sending on one session append to the same chat id and the later + * `saveChat` silently discards the other's turn — after its tool calls have + * already run against the workspace. */ +export async function withSessionRunLock( + sessionId: string, + body: () => Promise +): Promise { + if (!BROWSER || !navigator.locks) return body() + 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 { + if (!BROWSER || !navigator.locks?.query) return true + 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. Callers + * schedule this; it is cheap and a no-op while every driver is ticking. */ +export 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) + emit('onTurnEnd', id, '') + } + } +} + +// --------------------------------------------------------------------------- +// Live mirroring +// --------------------------------------------------------------------------- + +export function broadcastTurnStart(sessionId: string, chatId: string): void { + post({ kind: 'turn-start', sessionId, chatId }) +} + +export function broadcastTurnEnd(sessionId: string, chatId: string): void { + post({ kind: 'turn-end', sessionId, chatId }) +} + +export function requestResync(sessionId: string): void { + post({ kind: 'resync-request', sessionId }) +} + +export function requestCancel(sessionId: string): void { + post({ kind: 'cancel-request', sessionId }) +} + +export function sendToolConfirmation(sessionId: string, toolId: string, confirmed: boolean): void { + post({ kind: 'tool-confirmation', sessionId, toolId, confirmed }) +} + +export function sendQuestionAnswer(sessionId: string, toolId: string, choices: string[]): void { + post({ kind: 'question-answer', sessionId, toolId, choices }) +} + +export type MirrorSnapshot = { + sessionId: string + chatId: string + displayMessages: DisplayMessage[] + loading: boolean + currentReply: string + currentReasoning: string + currentReasoningActive: boolean + loadingLabel: string | undefined + compacting: boolean +} + +/** Send the driver's current view of a run. `full` forces the whole transcript + * (answering a resync request); otherwise only the tail travels. */ +export function broadcastMirror(snap: MirrorSnapshot, { full = false } = {}): void { + const total = snap.displayMessages.length + const baseIndex = full ? 0 : Math.max(0, total - MIRROR_TAIL) + post({ + kind: 'mirror', + sessionId: snap.sessionId, + chatId: snap.chatId, + baseIndex, + tail: snap.displayMessages.slice(baseIndex), + loading: snap.loading, + currentReply: snap.currentReply, + currentReasoning: snap.currentReasoning, + currentReasoningActive: snap.currentReasoningActive, + loadingLabel: snap.loadingLabel, + compacting: snap.compacting + }) +} + +export type { MirrorMsg }