mirror of
https://github.com/stablyai/orca.git
synced 2026-11-03 02:27:43 +00:00
Stop the saved chat status writer from remembering every closed chat (#36105)
* fix(native-chat): stop the saved-status writer from remembering every closed chat The writer deduped turn-edge writes with a per-chat memo that was never cleared, so a long-running app kept one entry for every chat it had ever shown. Dedupe against the saved record instead, holding only edges whose write is still in flight; a reopened chat still skips an edge already saved. * test: cover settled chat status writer retention * test: simplify saved status retention contracts
This commit is contained in:
+14
-3
@@ -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' })
|
||||
|
||||
@@ -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<string, string>()
|
||||
private readonly queued = new Map<string, { edge: string | null }>()
|
||||
private readonly pending = new Set<Promise<void>>()
|
||||
|
||||
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<void> {
|
||||
|
||||
@@ -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<string, AgentSessionRecord>()
|
||||
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<void>; done: Promise<AgentSessionRecord> }[] = []
|
||||
let tail = Promise.resolve()
|
||||
const transition = vi.spyOn(store, 'transitionHandoff').mockImplementation((id, apply) => {
|
||||
const gate = Promise.withResolvers<void>()
|
||||
// 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' })
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user