From cb3b878f739f9065da09307af5371ea4d10ea7a7 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Mon, 31 Aug 2026 20:28:44 +0200 Subject: [PATCH] fix(ai-sessions): rewind a resend inside the run guard, not before it Co-Authored-By: Claude Opus 5 --- .../copilot/chat/AIChatManager.svelte.ts | 68 +++++++++++-------- .../copilot/chat/AIChatManager.test.ts | 28 ++++++++ .../sessions/sessionRunOwner.svelte.ts | 10 ++- .../sessions/sessionState.svelte.ts | 48 ++++++++++--- 4 files changed, 116 insertions(+), 38 deletions(-) diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index 68f209e897..4bea8848aa 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -2441,8 +2441,12 @@ export class AIChatManager { retryRequest = (messageIndex: number) => { const message = this.displayMessages[messageIndex] if (message && message.role === 'user') { - this.restartGeneration(messageIndex) - message.error = false + // Cleared only once the resend has started. A refusal leaves the message + // standing, so clearing it regardless would take away the Retry button + // that is the user's only way to run the retry that did not happen. + void this.restartGeneration(messageIndex).then((started) => { + if (started) message.error = false + }) } else { throw new Error('No user message found at the specified index') } @@ -2925,7 +2929,15 @@ export class AIChatManager { * spawns itself are recognised as its own rather than as rivals. */ #inGuardedRun = false - sendRequest = async (options: Parameters[0] = {}) => { + sendRequest = async ( + options: NonNullable[0]> & { + /** Rewind the conversation for a resend. Run once this tab holds the turn, + * never before: whether another tab owns the session is only settled by + * the lock inside the guard, so a caller that rewound first would have + * truncated the transcript for a send the lock then refuses. */ + rewind?: () => void + } = {} + ) => { // A turn with nowhere to render still streams, spends tokens and applies // tool calls — entirely off-screen. Refuse instead. `sendInlineRequest` is // exempt: the ⌘K widget renders its own composer inside Monaco. @@ -2952,10 +2964,14 @@ export class AIChatManager { // counts pending sends, so two independent ones racing the pre-flight // would look like recursion and both skip the guard. const guard = this.#inGuardedRun ? undefined : this.runGuard - if (!guard) return await this.sendRequestImpl(options) + if (!guard) { + options.rewind?.() + return await this.sendRequestImpl(options) + } const outcome = await guard(async () => { this.#inGuardedRun = true try { + options.rewind?.() return await this.sendRequestImpl(options) } finally { this.#inGuardedRun = false @@ -3995,15 +4011,6 @@ export class AIChatManager { throw new Error('No user message found at the specified index') } - // Refused here rather than at the send below, which is reached only after the - // truncation has already rewound this tab's transcript: the driver's turn-end - // re-read would put it back, but until then the watcher sits on a conversation - // that lost its tail for a resend that never ran. - if (this.runHeldElsewhere) { - sendUserToast('This session is running in another tab. Retry once it finishes.', true) - return - } - // Resolve the API restart point BEFORE reserving bytes or truncating: a // stale index must fail while nothing has been mutated, or the transcript // would be left truncated with the reservation leaked. A negative index @@ -4033,16 +4040,6 @@ export class AIChatManager { resentFiles.reduce((sum, f) => sum + textByteLength(f.content), 0) ) - // Remove all messages including and after the specified user message - this.displayMessages = this.displayMessages.slice(0, displayMessageIndex) - this.messages = this.messages.slice(0, actualMessageIndex) - - // The last report described the pre-rewind history; clear it. Readers - // fall back to estimating the rewound history (contextTokens), so the - // compaction trigger stays armed — e.g. for Retry after a context-length - // error, which rewinds through here. - this.contextUsage = undefined - // Resend with the message's context, not the live selection. DOM selector // chips (and other context) are one-shot — cleared from the live selection // after the first send — so reading the current selection would lose or swap @@ -4052,10 +4049,26 @@ export class AIChatManager { // contextElements. `undefined` for modes that don't attach context leaves the // live-selection behavior. An empty array is a deliberate "no context". this.instructions = newContent ?? userMessage.content - // Prune the truncated messages' file registrations BEFORE the resend - // re-registers its own — the other way around would delete the fresh rows. - this.#syncMessageFiles() - this.sendRequest({ + // Everything that rewinds the conversation, deferred to the point the turn is + // ours. Doubles as the answer to "did the resend start": it runs exactly when + // the send was not refused. + let rewound = false + const rewind = () => { + rewound = true + // Remove all messages including and after the specified user message + this.displayMessages = this.displayMessages.slice(0, displayMessageIndex) + this.messages = this.messages.slice(0, actualMessageIndex) + // The last report described the pre-rewind history; clear it. Readers + // fall back to estimating the rewound history (contextTokens), so the + // compaction trigger stays armed — e.g. for Retry after a context-length + // error, which rewinds through here. + this.contextUsage = undefined + // Prune the truncated messages' file registrations BEFORE the resend + // re-registers its own — the other way around would delete the fresh rows. + this.#syncMessageFiles() + } + await this.sendRequest({ + rewind, pastes: pastes ?? userMessage.pastes, contextOverride: editedContext ?? userMessage.contextElements, contextOverrideOrigin: 'replay', @@ -4066,6 +4079,7 @@ export class AIChatManager { files: files ?? userMessage.files, resendReservationKey }) + return rewound } fix = () => { diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index 5b45fe195a..48bbbd4bad 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts @@ -245,6 +245,34 @@ describe('AIChatManager cross-tab run guard', () => { expect(manager.queuedMessage).toBe('follow up') }) + // Whether a rival tab owns the session is settled by the lock inside the guard, + // not by the heartbeat a send reads on the way in — two tabs sending at once + // both pass that check. So the rewind has to sit behind the guard: refused in + // front of it, the loser would keep a transcript truncated for a turn it never ran. + it('leaves both histories intact when the guard refuses a retry', async () => { + const manager = new AIChatManager() + manager.isSessionChat = true + manager.displayMessages = [ + { role: 'user', content: 'first', index: 0 }, + { role: 'assistant', content: 'answer' } + ] as any + manager.messages = [ + { role: 'user', content: 'first' }, + { role: 'assistant', content: 'answer' } + ] as any + manager.runGuard = async () => 'busy' + + const started = await manager.restartGeneration(0) + + expect(started).toBe(false) + expect(mocks.runChatLoop).not.toHaveBeenCalled() + expect(manager.displayMessages).toHaveLength(2) + expect(manager.messages).toHaveLength(2) + // The prompt is handed back rather than lost, so the retry is still the + // user's to run once the other tab finishes. + expect(manager.instructions).toBe('first') + }) + // A turn flushes its queued message by re-entering sendRequest, and the lock // behind the guard is not reentrant: applying it to that nested send would // refuse the queued message as though a rival tab held the session. diff --git a/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts index a5f9c438d4..120613a810 100644 --- a/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRunOwner.svelte.ts @@ -271,12 +271,16 @@ 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) { + .map(([id, p]) => [id, p.state === 'watching' ? p.runId : ''] as const) + for (const [id, runId] of stale) { // Re-checked after the await: a status message may have landed while the // query was in flight, and reaping then would tear down a live run. if (await runLockHeld(id)) continue - if (!isWatching(id)) continue + // By run, not just by session: the silent driver's turn can end and a + // successor's begin during that query, and this session would still read as + // watching — reaping on that alone releases a run that is very much alive. + const p = positions.get(id) + if (p?.state !== 'watching' || p.runId !== runId) continue noteRemoteTurnEnded(id) driverLost?.(id) } diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 8067a6f4c4..5575548869 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -340,6 +340,12 @@ export const sessionState = $state<{ // Keyed per session: a single shared timer would let a keystroke in one draft // cancel a sibling draft's pending flush, dropping that draft's first-touch write. const draftPromptFlushHandles = new Map>() +// Sessions whose typed draft has not reached the store, from the keystroke until +// its write lands: a row read before that commit still lacks the text, so +// adopting it would drop what the user typed. Keyed by edit, so an earlier flush +// settling late cannot clear a newer edit's protection. +const draftFlushPending = new Map() +let draftFlushSeq = 0 export function setSessionDraftPrompt(sessionId: string, text: string): void { const s = sessionState.sessions.find((x) => x.id === sessionId) if (!s || s.workspace_id) return @@ -352,12 +358,16 @@ export function setSessionDraftPrompt(sessionId: string, text: string): void { // isDiscardableDraft, not `transient`, is what stops createSession reusing a // typed draft. Only the IndexedDB write is debounced. s.draftPrompt = text + const flush = ++draftFlushSeq + draftFlushPending.set(sessionId, flush) clearTimeout(draftPromptFlushHandles.get(sessionId)) draftPromptFlushHandles.set( sessionId, setTimeout(() => { draftPromptFlushHandles.delete(sessionId) - persistTouched(s) + void persistTouched(s).finally(() => { + if (draftFlushPending.get(sessionId) === flush) draftFlushPending.delete(sessionId) + }) }, 400) ) } @@ -412,10 +422,10 @@ export function takeSessionAutoSend(sessionId: string): boolean { // (transient) pending session to a durable IndexedDB record on first touch. // Non-touch writers (runtime chatId seeding, unread watermark) call putSession // directly, so an untouched draft stays in memory and vanishes on reload. -function persistTouched(s: Session): void { +function persistTouched(s: Session): Promise { if (s.transient) delete s.transient s.lastActivityAt = Date.now() - void putSession(s) + return putSession(s) } // When the session was last used. Pre-dates-the-field records report their @@ -465,6 +475,7 @@ async function putSessionRow(db: IDBPDatabase, s: Session): Promi // opened. In-memory $state is the read surface, so callers fire-and-forget. export async function putSession(s: Session): Promise { if (!BROWSER) return + localWriteSeq.set(s.id, ++localWrites) if (s.transient) return // A record removed because its workspace is gone had its files and artifacts GC'd with // it, so writing it back resurrects an empty husk. Callers reach here holding a @@ -481,10 +492,14 @@ export async function putSession(s: Session): Promise { if (all.length > 0 && !all.some((w) => w.id === boundWs)) return } ensureSessionRootId(s) + // Snapshotted before the handle is awaited, not after: a row adopted from + // another tab inside that gap would otherwise be what this write stores, in + // place of the edit it was called for. + const row = $state.snapshot(s) const db = await sessionsDb.whenReady() if (!db) return try { - await putSessionRow(db, $state.snapshot(s)) + await putSessionRow(db, row) } catch (e) { console.error('Failed to persist session', e) } @@ -506,6 +521,11 @@ export async function deleteSessionRecord(id: string): Promise { const remoteReads = new Map() let remoteReadSeq = 0 +// Bumped by every local write, so a remote read can tell whether this tab edited +// the record while that read was in flight. +const localWriteSeq = new Map() +let localWrites = 0 + // Catch up on a record another tab wrote. The notification carries only an id: // message delivery and IndexedDB commit are separately ordered, so a shipped // copy could be older than what this tab has already committed, and applying it @@ -516,6 +536,7 @@ async function applyRemoteSessionPut(id: string): Promise { if (deletedSessionIds.has(id)) return const token = ++remoteReadSeq remoteReads.set(id, token) + const localAtStart = localWriteSeq.get(id) const db = await sessionsDb.whenReady() // Only the newest token is cleared on the way out: an older read that lost the // race must leave the winner's token standing. @@ -538,6 +559,11 @@ async function applyRemoteSessionPut(id: string): Promise { // answer, so drop this one rather than race it. if (remoteReads.get(id) !== token) return remoteReads.delete(id) + // This tab edited the record while the read was in flight, so what came back + // predates that edit. Adopting it would revert the edit on screen and then + // write the revert out on this tab's next touch. Dropped rather than merged: + // the other tab's change is re-announced by any later write to the record. + if (localWriteSeq.get(id) !== localAtStart) return if (!row || deletedSessionIds.has(id)) return const i = sessionState.sessions.findIndex((s) => s.id === id) if (i >= 0) { @@ -566,10 +592,11 @@ async function applyRemoteSessionPut(id: string): Promise { * the other tab, a seen-watermark bump included. */ export function adoptRemoteRow(held: Session, row: Session): void { const stamped = new Map((held.previewTabs ?? []).map((t) => [t.id, t])) - // A draft inside its debounce window is newer than anything the store can - // hold, and the pending flush writes this same object, so taking the row's - // older text here would end up persisted over what the user is still typing. - const pendingDraft = draftPromptFlushHandles.has(held.id) ? held.draftPrompt : undefined + // A draft whose write has not landed is newer than any row. Not restored onto a + // row the other tab committed: committing consumes the draft, so putting it + // back there would re-send a prompt already sent. + const pendingDraft = + draftFlushPending.has(held.id) && !row.workspace_id ? held.draftPrompt : undefined // This file clears a field by deleting it and `putSessionRow` stores a // snapshot, so a key the row lacks is how "no longer set" arrives. Assigning // alone keeps the stale value, which this tab's next write puts back. @@ -600,6 +627,10 @@ function applyRemoteSessionDelete(id: string): void { const i = sessionState.sessions.findIndex((s) => s.id === id) if (i >= 0) sessionState.sessions.splice(i, 1) if (sessionState.currentSessionId === id) sessionState.currentSessionId = undefined + // The tombstone stops writes that have not started; one already past that check + // can still land after the other tab's delete committed, leaving the row behind + // to reappear on reload. Removing it again here collects that straggler. + void deleteSessionRecord(id) } registerSyncHandlers({ @@ -1254,6 +1285,7 @@ export function deleteSession(id: string) { // a draft deleted inside the debounce window. clearTimeout(draftPromptFlushHandles.get(id)) draftPromptFlushHandles.delete(id) + draftFlushPending.delete(id) sessionState.sessions = sessionState.sessions.filter((x) => x.id !== id) if (sessionState.currentSessionId === id) { sessionState.currentSessionId = sessionState.sessions[0]?.id