diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.test.ts index 82bd46c9946..4476cb07a12 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.test.ts @@ -123,7 +123,10 @@ it('seeds first desktop and phone snapshots without opening, writing or reacting it('writes turn and attention edges, ignores streamed previews, and clears an empty context', async () => { await attach() - const { store, log } = hostTestState() + const { host, store, log } = hostTestState() + // Only this writer publishes the chat; a live host would write its own journal's edges. + await host.close(SESSION, 'evict') + await host.drainStatusWrites() const writer = new StructuredAgentSessionSavedStatus(() => ({ store, logger: log.logger })) const transition = vi.spyOn(store, 'transitionHandoff') const summary: AgentSessionStatusSummary = { @@ -144,6 +147,9 @@ it('writes turn and attention edges, ignores streamed previews, and clears an em at: 10, prompt: 'p'.repeat(280) }) + // A new writer dedupes against the saved row. + new StructuredAgentSessionSavedStatus(() => ({ store, logger: log.logger })).observe(summary) + expect(transition).toHaveBeenCalledTimes(1) writer.observe({ ...summary, status: 'attention', statusStartedAt: 11 }) writer.observe({ ...summary, @@ -346,7 +352,10 @@ it('lists only indexed unopened chats, keeps legacy targets, and opens history o it('keeps queued leftovers closed at startup and shows their paused cards on demand', async () => { const { host, store } = await interruptedRestart('queued-cards') await host.setSessionTabVisibility(SESSION, true) - await store.transitionHandoff(SESSION, ({ lastTurn: _lastTurn, ...record }) => record) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lastTurn: { state: 'open', at: 40, prompt: 'saved prompt from the closed chat' } + })) const openJournal = vi.spyOn(journalFactory, 'openAgentSessionJournal') openJournal.mockClear() await host.seedSessionTabs([SESSION]) @@ -355,7 +364,9 @@ it('keeps queued leftovers closed at startup and shows their paused cards on dem status: 'idle', turnOutcome: 'cancellation' }) - await host.restartResume.list() + expect(await host.restartResume.list()).toMatchObject([ + { sessionId: SESSION, latestPrompt: 'saved prompt from the closed chat' } + ]) await host.restartResume.listFailures() expect(openJournal).not.toHaveBeenCalled() const history = await host.history({ sessionId: SESSION, direction: 'tail' }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.ts index 568069954e5..e41e718c70d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-saved-status.ts @@ -52,9 +52,14 @@ export function savedStructuredAgentSessionStatus( } } -/** Only turn edges write; streamed prose and synthetic startup rows never reach this observer. */ +function savedTurnEdge(turn: AgentSessionLastTurn | undefined): string | null { + return turn ? JSON.stringify([turn.state, turn.outcome, turn.at]) : null +} + +/** Only turn edges write; streamed prose and synthetic startup rows never reach this observer. + * The saved record is the dedupe baseline, so nothing outlives a write in flight. */ export class StructuredAgentSessionSavedStatus { - private readonly edges = new Map() + private readonly queued = new Map() private readonly pending = new Set>() constructor( @@ -62,6 +67,7 @@ export class StructuredAgentSessionSavedStatus { ) {} observe = (summary: AgentSessionStatusSummary): void => { + const { sessionId } = summary const state = summary.status === 'working' || summary.status === 'attention' ? 'open' : 'settled' const outcome = state === 'settled' ? summary.turnOutcome : undefined @@ -70,11 +76,6 @@ export class StructuredAgentSessionSavedStatus { if (at === undefined && state === 'open') { return } - const key = JSON.stringify([state, outcome, at]) - if (this.edges.get(summary.sessionId) === key) { - return - } - this.edges.set(summary.sessionId, key) const lastTurn: AgentSessionLastTurn | undefined = at === undefined ? undefined @@ -87,26 +88,39 @@ export class StructuredAgentSessionSavedStatus { ? { reply: summary.lastAssistantMessage.slice(0, AGENT_SESSION_TURN_PREVIEW_CHARS) } : {}) } + const edge = savedTurnEdge(lastTurn) + const latest = this.queued.get(sessionId) + const baseline = latest + ? latest.edge + : savedTurnEdge(this.deps().store.getRecord(sessionId)?.lastTurn) + if (baseline === edge) { + return + } + const entry = { edge } + this.queued.set(sessionId, entry) const write = this.deps() - .store.transitionHandoff(summary.sessionId, (record) => { + .store.transitionHandoff(sessionId, (record) => { const { lastTurn: _previous, ...rest } = record return lastTurn ? { ...rest, lastTurn } : _previous ? rest : record }) .then( () => {}, (error: unknown) => { - if (this.edges.get(summary.sessionId) === key) { - this.edges.delete(summary.sessionId) - } this.deps().logger.warn('saving chat status failed', { scope: 'saved-status', - sessionId: summary.sessionId, + sessionId, error }) } ) this.pending.add(write) - void write.finally(() => this.pending.delete(write)) + void write.finally(() => { + this.pending.delete(write) + // A later queued write still owns the baseline; a failure falls back to the saved row and retries. + if (this.queued.get(sessionId) === entry) { + this.queued.delete(sessionId) + } + }) } async drain(): Promise { diff --git a/src/main/runtime/agent-session-saved-status-scale.test.ts b/src/main/runtime/agent-session-saved-status-scale.test.ts index 3b64a2bbf48..7649e990f52 100644 --- a/src/main/runtime/agent-session-saved-status-scale.test.ts +++ b/src/main/runtime/agent-session-saved-status-scale.test.ts @@ -2,7 +2,9 @@ import { expect, it, vi } from 'vitest' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { AGENT_JOURNAL_THREAD_SCOPE } from '../../shared/agent-session-journal-types' import { agentSessionOperationKey } from '../../shared/agent-session-operation-ledger' -import { scaleStore, scaleJournal } from './agent-session-scale.test-fixture' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import type { AgentSessionStatusSummary } from '../../shared/agent-session-wire' +import { scaleStore, scaleJournal, scaleRecord } from './agent-session-scale.test-fixture' import { editPersistedTestAgentSessionStore, openTestAgentSessionRecordStore @@ -131,3 +133,115 @@ it.each([10, 100])( unsubscribe() } ) + +// Count writer-owned collections to catch historical retention without a production test getter. +function retainedEntries(writer: StructuredAgentSessionSavedStatus): number { + return Object.values(writer).reduce( + (count, value) => count + (value instanceof Map || value instanceof Set ? value.size : 0), + 0 + ) +} + +function turnSummary(sessionId: string, at: number): AgentSessionStatusSummary { + return { + sessionId, + workspaceId: 'workspace-1', + agent: 'claude', + status: 'idle', + turnOutcome: 'success', + latestPrompt: 'saved prompt', + statusStartedAt: at, + updatedAt: at + } +} + +it.each([1_000, 10_000])( + 'releases production writer retention after %i distinct historical chats', + async (n) => { + const { store } = await scaleStore(1) + const rows = new Map() + vi.spyOn(store, 'getRecord').mockImplementation((id) => rows.get(id) ?? null) + // Benchmark writer retention only; avoid quadratic whole-store transactions in this corpus. + const transition = vi + .spyOn(store, 'transitionHandoff') + .mockImplementation(async (id, apply) => { + const record = rows.get(id) + if (!record) { + throw new Error('missing benchmark row') + } + const next = apply(record) + rows.set(id, next) + return next + }) + const writer = new StructuredAgentSessionSavedStatus(() => ({ + store, + logger: createStructuredAgentSessionLogger() + })) + for (let i = 0; i < n; i++) { + const id = `historical-${i}` + rows.set(id, scaleRecord(id)) + for (const at of [1, 2]) { + writer.observe(turnSummary(id, at)) + await writer.drain() + } + } + expect(transition).toHaveBeenCalledTimes(2 * n) + expect([...rows.values()].every((row) => row.lastTurn?.at === 2)).toBe(true) + expect(retainedEntries(writer)).toBe(0) + } +) + +it('preserves X/Y/X/Y writes and retries a failed final edge on publication', async () => { + const { store, rows } = await scaleStore(1) + const id = rows[0]!.sessionId + const original = store.transitionHandoff + const held: { gate: PromiseWithResolvers; done: Promise }[] = [] + let tail = Promise.resolve() + const transition = vi.spyOn(store, 'transitionHandoff').mockImplementation((id, apply) => { + const gate = Promise.withResolvers() + // Match production FIFO: only a successful commit publishes the row, rejection advances the tail. + const done = tail.then(() => gate.promise).then(() => original(id, apply)) + tail = done.then( + () => {}, + () => {} + ) + held.push({ gate, done }) + return done + }) + const warn = vi.fn() + const writer = new StructuredAgentSessionSavedStatus(() => ({ + store, + logger: { warn, error: vi.fn() } + })) + const x = turnSummary(id, 1) + const y = turnSummary(id, 2) + for (const summary of [x, y, x, y]) { + writer.observe(summary) + } + expect(transition).toHaveBeenCalledTimes(4) + for (let i = 0; i < held.length; i++) { + const write = held[i]! + if (i % 2) { + write.gate.reject(new Error('controlled persistence rejection')) + } else { + write.gate.resolve() + } + await write.done.catch(() => {}) + await Promise.resolve() + await Promise.resolve() + if (i < 3) { + writer.observe(y) + expect(transition).toHaveBeenCalledTimes(4) + } + } + await writer.drain() + expect(warn).toHaveBeenCalledTimes(2) + expect(store.getRecord(id)?.lastTurn?.at).toBe(1) + // A normal same-edge publication is the retry wake; drain alone schedules nothing. + expect(transition).toHaveBeenCalledTimes(4) + writer.observe(y) + expect(transition).toHaveBeenCalledTimes(5) + held[4]!.gate.resolve() + await writer.drain() + expect(store.getRecord(id)?.lastTurn).toMatchObject({ at: 2, prompt: 'saved prompt' }) +})