diff --git a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte index a39d003998..f21de0a209 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte +++ b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte @@ -716,8 +716,18 @@ return } if (aiChatManager.loading) { - // Queue the message instead of silently discarding it — it is - // auto-sent when the streaming turn completes successfully. + // The composer is locked while another tab drives, with one exception: + // a run parked on a question stays answerable from here. An answer + // carrying an attachment misses the branch above (it only routes plain + // choices) and would land on the queue — which in this tab nothing + // drains, because the turn that would flush it belongs elsewhere. Keep + // the draft where the user put it and say what will send. + if (aiChatManager.mirroringRemoteRun) { + sendUserToast('This session is running in another tab. Answer with text to send it there.') + return + } + // Queue the message instead of silently discarding it — it is auto-sent + // when this tab's own streaming turn completes successfully. // Editing-while-loading keeps the old discard behavior. Paste // tokens are expanded into the queued text (the queue is plain // strings), so the full content survives the auto-send. A GLOBAL diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index b870cf9b79..2d06dd9ef7 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -720,12 +720,6 @@ export class AIChatManager { 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. */ - lastTurnAcceptsFollowUp = false - // 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), @@ -1921,6 +1915,11 @@ export class AIChatManager { private restoreRefusedSend( options: NonNullable[0]> ): void { + // Every release site lives inside sendRequestImpl, and a refusal never gets + // there. A resend stages its files' bytes before the call, so without this + // a Retry that is refused charges them against the conversation's budget + // for the lifetime of the manager. + this.#releaseOutgoingReservation(options.resendReservationKey) if (options.queued) return const restored = this.aiChatInput?.restoreInstructions( options.instructions ?? '', @@ -1935,10 +1934,9 @@ export class AIChatManager { } } - /** 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. */ + /** Send the queued message, if there is one, as its own turn. The queue only + * ever fills while this tab is the one running, so this is the only thing + * that drains it. */ async flushQueuedMessage(): Promise { if (!this.#hasQueuedMessage()) return const next = this.#takeQueue() @@ -2968,9 +2966,6 @@ export class AIChatManager { // send exits before install. Kept in a mutable local so every exit path // releases the right key. let reservationKey = options.resendReservationKey - // Cleared up front so an exit before the verdict below (a refused mode, a - // pre-flight throw) reports this turn rather than the previous one's. - this.lastTurnAcceptsFollowUp = false const requestedMode = options.mode ?? this.mode if (!isAIModeVisible(requestedMode)) { this.#releaseOutgoingReservation(reservationKey) @@ -3861,8 +3856,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. - this.lastTurnAcceptsFollowUp = turnCommittedCleanly || this.wasCancelledByUser() - if (this.lastTurnAcceptsFollowUp) { + const acceptsFollowUp = turnCommittedCleanly || this.wasCancelledByUser() + if (acceptsFollowUp) { await this.flushQueuedMessage() } // A background job may have finished mid-turn: its note missed this turn's diff --git a/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts index 04ca8e77ea..f9cd2163df 100644 --- a/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts @@ -101,6 +101,14 @@ export function noteCaughtUp(sessionId: string): void { * down tab can't leave a position behind for a session nothing is watching. */ export function clearRunPosition(sessionId: string): void { positions.delete(sessionId) + probeAnswers.delete(sessionId) +} + +/** True from the driver's turn-end until this tab has re-read what it left + * behind. The catch-up retry reads it to tell "still owed a read" from "a new + * turn started" and from "this session is gone". */ +export function isCatchingUp(sessionId: string): boolean { + return runPosition(sessionId).state === 'catchingUp' } function lockName(sessionId: string): string { @@ -147,12 +155,43 @@ export async function withSessionRunLock( } } -/** 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. */ +/** How long a probe waits for the driver to answer before its silence counts. + * Long enough to survive a busy main thread, short enough that the pre-flight + * is not felt; only origins without Web Locks ever pay it. */ +const PROBE_GRACE_MS = 750 + +let sendDriverProbe: ((sessionId: string) => void) | undefined + +/** Registered by sessionSync, which owns the channel this is asked over. */ +export function setDriverProbe(fn: (sessionId: string) => void): void { + sendDriverProbe = fn +} + +const probeAnswers = new Set() + +/** A driver answered a probe. Recorded rather than folded into the position: + * the answer proves a run is alive, but carries none of the frame state a + * `watching` position is made of. */ +export function noteDriverAnswered(sessionId: string): void { + probeAnswers.add(sessionId) +} + +/** Exclusion without a lock. Silence is not enough to conclude a driver is + * gone — a hidden tab's heartbeat is throttled to as little as once a minute + * while its turn runs on perfectly well, and the reaper will have retired it + * long before that. So ask, and treat only an unanswered probe as absence. + * Getting this wrong runs a second turn against the same chat id, which is the + * duplicate tool calls and lost transcript this whole module exists to stop. */ async function bestEffort(sessionId: string, body: () => Promise): Promise { if (isWatching(sessionId)) return 'busy' + if (sendDriverProbe) { + probeAnswers.delete(sessionId) + sendDriverProbe(sessionId) + await new Promise((resolve) => setTimeout(resolve, PROBE_GRACE_MS)) + if (probeAnswers.delete(sessionId)) return 'busy' + // A frame may also have arrived while the probe was outstanding. + if (isWatching(sessionId)) return 'busy' + } return await drive(sessionId, body) } @@ -171,11 +210,11 @@ async function drive(sessionId: string, body: () => Promise): Promise { * 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. + // Nothing to consult without the lock API, so a silent driver is reaped on + // silence alone. That only ever frees the UI: reaching `idle` does not by + // itself entitle this tab to drive, because {@link bestEffort} probes for a + // live driver at the moment it matters rather than trusting a conclusion + // drawn from silence up to ten seconds earlier. if (!EXCLUSIVE_OWNERSHIP) return false try { const state = await navigator.locks.query() diff --git a/frontend/src/lib/components/sessions/sessionRunOwner.test.ts b/frontend/src/lib/components/sessions/sessionRunOwner.test.ts new file mode 100644 index 0000000000..591e882131 --- /dev/null +++ b/frontend/src/lib/components/sessions/sessionRunOwner.test.ts @@ -0,0 +1,44 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + clearRunPosition, + noteDriverAnswered, + setDriverProbe, + withSessionRunLock +} from './sessionRunOwner.svelte' + +const SESSIONS = ['session-throttled-driver', 'session-driver-gone'] + +// `positions` is module state and a watching entry keeps the reaper interval +// armed, so leave nothing behind for the next suite. +afterEach(() => SESSIONS.forEach(clearRunPosition)) + +// Under the test env there is no Web Locks API, which is the same footing a +// self-hosted instance served over plain HTTP runs on — so these exercise the +// path that has nothing but the channel to arbitrate with. +// +// A driving tab that is hidden has its timers throttled to as little as once a +// minute, so its heartbeat stops long before its turn does and the reaper +// retires it. Concluding from that silence that the run is over starts a second +// turn against the same chat id: duplicate tool calls against the workspace, +// and whichever save lands last discards the other's transcript. +describe('withSessionRunLock with no lock to take', () => { + it('refuses the run when a driver answers the probe', async () => { + setDriverProbe((sessionId) => noteDriverAnswered(sessionId)) + const body = vi.fn(async () => 'ran') + + const outcome = await withSessionRunLock('session-throttled-driver', body) + + expect(outcome).toBe('busy') + expect(body).not.toHaveBeenCalled() + }) + + it('takes the run when nothing answers', async () => { + setDriverProbe(() => {}) + const body = vi.fn(async () => 'ran') + + const outcome = await withSessionRunLock('session-driver-gone', body) + + expect(outcome).toBe('ran') + expect(body).toHaveBeenCalledTimes(1) + }) +}) diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index cb0c9c1b27..f42c93b923 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -108,6 +108,7 @@ import { } from './sessionSync.svelte' import { clearRunPosition, + isCatchingUp, isDriving, isWatching, noteCaughtUp, @@ -1147,11 +1148,15 @@ 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): Promise { +async function applyTurnEnd(sessionId: string, chatId: string, attempt = 0): Promise { if (isDriving(sessionId)) return + let caughtUp = false try { const runtime = runtimes.get(sessionId) - if (!runtime) return + if (!runtime) { + caughtUp = true + 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. @@ -1162,8 +1167,20 @@ async function applyTurnEnd(sessionId: string, chatId: string): Promise { m.loadingLabel = undefined m.compacting = false const id = chatId || m.historyManager.getCurrentChatId() - if (!id) return + if (!id) { + caughtUp = true + return + } const found = await m.historyManager.reloadChat(id) + if (found === 'unavailable') { + // The store is unreadable *right now*, which says nothing about the + // conversation — so this tab still pairs the mirrored transcript with + // the history it held before the run. Freeing it here would let a turn + // go out under that pair and save it over what the driver wrote. Stay + // gated and keep asking; the read is the only thing that can settle it. + scheduleCatchUpRetry(sessionId, chatId, attempt) + return + } if (found === 'loaded') { // `refresh`: the same conversation caught up from the store, so whatever // this manager still holds for it — a queued message, the background-job @@ -1178,16 +1195,42 @@ async function applyTurnEnd(sessionId: string, chatId: string): Promise { m.adoptEmptyChat(id) setSessionChatId(sessionId, id) } - // 'unavailable' leaves everything as it is: the store is unreadable right - // now, which says nothing about the conversation. + caughtUp = true + } catch (e) { + // A read that threw leaves the same mismatched pair an 'unavailable' one + // does, so it earns the same answer rather than a silent release. + console.error('sessionRuntime: catch-up failed', e) + scheduleCatchUpRetry(sessionId, chatId, attempt) } finally { - // 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) + if (caughtUp) noteCaughtUp(sessionId) } } +/** Backoff for a catch-up that could not read the store, capped so a long + * outage settles into polling rather than growing without bound. */ +const CATCH_UP_BACKOFF_MS = [300, 700, 1500, 3000, 5000] +const catchUpRetries = new Map>() + +function scheduleCatchUpRetry(sessionId: string, chatId: string, attempt: number): void { + clearTimeout(catchUpRetries.get(sessionId)) + const delay = CATCH_UP_BACKOFF_MS[Math.min(attempt, CATCH_UP_BACKOFF_MS.length - 1)] + catchUpRetries.set( + sessionId, + setTimeout(() => { + catchUpRetries.delete(sessionId) + // Only still owed if nothing else has moved the position on: a new turn + // starting, or the session going away, both settle this on their own. + if (!isCatchingUp(sessionId)) return + void applyTurnEnd(sessionId, chatId, attempt + 1) + }, delay) + ) +} + +function cancelCatchUpRetry(sessionId: string): void { + clearTimeout(catchUpRetries.get(sessionId)) + catchUpRetries.delete(sessionId) +} + /** A Stop pressed in a watching tab reaches the run here. */ function applyCancelRequest(sessionId: string): void { if (!isDriving(sessionId)) return @@ -1248,9 +1291,11 @@ export function disposeRuntime(sessionId: string) { 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. + // frame timer with no runtime to read keeps ticking, a pending catch-up would + // re-read for a manager that is gone, and a leftover position would answer + // for a session this tab no longer holds. stopMirroring(sessionId) + cancelCatchUpRetry(sessionId) clearRunPosition(sessionId) } diff --git a/frontend/src/lib/components/sessions/sessionSync.svelte.ts b/frontend/src/lib/components/sessions/sessionSync.svelte.ts index 057a97b0f6..75b3f93917 100644 --- a/frontend/src/lib/components/sessions/sessionSync.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionSync.svelte.ts @@ -1,7 +1,13 @@ import { BROWSER } from 'esm-env' import { onUserChange, scopedKey } from '$lib/userScopedStorage' import type { DisplayMessage } from '$lib/components/copilot/chat/shared' -import { noteDriverAlive, noteRemoteTurnEnded } from './sessionRunOwner.svelte' +import { + isDriving, + noteDriverAlive, + noteDriverAnswered, + noteRemoteTurnEnded, + setDriverProbe +} from './sessionRunOwner.svelte' // 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 @@ -81,6 +87,13 @@ type QuestionAnswerMsg = { choices: string[] } +/** "Is anyone still running this?", asked by a tab about to drive where no lock + * can answer for it. Delivery is a task rather than a timer, so a driver whose + * timers have been throttled to a crawl in a hidden tab still receives it and + * still answers — which is exactly the case silence alone gets wrong. */ +type DriverProbeMsg = { kind: 'driver-probe'; sessionId: string } +type DriverAliveMsg = { kind: 'driver-alive'; sessionId: string } + type SyncMsg = | SessionPutMsg | SessionDeleteMsg @@ -91,6 +104,8 @@ type SyncMsg = | CancelRequestMsg | ToolConfirmationMsg | QuestionAnswerMsg + | DriverProbeMsg + | DriverAliveMsg type Handlers = { onSessionPut: (id: string) => void @@ -188,6 +203,15 @@ function receive(msg: SyncMsg): void { case 'question-answer': emit('onQuestionAnswer', msg.sessionId, msg.toolId, msg.choices) break + case 'driver-probe': + // Answered from inside the handler, never off a timer: a throttled tab + // still runs tasks, and routing the answer through a timeout would make + // it look dead for exactly the reason the probe exists to rule out. + if (isDriving(msg.sessionId)) post({ kind: 'driver-alive', sessionId: msg.sessionId }) + break + case 'driver-alive': + noteDriverAnswered(msg.sessionId) + break } } @@ -221,6 +245,10 @@ export function broadcastSessionArtifact(sessionId: string, artifactId: string): // Live mirroring // --------------------------------------------------------------------------- +// sessionRunOwner decides when a probe is worth sending; posting one is this +// module's job, and registering it here is what keeps the two out of a cycle. +setDriverProbe((sessionId) => post({ kind: 'driver-probe', sessionId })) + export function broadcastTurnEnd(sessionId: string, chatId: string): void { post({ kind: 'turn-end', sessionId, chatId }) }