fix(ai-sessions): probe for a live driver before taking a lockless run

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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

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<Parameters<typeof this.sendRequestImpl>[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<void> {
 		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

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<T>(
 	}
 }
 
-/** 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<string>()
+
+/** 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<T>(sessionId: string, body: () => Promise<T>): Promise<T | 'busy'> {
 	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<T>(sessionId: string, body: () => Promise<T>): Promise<T> {
  *  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<boolean> {
-	// 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()

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)
+	})
+})

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<void> {
+async function applyTurnEnd(sessionId: string, chatId: string, attempt = 0): Promise<void> {
 	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<void> {
 		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<void> {
 			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<string, ReturnType<typeof setTimeout>>()
+
+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)
 }
 

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 })
 }