diff --git a/config/scripts/pr-code-change-scope.mjs b/config/scripts/pr-code-change-scope.mjs index 13a13794459..3ea588aaf8f 100644 --- a/config/scripts/pr-code-change-scope.mjs +++ b/config/scripts/pr-code-change-scope.mjs @@ -96,11 +96,14 @@ const CROSS_VERSION_WIRE_PREFIXES = [ 'src/shared/agent-session-wire', 'src/shared/agent-session-mutation-envelope', 'src/shared/agent-session-journal-', + 'src/main/ai-vault/structured-session-ownership.ts', 'src/main/native-chat/agent-session-journal/', 'src/main/native-chat/agent-session-wire/', 'src/main/runtime/agent-session-record-store', 'src/main/runtime/rpc/dispatcher', + 'src/main/runtime/rpc/methods/ai-vault.ts', 'src/main/runtime/rpc/methods/browser-tab-create-schema', + 'src/main/runtime/rpc/methods/session-tabs.ts', 'src/main/runtime/rpc/methods/structured-agent-session', 'src/main/runtime/rpc/methods/terminal', 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer' diff --git a/config/scripts/pr-code-change-scope.test.mjs b/config/scripts/pr-code-change-scope.test.mjs index 99fa7685372..3f35dae1dd9 100644 --- a/config/scripts/pr-code-change-scope.test.mjs +++ b/config/scripts/pr-code-change-scope.test.mjs @@ -185,13 +185,16 @@ describe('per-job path classification', () => { 'src/shared/agent-session-mutation-envelope.ts', 'src/shared/agent-session-journal-item-key.ts', 'src/shared/agent-session-journal-types.ts', + 'src/main/ai-vault/structured-session-ownership.ts', 'src/main/native-chat/agent-session-journal/journal-cursor.ts', 'src/main/native-chat/agent-session-journal/journal-reducer.ts', 'src/main/native-chat/agent-session-journal/journal-row-schema.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-host.ts', 'src/main/runtime/agent-session-record-store.ts', 'src/main/runtime/rpc/dispatcher.ts', + 'src/main/runtime/rpc/methods/ai-vault.ts', 'src/main/runtime/rpc/methods/browser-tab-create-schema.ts', + 'src/main/runtime/rpc/methods/session-tabs.ts', 'src/main/runtime/rpc/methods/structured-agent-session.ts', 'src/main/runtime/rpc/methods/structured-agent-session-gate.ts', 'src/main/runtime/rpc/methods/structured-agent-session-hold.ts', diff --git a/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts b/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts index 4b4e7f64968..429881e274a 100644 --- a/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts +++ b/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts @@ -2,10 +2,12 @@ // keeps its intact prefix live and moves the unreadable remainder aside, so the // bytes stay on disk for inspection instead of being rebuilt into an empty epoch. +import { readFile } from 'node:fs/promises' +import { join } from 'node:path' import { + JOURNAL_SNAPSHOT_FILE, quarantineJournalRemainder, readJournalLog, - readJournalSnapshotFile, rewriteJournalLog } from './journal-log-file' import type { JournalRow } from './journal-row-schema' @@ -23,12 +25,14 @@ export async function quarantineCorruptSuffix( } /** Copy everything aside before a read-only journal is rebuilt under a newer - * schema: those rows are unreadable to THIS build, not worthless. */ + * schema: those rows are unreadable to THIS build, not worthless. The + * snapshot is preserved as raw bytes — a future-version snapshot does not + * parse under this build's schema, and its bytes must survive verbatim. */ export async function quarantineUnreadableSchema(journalDir: string): Promise { - const snapshot = await readJournalSnapshotFile(journalDir) + const snapshot = await readSnapshotBytes(journalDir) const log = await readJournalLog(journalDir) const preserved = [ - snapshot ? JSON.stringify(snapshot) : '', + snapshot ?? '', log.rows.map((row) => JSON.stringify(row)).join('\n'), log.remainder ?? '' ] @@ -39,6 +43,17 @@ export async function quarantineUnreadableSchema(journalDir: string): Promise { + try { + return await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return null + } + throw error + } +} + /** The disclosure row for lines that failed to parse. Skipped lines are lost * rows; counting them silently is the drop this exists to prevent. */ export function malformedRowsDisclosure(count: number): { diff --git a/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts b/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts index d381f8a8dc4..ea0dfb5cba9 100644 --- a/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts @@ -194,3 +194,78 @@ describe('oversized identity bounding on Unicode boundaries', () => { expect(() => encodeURIComponent(bounded)).not.toThrow() }) }) + +describe('ill-formed UTF-16 identity totality', () => { + // JSON.parse admits lone surrogates, so any JSON string is a legal component; + // pre-fix these threw `URIError: URI malformed` in `encodeURIComponent`. + const LONE_HIGH = '\ud83d' + const LONE_LOW = '\ude00' + + it('keys a lone-high-surrogate id without throwing, deterministically', () => { + const identity: AgentJournalItemIdentity = { + provider: 'claude', + sessionId: LONE_HIGH, + uuid: 'u-1' + } + expect(() => agentJournalItemKey(identity)).not.toThrow() + expect(agentJournalItemKey(identity)).toBe(agentJournalItemKey(identity)) + }) + + it('keys an oversized id carrying a lone low surrogate inside the head', () => { + const identity: AgentJournalItemIdentity = { + provider: 'codex', + threadId: THREAD, + turnId: `${LONE_LOW}${'x'.repeat(1100)}`, + ordinal: 0 + } + expect(() => agentJournalItemKey(identity)).not.toThrow() + }) + + it('keys an oversized id with a lone high surrogate away from the head cut', () => { + const identity: AgentJournalItemIdentity = { + provider: 'codex', + threadId: THREAD, + turnId: `${'a'.repeat(10)}${LONE_HIGH}${'b'.repeat(1100)}`, + ordinal: 0 + } + expect(() => agentJournalItemKey(identity)).not.toThrow() + }) + + it('cannot collide an ill-formed id with its replacement-character spelling', () => { + const keyFor = (sessionId: string) => + agentJournalItemKey({ provider: 'claude', sessionId, uuid: 'u-1' }) + expect(keyFor(LONE_HIGH)).not.toBe(keyFor('�')) + expect(keyFor(LONE_HIGH)).not.toBe(keyFor(LONE_LOW)) + const oversized = (head: string) => `${head}${'x'.repeat(1100)}` + expect(keyFor(oversized(LONE_HIGH))).not.toBe(keyFor(oversized('�'))) + }) + + it('re-deriving from a parsed ill-formed key is a fixed point', () => { + const key = agentJournalItemKey({ provider: 'claude', sessionId: LONE_HIGH, uuid: 'u-1' }) + const parsed = parseAgentJournalItemKey(key) + expect(parsed).not.toBeNull() + expect(agentJournalItemKey(parsed as AgentJournalItemIdentity)).toBe(key) + }) + + it('leaves well-formed short components verbatim', () => { + expect(boundJournalKeyComponent('turn-1')).toBe('turn-1') + expect(boundJournalKeyComponent('😀 café')).toBe('😀 café') + }) +}) + +describe('malformed persisted keys decode to null instead of throwing', () => { + it('returns null for malformed percent sequences', () => { + for (const key of ['%', 'claude:%E0%A4%A:u-1', 'codex:a:b:1%ZZ', 'claude:%ED%A0%BD:u-1']) { + expect(parseAgentJournalItemKey(key)).toBeNull() + } + }) + + it('still round-trips well-formed keys containing the delimiter and spaces', () => { + const identity: AgentJournalItemIdentity = { + provider: 'claude', + sessionId: 's:1', + uuid: 'u 1' + } + expect(parseAgentJournalItemKey(agentJournalItemKey(identity))).toEqual(identity) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.test.ts b/src/main/native-chat/agent-session-journal/journal-log-file.test.ts index e9a98f45374..ff710db67ed 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-log-file.test.ts @@ -234,6 +234,39 @@ describe('readJournalSnapshot validation', () => { }) }) +describe('future-version snapshot classification', () => { + it('classifies a future version before shape validation so unknown bodies stay unreadable', async () => { + // The version can only advance because bodies changed, so a future snapshot + // legitimately carries kinds this build cannot parse. That is the + // schema-unreadable contract, not corruption. + const future = validSnapshot() as unknown as Record + future.v = 99 + future.items = [ + { + itemId: 'codex:thread-1:turn-1:1', + revision: 1, + body: { kind: 'future-render-kind', payload: { anything: true } }, + sequence: 2, + observedAt: 1_000 + } + ] + await writeSnapshot(future) + expect((await readJournalSnapshot(root)).status).toBe('unreadable') + }) + + it('classifies a future version as unreadable even when its shapes still parse today', async () => { + await writeSnapshot({ ...validSnapshot(), v: 99 }) + expect((await readJournalSnapshot(root)).status).toBe('unreadable') + }) + + it('treats a non-integer or sub-1 version as invalid, matching row admission', async () => { + for (const v of [0, 1.5]) { + await writeSnapshot({ ...validSnapshot(), v }) + expect((await readJournalSnapshot(root)).status).toBe('invalid') + } + }) +}) + describe('journal startup isolation from a malformed snapshot', () => { it('quarantines a JSON-valid malformed snapshot instead of throwing through open', async () => { await writeSnapshot({ ...validSnapshot(), tombstones: {} }) diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts index 349953c8389..b556177b889 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.ts +++ b/src/main/native-chat/agent-session-journal/journal-log-file.ts @@ -12,9 +12,10 @@ import { appendFile, mkdir, open, readFile, type FileHandle } from 'node:fs/prom import { randomUUID } from 'node:crypto' import { join } from 'node:path' import { durableWriteTempPath, renameDurable, writeFileDurable } from '../../durable-file-write' -import type { - AgentJournalRenderItem, - AgentJournalSubmission +import { + AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + type AgentJournalRenderItem, + type AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { isAdmissibleAgentJournalRenderItem, @@ -68,6 +69,9 @@ export type JournalSnapshotReadResult = | { status: 'missing' } | { status: 'valid'; snapshot: JournalSnapshotFile } | { status: 'invalid' } + /** A future schema version: unreadable by this build, not corrupt. The file + * stays authoritative in place and the caller degrades to read-only. */ + | { status: 'unreadable' } const NEWLINE_BYTE = 0x0a @@ -75,17 +79,20 @@ export async function ensureJournalDir(journalDir: string): Promise { await mkdir(journalDir, { recursive: true }) } -export async function readJournalSnapshotFile( - journalDir: string -): Promise { - const result = await readJournalSnapshot(journalDir) - return result.status === 'valid' ? result.snapshot : null -} - export async function readJournalSnapshot(journalDir: string): Promise { try { const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') const parsed: unknown = JSON.parse(raw) + const version = snapshotSchemaVersion(parsed) + if (version === null) { + return { status: 'invalid' } + } + // Version is classified BEFORE shape validation, matching row admission: a + // version only advances because bodies changed, so a valid newer snapshot + // carries kinds this build cannot parse — unreadable, never corruption. + if (version > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) { + return { status: 'unreadable' } + } return isJournalSnapshotFile(parsed) ? { status: 'valid', snapshot: parsed } : { status: 'invalid' } @@ -151,6 +158,14 @@ export async function readJournalLog(journalDir: string): Promise 0 } } +/** Row admission requires an integer version of at least 1; a snapshot whose + * version cannot even be read is malformed, not a schema statement. */ +function snapshotSchemaVersion(value: unknown): number | null { + const snapshot = recordOf(value) + const version = snapshot?.v + return typeof version === 'number' && Number.isInteger(version) && version >= 1 ? version : null +} + function isJournalSnapshotFile(value: unknown): value is JournalSnapshotFile { if (!value || typeof value !== 'object' || Array.isArray(value)) { return false diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 1ab61240ac1..0dbee7b4005 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -14,7 +14,6 @@ import { readJournalSnapshot, type JournalSnapshotFile } from './journal-log-file' -import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' import { applyJournalRow, createJournalReducerState, @@ -46,6 +45,13 @@ export async function loadJournal( sessionId: string ): Promise { const snapshotRead = await readJournalSnapshot(journalDir) + if (snapshotRead.status === 'unreadable') { + // Written by a newer schema: not corrupt, so never quarantined. The file + // stays authoritative in place, and this build must not write, compact, + // delete, or render a partial timeline from rows it cannot anchor to the + // snapshot it cannot read. + return emptyReadOnlyLoad(sessionId) + } if (snapshotRead.status === 'invalid') { await quarantineInvalidJournalSnapshot(journalDir) } @@ -94,7 +100,9 @@ export async function loadJournal( state, tailRows, compactedThrough, - readOnly: log.unreadable || (snapshot?.v ?? 0) > AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + // A future-version snapshot never reaches here: it is classified + // unreadable above, so `valid` implies a version this build can write. + readOnly: log.unreadable, corrupt, malformedRows: log.malformed, sizeBytes: tailRows.reduce((total, row) => total + journalRowByteLength(row), 0), diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts index d09f4fab2d7..cab7c688e1c 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts @@ -407,3 +407,14 @@ describe('blob retention', () => { expect(referencedBlobDigests(state).size).toBe(0) }) }) + +describe('malformed persisted item keys', () => { + it('degrades a malformed-percent item id to an opaque key instead of throwing', () => { + // A user-message body drives identity resolution through the key parser; + // pre-fix `parseAgentJournalItemKey('%')` threw `URIError: URI malformed`. + const state = fold([ + { kind: 'item', itemId: '%', revision: 1, body: userText('hi'), ...base(1) } + ]) + expect(renderJournalState(state).items[0]?.itemId).toBe('%') + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index ad48723604e..42ace58d1cc 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -568,6 +568,87 @@ describe('schema', () => { }) }) + it('preserves a future-version snapshot with an unknown body kind in place instead of quarantining it', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE) + const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as Record + snapshot.v = 99 + // The version advances because bodies changed: a valid newer snapshot + // carries kinds this build cannot parse and must stay unreadable in place. + snapshot.items = [ + { + itemId: 'codex:thread-1:turn-1:1', + revision: 1, + body: { kind: 'future-render-kind', payload: { anything: true } }, + sequence: 1, + observedAt: 1_000 + } + ] + await writeFile(snapshotPath, JSON.stringify(snapshot), 'utf-8') + + const reopened = await open() + const entries = await readdir(root) + expect(entries.some((name) => name.startsWith('quarantine-'))).toBe(false) + expect(entries.includes(JOURNAL_SNAPSHOT_FILE)).toBe(true) + expect(reopened.isReadOnly).toBe(true) + expect(reopened.snapshot().items).toHaveLength(0) + await expect(reopened.appendItem(item(1), body('b'), { fence: 1 })).rejects.toMatchObject({ + code: 'journal_read_only' + }) + }) + + it('keeps the future-version snapshot bytes when the schema escape hatch rolls the epoch', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE) + const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as Record + snapshot.v = 99 + snapshot.items = [ + { + itemId: 'codex:thread-1:turn-1:1', + revision: 1, + body: { kind: 'future-render-kind', payload: { anything: true } }, + sequence: 1, + observedAt: 1_000 + } + ] + await writeFile(snapshotPath, JSON.stringify(snapshot), 'utf-8') + const reopened = await open() + // Still live in place before the explicit escape hatch runs. + expect((await readdir(root)).some((name) => name.startsWith('quarantine-'))).toBe(false) + + await reopened.rollEpoch('schema_unreadable', 2) + expect(reopened.isReadOnly).toBe(false) + const quarantine = (await readdir(root)).find((name) => name.startsWith('quarantine-')) + expect(quarantine).toBeDefined() + expect(await readFile(join(root, quarantine!), 'utf-8')).toContain('future-render-kind') + }) + + it('reopens a log holding an admitted malformed-percent item id without throwing', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const logPath = join(root, JOURNAL_LOG_FILE) + // `parseJournalRow` admits any string itemId, so replay must degrade a + // malformed percent key to an opaque id instead of throwing URIError. + const malformedKeyRow = JSON.stringify({ + v: 1, + epoch: journal.epoch, + seq: journal.cursor().sequence + 1, + fence: 1, + ts: 1, + kind: 'item', + itemId: '%', + revision: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] } + }) + await writeFile(logPath, `${await readFile(logPath, 'utf-8')}${malformedKeyRow}\n`, 'utf-8') + + const reopened = await open() + expect(reopened.isReadOnly).toBe(false) + expect(reopened.snapshot().items.some((entry) => entry.itemId === '%')).toBe(true) + }) + it('allows the explicit schema-unreadable epoch escape hatch while preserving the old files', async () => { const journal = await open() await journal.appendItem(item(0), body('a'), { fence: 1 }) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts index 96f7a6f7a0f..4c43326acb1 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts @@ -54,6 +54,8 @@ function createReadOwner( } let stopActiveRun: (() => void) | null = null let refreshActiveRun = (): void => {} + const retiredHistoryRead = (): boolean => true + let captureActiveHistoryReadGuard = (): (() => boolean) => retiredHistoryRead const activations = new Set() const listeners = new Set<() => void>() @@ -80,6 +82,11 @@ function createReadOwner( setSnapshot({ ...snapshot, providerSession }) } } + const clearLoadingOlder = (): void => { + if (snapshot.loadingOlder) { + setSnapshot({ ...snapshot, loadingOlder: false }) + } + } const refreshTail = async (shouldStop: () => boolean): Promise => { const result = await callStructuredAgentSession( target, @@ -91,6 +98,9 @@ function createReadOwner( } setProviderSession(result.providerSession) if (!result.ok) { + if (shouldStop()) { + return + } apply({ type: 'event', event: { @@ -103,7 +113,13 @@ function createReadOwner( }) return } + if (shouldStop()) { + return + } apply({ type: 'tail-page', page: result.page }) + if (shouldStop()) { + return + } let restored = snapshot.state.items.filter(countsTowardInitialHistory).length while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) { const oldest = oldestStructuredAgentSessionCursor(snapshot.state) @@ -127,7 +143,13 @@ function createReadOwner( if (!older.ok || older.page.window.oldest?.sequence === oldest.sequence) { break } + if (shouldStop()) { + return + } apply({ type: 'older-page', requestedEpoch: oldest.epoch, page: older.page }) + if (shouldStop()) { + return + } restored = snapshot.state.items.filter(countsTowardInitialHistory).length } } @@ -140,12 +162,15 @@ function createReadOwner( applyEvent: (event) => apply({ type: 'event', event }), applyError: (message) => apply({ type: 'error', message }), getCursor: () => snapshot.state.cursor, + onHistoryReadInvalidated: clearLoadingOlder, refreshTail, sessionId, target }) + captureActiveHistoryReadGuard = transport.captureHistoryReadGuard refreshActiveRun = transport.refresh stopActiveRun = () => { + captureActiveHistoryReadGuard = () => retiredHistoryRead refreshActiveRun = (): void => {} transport.dispose() stopActiveRun = null @@ -180,10 +205,17 @@ function createReadOwner( }, getSnapshot: () => snapshot, loadOlder: async () => { + const shouldStop = captureActiveHistoryReadGuard() + if (shouldStop()) { + return + } const cursor = oldestStructuredAgentSessionCursor(snapshot.state) if (!cursor || !snapshot.state.hasOlder || snapshot.loadingOlder) { return } + if (shouldStop()) { + return + } setSnapshot({ ...snapshot, loadingOlder: true }) try { const result = await callStructuredAgentSession( @@ -191,13 +223,20 @@ function createReadOwner( 'agentSession.history', { sessionId, direction: 'before', cursor, limit: AGENT_SESSION_HISTORY_MAX_LIMIT } ) - if (result.ok) { + if (shouldStop()) { + return + } + if (result.ok && !shouldStop()) { apply({ type: 'older-page', requestedEpoch: cursor.epoch, page: result.page }) } } catch (error) { - apply({ type: 'error', message: String(error) }) + if (!shouldStop()) { + apply({ type: 'error', message: String(error) }) + } } finally { - setSnapshot({ ...snapshot, loadingOlder: false }) + if (!shouldStop()) { + clearLoadingOlder() + } } }, refresh: () => refreshActiveRun(), diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts index 053492cf908..b420cc5b529 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts @@ -69,6 +69,7 @@ describe('structured agent-session read transport generations', () => { applyEvent, applyError, getCursor: () => null, + onHistoryReadInvalidated: () => undefined, refreshTail: async () => undefined, sessionId: 'session-a', target diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts index 5065759a201..79232a4ead7 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts @@ -32,27 +32,54 @@ export function startStructuredAgentSessionReadTransport(args: { applyEvent: (event: AgentSessionSubscribeEvent) => void applyError: (message: string) => void getCursor: () => AgentJournalCursor | null + onHistoryReadInvalidated: () => void refreshTail: (shouldStop: () => boolean) => Promise sessionId: string target: RuntimeClientTarget -}): { dispose: () => void; refresh: () => void } { +}): { + captureHistoryReadGuard: () => () => boolean + dispose: () => void + refresh: () => void +} { let stopped = false let connected = false let opening = false - let generation = 0 + let openGeneration = 0 + let stateGeneration = 0 let unsubscribe = (): void => {} let resumeCursor = args.getCursor() - const coalescer = createStructuredAgentSessionEventCoalescer(args.applyEvent) + let shouldStopCoalescedEvent = (): boolean => true + const coalescer = createStructuredAgentSessionEventCoalescer((event) => { + if (!shouldStopCoalescedEvent()) { + args.applyEvent(event) + } + }) const reconnectScheduler = createReconnectScheduler({ shouldStop: () => stopped || connected, reconnect: () => void open() }) - const isCurrentGeneration = (candidate: number): boolean => !stopped && candidate === generation - const handleEvent = (event: AgentSessionSubscribeEvent, eventGeneration: number): void => { - if (!isCurrentGeneration(eventGeneration)) { + const isCurrentOpenGeneration = (candidate: number): boolean => + !stopped && candidate === openGeneration + const captureHistoryReadGuard = (): (() => boolean) => { + const readOpenGeneration = openGeneration + const readStateGeneration = stateGeneration + return () => + !isCurrentOpenGeneration(readOpenGeneration) || readStateGeneration !== stateGeneration + } + const handleEvent = (event: AgentSessionSubscribeEvent, eventOpenGeneration: number): void => { + if (!isCurrentOpenGeneration(eventOpenGeneration)) { return } if (event.type === 'snapshot' || event.type === 'reset') { + coalescer.flush() + if (!isCurrentOpenGeneration(eventOpenGeneration)) { + return + } + stateGeneration += 1 + args.onHistoryReadInvalidated() + if (!isCurrentOpenGeneration(eventOpenGeneration)) { + return + } resumeCursor = event.page.liveCursor ?? event.page.window.nextCursor } else if ( event.type === 'batch' && @@ -63,6 +90,7 @@ export function startStructuredAgentSessionReadTransport(args: { connected = false reconnectScheduler.schedule() } + shouldStopCoalescedEvent = captureHistoryReadGuard() coalescer.push(event) } async function open(): Promise { @@ -75,17 +103,25 @@ export function startStructuredAgentSessionReadTransport(args: { } opening = true coalescer.flush() - const openGeneration = ++generation + if (stopped) { + opening = false + return + } + const currentOpenGeneration = ++openGeneration + args.onHistoryReadInvalidated() unsubscribe() unsubscribe = (): void => {} try { + if (!isCurrentOpenGeneration(currentOpenGeneration)) { + return + } let closedDuringOpen = false const handle = await subscribeStructuredAgentSession( args.target, { sessionId: args.sessionId, ...(resumeCursor ? { cursor: resumeCursor } : {}) }, - (event) => handleEvent(event, openGeneration), + (event) => handleEvent(event, currentOpenGeneration), (error) => { - if (!isCurrentGeneration(openGeneration)) { + if (!isCurrentOpenGeneration(currentOpenGeneration)) { return } closedDuringOpen = true @@ -94,7 +130,7 @@ export function startStructuredAgentSessionReadTransport(args: { reconnectScheduler.schedule() }, () => { - if (!isCurrentGeneration(openGeneration)) { + if (!isCurrentOpenGeneration(currentOpenGeneration)) { return } closedDuringOpen = true @@ -102,9 +138,9 @@ export function startStructuredAgentSessionReadTransport(args: { reconnectScheduler.schedule() } ) - if (!isCurrentGeneration(openGeneration) || closedDuringOpen) { + if (!isCurrentOpenGeneration(currentOpenGeneration) || closedDuringOpen) { handle.unsubscribe() - if (isCurrentGeneration(openGeneration)) { + if (isCurrentOpenGeneration(currentOpenGeneration)) { reconnectScheduler.schedule() } } else { @@ -112,24 +148,24 @@ export function startStructuredAgentSessionReadTransport(args: { unsubscribe = handle.unsubscribe } } catch (error) { - if (!isCurrentGeneration(openGeneration)) { + if (!isCurrentOpenGeneration(currentOpenGeneration)) { return } connected = false args.applyError(String(error)) reconnectScheduler.schedule() } finally { - if (openGeneration === generation) { + if (currentOpenGeneration === openGeneration) { opening = false } } } const refresh = (): void => { - const refreshGeneration = generation + const shouldStop = captureHistoryReadGuard() void args - .refreshTail(() => !isCurrentGeneration(refreshGeneration)) + .refreshTail(shouldStop) .then(() => { - if (!isCurrentGeneration(refreshGeneration)) { + if (shouldStop()) { return } resumeCursor = args.getCursor() @@ -138,31 +174,33 @@ export function startStructuredAgentSessionReadTransport(args: { } }) .catch((error) => { - if (isCurrentGeneration(refreshGeneration)) { + if (!shouldStop()) { args.applyError(String(error)) } }) } - const initialGeneration = generation + const shouldStopInitialRead = captureHistoryReadGuard() void args - .refreshTail(() => !isCurrentGeneration(initialGeneration)) + .refreshTail(shouldStopInitialRead) .then(() => { - if (!isCurrentGeneration(initialGeneration)) { + if (shouldStopInitialRead()) { return } resumeCursor = args.getCursor() return open() }) .catch((error) => { - if (isCurrentGeneration(initialGeneration)) { + if (!shouldStopInitialRead()) { args.applyError(String(error)) reconnectScheduler.schedule() } }) return { + captureHistoryReadGuard, dispose: () => { stopped = true - generation += 1 + openGeneration += 1 + args.onHistoryReadInvalidated() reconnectScheduler.dispose() coalescer.dispose() unsubscribe() diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx index 97e68eb6ccc..2d320694c9c 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx @@ -8,7 +8,8 @@ import type { } from '../../../../shared/agent-session-journal-types' import { AGENT_SESSION_HISTORY_MAX_LIMIT, - type AgentSessionHistoryPage + type AgentSessionHistoryPage, + type AgentSessionSubscribeEvent } from '../../../../shared/agent-session-wire' const mocks = vi.hoisted(() => ({ call: vi.fn(), subscribe: vi.fn() })) @@ -57,14 +58,15 @@ function providerFrame(id: string, sequence: number): AgentJournalRenderItem { function page( direction: 'tail' | 'before', items: AgentJournalRenderItem[], - hasOlder: boolean + hasOlder: boolean, + epoch = 'epoch-a' ): AgentSessionHistoryPage { - const cursor = (sequence: number): AgentJournalCursor => ({ epoch: 'epoch-a', sequence }) + const cursor = (sequence: number): AgentJournalCursor => ({ epoch, sequence }) const oldest = items[0]?.sequence ?? 0 const newest = items.at(-1)?.sequence ?? oldest return { sessionId: 'session-a', - epoch: 'epoch-a', + epoch, direction, items, removedItemIds: [], @@ -255,6 +257,82 @@ describe('useStructuredAgentSessionRead history window', () => { } }) + it.each(['snapshot', 'reset'] as const)( + 'drops a delayed refresh after a same-stream %s advances the epoch', + async (eventType) => { + const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true) + const delayedRefresh = Promise.withResolvers<{ + ok: true + page: AgentSessionHistoryPage + providerSession: { key: 'session_id'; id: string } + }>() + const onEvents: ((event: AgentSessionSubscribeEvent) => void)[] = [] + const initialProviderSession = { key: 'session_id', id: 'provider-initial' } as const + mocks.call + .mockResolvedValueOnce({ + ok: true, + page: page('tail', [message('initial', 1, 'assistant')], false), + providerSession: initialProviderSession + }) + .mockReturnValueOnce(delayedRefresh.promise) + mocks.subscribe.mockImplementation((_target, _params, onEvent) => { + onEvents.push(onEvent) + return Promise.resolve({ unsubscribe: vi.fn() }) + }) + + const view = renderHook(() => + useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET }) + ) + + try { + await waitFor(() => expect(onEvents).toHaveLength(1)) + act(() => window.dispatchEvent(new Event('focus'))) + await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2)) + + const replacementPage = page( + 'tail', + [message('new-epoch', 2, 'assistant')], + false, + 'epoch-b' + ) + const replacementEvent: AgentSessionSubscribeEvent = + eventType === 'reset' + ? { + type: 'reset', + sessionId: 'session-a', + reset: 'epoch_changed', + page: replacementPage, + fence: 2 + } + : { type: 'snapshot', sessionId: 'session-a', page: replacementPage, fence: 2 } + act(() => onEvents[0]?.(replacementEvent)) + + expect(view.result.current.state.epoch).toBe('epoch-b') + expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch') + expect(view.result.current.providerSession).toBe(initialProviderSession) + const stateAfterReplacement = view.result.current.state + + await act(async () => { + delayedRefresh.resolve({ + ok: true, + page: page('tail', [message('stale-refresh', 3, 'assistant')], false), + providerSession: { key: 'session_id', id: 'provider-stale' } + }) + await delayedRefresh.promise + await Promise.resolve() + }) + + expect(view.result.current.state).toBe(stateAfterReplacement) + expect(view.result.current.state.epoch).toBe('epoch-b') + expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch') + expect(view.result.current.providerSession).toBe(initialProviderSession) + } finally { + view.unmount() + hasFocus.mockRestore() + } + } + ) + it('does no host work for retained inactive sessions', async () => { const first = renderHook(() => useStructuredAgentSessionRead({ diff --git a/src/shared/agent-session-journal-item-key.ts b/src/shared/agent-session-journal-item-key.ts index 8b6ed41e9dd..e53d3c5f15e 100644 --- a/src/shared/agent-session-journal-item-key.ts +++ b/src/shared/agent-session-journal-item-key.ts @@ -14,29 +14,72 @@ const KEY_DELIMITER = ':' export const MAX_JOURNAL_KEY_COMPONENT_CHARS = 1024 /** - * Deterministic stand-in for an oversized key component: same input, same - * output, so revisions and tombstones of one identity still share a key, and - * re-deriving from a parsed key is a fixed point (the bounded form is far - * below the cap). The head keeps keys debuggable; length plus two independent - * hashes makes an accidental collision practically impossible. Pure JS because - * clients derive keys too and cannot reach node:crypto. + * Deterministic stand-in for an oversized or ill-formed key component: same + * input, same output, so revisions and tombstones of one identity still share + * a key, and re-deriving from a parsed key is a fixed point (the bounded form + * is well-formed and far below the cap). The head keeps keys debuggable; + * length plus two independent hashes makes an accidental collision practically + * impossible. Pure JS because clients derive keys too and cannot reach + * node:crypto. + * + * JSON strings are arbitrary UTF-16 code units, so a component can carry a + * lone surrogate that `encodeURIComponent` throws on. Those values take the + * digest form too: the hashes run over the raw code units, so a value and its + * replacement-character spelling keep distinct keys. */ export function boundJournalKeyComponent(value: string): string { - if (value.length <= MAX_JOURNAL_KEY_COMPONENT_CHARS) { + if (value.length <= MAX_JOURNAL_KEY_COMPONENT_CHARS && !hasLoneSurrogate(value)) { return value } const h1 = fnv1a32(value, 0x811c9dc5).toString(16).padStart(8, '0') const h2 = fnv1a32(value, 0x0100_0193).toString(16).padStart(8, '0') - return `${codePointBoundedHead(value, 40)}~orca-oversized~${value.length}~${h1}${h2}` + return `${wellFormedBoundedHead(value, 40)}~orca-oversized~${value.length}~${h1}${h2}` } -/** `slice` cuts UTF-16 code units, so a head ending mid-astral-character would - * carry a lone high surrogate and make `encodeURIComponent` throw on a valid - * id. Dropping that unit keeps the head valid Unicode and deterministic. */ -function codePointBoundedHead(value: string, maxUnits: number): string { - const head = value.slice(0, maxUnits) - const last = head.charCodeAt(head.length - 1) - return last >= 0xd800 && last <= 0xdbff ? head.slice(0, -1) : head +/** The diagnostic head must be valid Unicode for `encodeURIComponent`: a pair + * split by the cut is dropped and a lone surrogate becomes U+FFFD — the + * hashes over the raw units keep the full key collision-safe regardless. + * Digest forms are persisted, so for well-formed input the head must stay + * byte-stable across builds or one identity would stop sharing a key. */ +function wellFormedBoundedHead(value: string, maxUnits: number): string { + let head = '' + let index = 0 + while (index < value.length && index < maxUnits) { + const unit = value.charCodeAt(index) + if (unit >= 0xd800 && unit <= 0xdbff) { + const next = index + 1 < value.length ? value.charCodeAt(index + 1) : 0 + if (next >= 0xdc00 && next <= 0xdfff) { + if (index + 1 >= maxUnits) { + break + } + head += value.charAt(index) + value.charAt(index + 1) + index += 2 + continue + } + head += '�' + index += 1 + continue + } + head += unit >= 0xdc00 && unit <= 0xdfff ? '�' : value.charAt(index) + index += 1 + } + return head +} + +function hasLoneSurrogate(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const unit = value.charCodeAt(index) + if (unit >= 0xd800 && unit <= 0xdbff) { + const next = index + 1 < value.length ? value.charCodeAt(index + 1) : 0 + if (next < 0xdc00 || next > 0xdfff) { + return true + } + index += 1 + } else if (unit >= 0xdc00 && unit <= 0xdfff) { + return true + } + } + return false } function fnv1a32(value: string, seed: number): number { @@ -97,7 +140,16 @@ export function agentJournalSubmissionKey(clientMessageId: string): string { * `encodeURIComponent` escapes the delimiter, so the split is unambiguous. */ export function parseAgentJournalItemKey(key: string): AgentJournalItemIdentity | null { - const parts = key.split(KEY_DELIMITER).map((part) => decodeURIComponent(part)) + // Persisted keys can be corrupted: a malformed percent sequence must fail + // the parse, never throw through journal replay or open. + const parts: string[] = [] + for (const part of key.split(KEY_DELIMITER)) { + try { + parts.push(decodeURIComponent(part)) + } catch { + return null + } + } const [provider, ...rest] = parts if (provider === 'codex' && rest.length === 3) { const ordinal = Number(rest[2])