From 1af297dae2757349bc318f5b7f0d594c29e0dd1d Mon Sep 17 00:00:00 2001 From: likun Date: Mon, 15 Jun 2026 15:53:06 +0800 Subject: [PATCH 1/2] Prefer RuntimeEvent read model for session reads --- .../runtime-event-read-model.test.ts | 82 ++- .../src/__tests__/session-manager.test.ts | 486 ++++++++++++++++++ .../runtime/src/runtime-event-read-model.ts | 83 ++- packages/runtime/src/session-manager.ts | 161 +++++- 4 files changed, 801 insertions(+), 11 deletions(-) diff --git a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts index 639540d162..b331347030 100644 --- a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts +++ b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts @@ -313,6 +313,39 @@ describe('projectRuntimeEventsToStoredMessages', () => { expect(out.diagnostics.map((diag) => diag.code)).toEqual(['partial_skipped']); }); + test('model thinking attaches to same-turn assistant text without breaking compatibility', () => { + const out = projectRuntimeEventsToStoredMessages([ + ev({ + id: 'evt-thinking', + ts: ts + 5, + role: 'model', + author: 'agent', + content: { kind: 'thinking', text: 'private reasoning', signature: 'sig-1' }, + }), + ev({ + id: 'evt-assistant', + ts: ts + 6, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'visible answer' }, + refs: { storedMessageId: 'legacy-assistant' }, + }), + ], { runHeaders: [header] }); + const legacy: StoredMessage[] = [{ + type: 'assistant', + id: 'legacy-assistant', + turnId, + ts: ts + 6, + text: 'visible answer', + modelId: 'claude-sonnet-4-5', + thinking: { text: 'private reasoning', signature: 'sig-1' }, + }]; + + expect(out.messages).toEqual(legacy); + expect(out.diagnostics).toEqual([]); + expect(compareRuntimeReadModelMessages(out.messages, legacy).compatible).toBe(true); + }); + test('unsupported and incomplete events are diagnostic-only', () => { const out = projectRuntimeEventsToStoredMessages([ ev({ @@ -345,12 +378,11 @@ describe('projectRuntimeEventsToStoredMessages', () => { expect(out.messages).toEqual([]); expect(out.diagnostics.map((diag) => diag.code)).toEqual([ - 'unsupported_event', - 'unsupported_event', 'incomplete_event', 'unsupported_event', 'incomplete_event', 'unsupported_event', + 'unsupported_event', ]); }); @@ -378,6 +410,52 @@ describe('projectRuntimeEventsToStoredMessages', () => { }]); expect(out.diagnostics).toEqual([]); }); + + test('aborted terminal RuntimeEvent preserves abort source from runtime state', () => { + const out = projectRuntimeEventsToStoredMessages([ + ev({ + id: 'evt-aborted', + ts: ts + 9, + status: 'aborted', + actions: { endInvocation: true, stateDelta: { abortSource: 'renderer.stop_button' } }, + }), + ], { + runHeaders: [{ ...header, status: 'cancelled' }], + }); + + expect(out.messages).toEqual([{ + type: 'turn_state', + id: 'evt-aborted', + turnId, + ts: ts + 9, + status: 'aborted', + parentTurnId: 'parent-turn', + abortedAt: ts + 9, + abortSource: 'renderer.stop_button', + partialOutputRetained: false, + }]); + expect(out.diagnostics).toEqual([]); + }); + + test('aborted terminal RuntimeEvent keeps an explicit diagnostic when abort source is unavailable', () => { + const out = projectRuntimeEventsToStoredMessages([ + ev({ + id: 'evt-aborted', + ts: ts + 9, + status: 'aborted', + actions: { endInvocation: true }, + }), + ], { + runHeaders: [{ ...header, status: 'cancelled' }], + }); + + expect(out.messages[0]).toMatchObject({ + type: 'turn_state', + status: 'aborted', + abortedAt: ts + 9, + }); + expect(out.diagnostics.map((diag) => diag.code)).toEqual(['incomplete_event']); + }); }); describe('compareRuntimeReadModelMessages', () => { diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index c644805197..3a89a72145 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -290,6 +290,305 @@ describe('SessionManager permission mode updates', () => { expect(await runStore.readRuntimeEvents(session.id, run.runId)).toEqual([]); }); + test('getMessages prefers RuntimeEvent-projected messages when legacy rows are present', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const seeded = await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'turn-1', + runId: 'run-1', + userText: 'runtime question', + assistantText: 'runtime answer', + legacyIdPrefix: 'legacy', + }); + + const messages = await manager.getMessages(session.id); + + expect(messages).toEqual(seeded.projectedMessages); + expect(JSON.stringify(messages.map((message) => message.id)) === JSON.stringify(seeded.legacyMessages.map((message) => message.id))).toBe(false); + }); + + test('getMessages falls back to legacy when projection is missing a legacy semantic row', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const legacyMessages: StoredMessage[] = [ + { type: 'user', id: 'legacy-user', turnId: 'turn-1', ts: 101, text: 'question' }, + { type: 'assistant', id: 'legacy-assistant', turnId: 'turn-1', ts: 102, text: 'legacy answer', modelId: 'fake-model' }, + { type: 'turn_state', id: 'legacy-state', turnId: 'turn-1', ts: 103, status: 'completed', partialOutputRetained: true }, + ]; + await store.appendMessages(session.id, legacyMessages); + await seedRuntimeRun(runStore, makeRunHeader({ + sessionId: session.id, + runId: 'run-1', + turnId: 'turn-1', + status: 'completed', + createdAt: 100, + updatedAt: 103, + completedAt: 103, + }), [ + runtimeEvent({ id: 'rt-user', sessionId: session.id, runId: 'run-1', turnId: 'turn-1', ts: 101, role: 'user', author: 'user', content: { kind: 'text', text: 'question' } }), + runtimeEvent({ id: 'rt-complete', sessionId: session.id, runId: 'run-1', turnId: 'turn-1', ts: 103, role: 'system', author: 'system', status: 'completed', actions: { endInvocation: true } }), + ]); + + expect(await manager.getMessages(session.id)).toEqual(legacyMessages); + }); + + test('getMessages falls back to legacy when a terminal run has no runtime ledger', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const legacyMessages: StoredMessage[] = [ + { type: 'user', id: 'legacy-user', turnId: 'turn-1', ts: 101, text: 'legacy only' }, + { type: 'turn_state', id: 'legacy-state', turnId: 'turn-1', ts: 102, status: 'completed', partialOutputRetained: false }, + ]; + await store.appendMessages(session.id, legacyMessages); + await runStore.createRun(makeRunHeader({ + sessionId: session.id, + runId: 'run-1', + turnId: 'turn-1', + status: 'completed', + completedAt: 102, + })); + + expect(await manager.getMessages(session.id)).toEqual(legacyMessages); + }); + + test('listTurns derives from the RuntimeEvent-primary message view', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'turn-1', + runId: 'run-1', + userText: 'runtime question', + assistantText: 'runtime answer', + legacyIdPrefix: 'legacy', + }); + store.failListTurnsFor.add(session.id); + + const turns = await manager.listTurns(session.id); + + expect(turns).toEqual([ + { + turnId: 'turn-1', + status: 'completed', + partialOutputRetained: true, + }, + ]); + }); + + test('mixed legacy-only system notes force legacy fallback instead of disappearing', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const seeded = await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'turn-1', + runId: 'run-1', + userText: 'question', + assistantText: 'answer', + legacyIdPrefix: 'legacy', + }); + const legacyNote: StoredMessage = { + type: 'system_note', + id: 'legacy-note', + ts: 104, + kind: 'mode_change', + data: { from: 'ask', to: 'execute' }, + }; + await store.appendMessage(session.id, legacyNote); + + const messages = await manager.getMessages(session.id); + + expect(messages).toEqual([...seeded.legacyMessages, legacyNote]); + }); + + test('retry finds aborted source turns and user messages through the RuntimeEvent-primary view', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + backends.register('fake', (ctx) => new EventBackend(ctx, [ + { type: 'complete', stopReason: 'end_turn' }, + ])); + const manager = new SessionManager({ store, runStore, backends, newId: nextId(), now: nextNow(6_760) }); + const session = await manager.createSession(makeInput()); + await seedRuntimeRun(runStore, makeRunHeader({ + sessionId: session.id, + runId: 'source-run', + turnId: 'source', + status: 'cancelled', + createdAt: 100, + updatedAt: 102, + completedAt: 102, + }), [ + runtimeEvent({ id: 'source-user', sessionId: session.id, runId: 'source-run', turnId: 'source', ts: 101, role: 'user', author: 'user', content: { kind: 'text', text: 'runtime retry text' } }), + runtimeEvent({ id: 'source-abort', sessionId: session.id, runId: 'source-run', turnId: 'source', ts: 102, role: 'system', author: 'system', status: 'aborted', actions: { endInvocation: true, stateDelta: { abortSource: 'renderer.stop_button' } } }), + ]); + store.failNextReadMessagesFor.set(session.id, 1); + + await drain(manager.retryTurn(session.id, { sourceTurnId: 'source', turnId: 'retry-1' })); + + const retryUser = (await store.readMessages(session.id)) + .find((message) => message.type === 'user' && message.turnId === 'retry-1'); + expect(retryUser?.type === 'user' ? retryUser.text : undefined).toBe('runtime retry text'); + }); + + test('regenerate finds completed source turns through the RuntimeEvent-primary view', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + backends.register('fake', (ctx) => new EventBackend(ctx, [ + { type: 'complete', stopReason: 'end_turn' }, + ])); + const manager = new SessionManager({ store, runStore, backends, newId: nextId(), now: nextNow(6_770) }); + const session = await manager.createSession(makeInput()); + await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'source', + runId: 'source-run', + userText: 'runtime regenerate text', + assistantText: 'runtime answer', + legacyIdPrefix: 'legacy', + }); + store.failNextReadMessagesFor.set(session.id, 1); + + await drain(manager.regenerateTurn(session.id, { sourceTurnId: 'source', turnId: 'regen-1' })); + + const messages = await store.readMessages(session.id); + const regenUser = messages.find((message) => message.type === 'user' && message.turnId === 'regen-1'); + expect(regenUser?.type === 'user' ? regenUser.text : undefined).toBe('runtime regenerate text'); + const regenState = deriveTurnRecords(messages).find((turn) => turn.turnId === 'regen-1'); + expect(regenState?.regeneratedFromTurnId).toBe('source'); + }); + + test('branchFromTurn copies through the RuntimeEvent-primary message boundary', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput({ name: 'Parent' })); + await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'source', + runId: 'source-run', + userText: 'runtime branch context', + assistantText: 'runtime branch answer', + legacyIdPrefix: 'legacy', + }); + store.failNextReadMessagesFor.set(session.id, 1); + + const child = await manager.branchFromTurn(session.id, { sourceTurnId: 'source', name: 'Child' }); + + const childMessages = await store.readMessages(child.id); + expect(childMessages[0]).toMatchObject({ type: 'user', turnId: 'source', text: 'runtime branch context' }); + expect(childMessages[1]).toMatchObject({ type: 'assistant', turnId: 'source', text: 'runtime branch answer' }); + expect(childMessages[2]).toMatchObject({ type: 'system_note', kind: 'session_start' }); + expect(childMessages.some((message) => message.type === 'turn_state')).toBe(false); + }); + + test('multi-run RuntimeEvent projection preserves retry regenerate and branch lineage on turns', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'root', + runId: 'root-run', + userText: 'root question', + assistantText: 'root answer', + legacyIdPrefix: 'root-legacy', + }); + await seedRuntimeReadTurnWithHeader({ + store, + runStore, + sessionId: session.id, + turnId: 'retry', + runId: 'retry-run', + userText: 'retry question', + assistantText: 'retry answer', + legacyIdPrefix: 'retry-legacy', + header: { parentTurnId: 'root', retriedFromTurnId: 'root' }, + tsBase: 200, + }); + await seedRuntimeReadTurnWithHeader({ + store, + runStore, + sessionId: session.id, + turnId: 'regen', + runId: 'regen-run', + userText: 'regen question', + assistantText: 'regen answer', + legacyIdPrefix: 'regen-legacy', + header: { parentTurnId: 'root', regeneratedFromTurnId: 'root' }, + tsBase: 300, + }); + await seedRuntimeReadTurnWithHeader({ + store, + runStore, + sessionId: session.id, + turnId: 'branch', + runId: 'branch-run', + userText: 'branch question', + assistantText: 'branch answer', + legacyIdPrefix: 'branch-legacy', + header: { parentSessionId: 'parent-session', branchOfTurnId: 'root' }, + tsBase: 400, + }); + store.failNextReadMessagesFor.set(session.id, 1); + + const turns = await manager.listTurns(session.id); + + expect(turns.find((turn) => turn.turnId === 'retry')).toMatchObject({ + status: 'completed', + parentTurnId: 'root', + retriedFromTurnId: 'root', + }); + expect(turns.find((turn) => turn.turnId === 'regen')).toMatchObject({ + status: 'completed', + parentTurnId: 'root', + regeneratedFromTurnId: 'root', + }); + expect(turns.find((turn) => turn.turnId === 'branch')).toMatchObject({ + status: 'completed', + parentSessionId: 'parent-session', + branchOfTurnId: 'root', + }); + }); + + test('getMessages keeps legacy SessionStore behavior when no runStore is provided', async () => { + const store = new MemorySessionStore(); + const backends = new BackendRegistry(); + backends.register('fake', (ctx) => new TestBackend(ctx)); + const manager = new SessionManager({ store, backends, newId: nextId(), now: nextNow(6_760) }); + const session = await manager.createSession(makeInput()); + const legacyMessages: StoredMessage[] = [ + { type: 'user', id: 'legacy-user', turnId: 'turn-1', ts: 101, text: 'legacy only' }, + ]; + await store.appendMessages(session.id, legacyMessages); + + expect(await manager.getMessages(session.id)).toEqual(legacyMessages); + }); + test('next turn receives complete prior RuntimeEvent context alongside legacy context', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); @@ -1112,6 +1411,8 @@ class MemorySessionStore implements SessionStore { private headers = new Map(); private messages = new Map(); readonly failReadMessagesFor = new Set(); + readonly failNextReadMessagesFor = new Map(); + readonly failListTurnsFor = new Set(); disposeCount = 0; async create(input: CreateSessionInput): Promise { @@ -1154,11 +1455,18 @@ class MemorySessionStore implements SessionStore { } async readMessages(sessionId: string): Promise { + const remainingFailures = this.failNextReadMessagesFor.get(sessionId) ?? 0; + if (remainingFailures > 0) { + if (remainingFailures === 1) this.failNextReadMessagesFor.delete(sessionId); + else this.failNextReadMessagesFor.set(sessionId, remainingFailures - 1); + throw new Error(`Cannot read messages for ${sessionId}`); + } if (this.failReadMessagesFor.has(sessionId)) throw new Error(`Cannot read messages for ${sessionId}`); return [...(this.messages.get(sessionId) ?? [])]; } async listTurns(sessionId: string): Promise { + if (this.failListTurnsFor.has(sessionId)) throw new Error(`Cannot list turns for ${sessionId}`); return deriveTurnRecords(await this.readMessages(sessionId)); } @@ -1306,6 +1614,158 @@ function makeRunEvent(overrides: Partial = {}): AgentRunEvent { }; } +function makeManagerForReadCutover(store: MemorySessionStore, runStore: AgentRunStore): SessionManager { + const backends = new BackendRegistry(); + backends.register('fake', (ctx) => new TestBackend(ctx)); + return new SessionManager({ store, runStore, backends, newId: nextId(), now: nextNow(6_755) }); +} + +async function seedRuntimeReadTurn(input: { + store: MemorySessionStore; + runStore: AgentRunStore; + sessionId: string; + turnId: string; + runId: string; + userText: string; + assistantText: string; + legacyIdPrefix: string; +}): Promise<{ legacyMessages: StoredMessage[]; projectedMessages: StoredMessage[] }> { + const header = makeRunHeader({ + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + status: 'completed', + createdAt: 100, + updatedAt: 103, + completedAt: 103, + }); + const events = [ + runtimeEvent({ + id: `${input.runId}-user-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: 101, + role: 'user', + author: 'user', + content: { kind: 'text', text: input.userText }, + refs: { storedMessageId: `${input.runId}-projected-user` }, + }), + runtimeEvent({ + id: `${input.runId}-assistant-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: 102, + role: 'model', + author: 'agent', + content: { kind: 'text', text: input.assistantText }, + refs: { storedMessageId: `${input.runId}-projected-assistant` }, + }), + runtimeEvent({ + id: `${input.runId}-complete-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: 103, + role: 'system', + author: 'system', + status: 'completed', + actions: { endInvocation: true }, + }), + ]; + const legacyMessages: StoredMessage[] = [ + { type: 'user', id: `${input.legacyIdPrefix}-user`, turnId: input.turnId, ts: 101, text: input.userText }, + { type: 'assistant', id: `${input.legacyIdPrefix}-assistant`, turnId: input.turnId, ts: 102, text: input.assistantText, modelId: 'fake-model' }, + { type: 'turn_state', id: `${input.legacyIdPrefix}-state`, turnId: input.turnId, ts: 103, status: 'completed', partialOutputRetained: true }, + ]; + const projectedMessages: StoredMessage[] = [ + { type: 'user', id: `${input.runId}-projected-user`, turnId: input.turnId, ts: 101, text: input.userText }, + { type: 'assistant', id: `${input.runId}-projected-assistant`, turnId: input.turnId, ts: 102, text: input.assistantText, modelId: 'fake-model' }, + { type: 'turn_state', id: `${input.runId}-complete-event`, turnId: input.turnId, ts: 103, status: 'completed', partialOutputRetained: true }, + ]; + await input.store.appendMessages(input.sessionId, legacyMessages); + await seedRuntimeRun(input.runStore, header, events); + return { legacyMessages, projectedMessages }; +} + +async function seedRuntimeReadTurnWithHeader(input: { + store: MemorySessionStore; + runStore: AgentRunStore; + sessionId: string; + turnId: string; + runId: string; + userText: string; + assistantText: string; + legacyIdPrefix: string; + header: Partial; + tsBase: number; +}): Promise { + const header = makeRunHeader({ + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + status: 'completed', + createdAt: input.tsBase, + updatedAt: input.tsBase + 3, + completedAt: input.tsBase + 3, + ...input.header, + }); + const events = [ + runtimeEvent({ + id: `${input.runId}-user-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: input.tsBase + 1, + role: 'user', + author: 'user', + content: { kind: 'text', text: input.userText }, + refs: { storedMessageId: `${input.runId}-projected-user` }, + }), + runtimeEvent({ + id: `${input.runId}-assistant-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: input.tsBase + 2, + role: 'model', + author: 'agent', + content: { kind: 'text', text: input.assistantText }, + refs: { storedMessageId: `${input.runId}-projected-assistant` }, + }), + runtimeEvent({ + id: `${input.runId}-complete-event`, + sessionId: input.sessionId, + runId: input.runId, + turnId: input.turnId, + ts: input.tsBase + 3, + role: 'system', + author: 'system', + status: 'completed', + actions: { endInvocation: true }, + }), + ]; + await input.store.appendMessages(input.sessionId, [ + { type: 'user', id: `${input.legacyIdPrefix}-user`, turnId: input.turnId, ts: input.tsBase + 1, text: input.userText }, + { type: 'assistant', id: `${input.legacyIdPrefix}-assistant`, turnId: input.turnId, ts: input.tsBase + 2, text: input.assistantText, modelId: 'fake-model' }, + { + type: 'turn_state', + id: `${input.legacyIdPrefix}-state`, + turnId: input.turnId, + ts: input.tsBase + 3, + status: 'completed', + ...(input.header.parentTurnId ? { parentTurnId: input.header.parentTurnId } : {}), + ...(input.header.retriedFromTurnId ? { retriedFromTurnId: input.header.retriedFromTurnId } : {}), + ...(input.header.regeneratedFromTurnId ? { regeneratedFromTurnId: input.header.regeneratedFromTurnId } : {}), + ...(input.header.branchOfTurnId ? { branchOfTurnId: input.header.branchOfTurnId } : {}), + ...(input.header.parentSessionId ? { parentSessionId: input.header.parentSessionId } : {}), + partialOutputRetained: true, + }, + ]); + await seedRuntimeRun(input.runStore, header, events); +} + async function seedRun( runStore: AgentRunStore, header: AgentRunHeader, @@ -1317,6 +1777,32 @@ async function seedRun( } } +async function seedRuntimeRun( + runStore: AgentRunStore, + header: AgentRunHeader, + events: RuntimeEvent[], +): Promise { + await runStore.createRun(header); + for (const event of events) { + await runStore.appendRuntimeEvent(header.sessionId, header.runId, event); + } +} + +function runtimeEvent(overrides: Partial): RuntimeEvent { + return { + id: 'rt-event', + invocationId: 'inv-1', + runId: 'run-1', + sessionId: 'session-1', + turnId: 'turn-1', + ts: 100, + partial: false, + role: 'system', + author: 'system', + ...overrides, + }; +} + async function seedRunningTurn(store: MemorySessionStore, sessionId: string, turnId: string): Promise { await store.appendMessages(sessionId, [ { type: 'user', id: `${turnId}-user`, turnId, ts: 9, text: 'interrupted turn' }, diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts index e8d7d03b7e..7221c6e668 100644 --- a/packages/runtime/src/runtime-event-read-model.ts +++ b/packages/runtime/src/runtime-event-read-model.ts @@ -54,6 +54,13 @@ interface ProjectionState { toolName: string; hint?: string; }>; + thinkingByTurn: Map; +} + +interface PendingThinking { + event: RuntimeEvent; + text: string; + signature?: string; } export function projectRuntimeEventsToStoredMessages( @@ -65,6 +72,7 @@ export function projectRuntimeEventsToStoredMessages( diagnostics: [], toolNameByUseId: new Map(), permissionRequestById: new Map(), + thinkingByTurn: new Map(), }; const messages: StoredMessage[] = []; @@ -88,7 +96,7 @@ export function projectRuntimeEventsToStoredMessages( projected = projectFunctionResponse(event, state, messages) || projected; break; case 'thinking': - diagnostic(state, event, 'unsupported_event', 'thinking content has no legacy read-model row'); + projected = projectThinking(event, state, messages) || projected; break; case 'error': if (!isTerminalRuntimeEvent(event)) { @@ -131,6 +139,10 @@ export function projectRuntimeEventsToStoredMessages( } } + for (const pending of state.thinkingByTurn.values()) { + diagnostic(state, pending.event, 'unsupported_event', 'thinking content has no same-turn assistant text row'); + } + return { messages, diagnostics: state.diagnostics }; } @@ -201,6 +213,7 @@ function projectText( text: event.content.text, modelId: header.modelId, }); + attachPendingThinking(event, state, messages); return true; } @@ -208,6 +221,22 @@ function projectText( return false; } +function projectThinking( + event: RuntimeEvent, + state: ProjectionState, + messages: StoredMessage[], +): boolean { + if (event.content?.kind !== 'thinking') return false; + const pending: PendingThinking = { + event, + text: event.content.text, + ...(event.content.signature !== undefined ? { signature: event.content.signature } : {}), + }; + if (attachThinkingToAssistant(event, pending, messages)) return true; + state.thinkingByTurn.set(thinkingKey(event), pending); + return true; +} + function projectFunctionCall( event: RuntimeEvent, state: ProjectionState, @@ -348,6 +377,7 @@ function projectTerminalTurnState( diagnostic(state, event, 'incomplete_event', 'terminal RuntimeEvent status cannot be mapped to a legacy TurnStatus'); return false; } + const abortSource = status === 'aborted' ? abortSourceFromRuntime(event, header) : undefined; const partialOutputRetained = messages.some((message) => message.turnId === event.turnId && ((message.type === 'assistant' && message.text.trim().length > 0) || message.type === 'tool_result') @@ -364,18 +394,67 @@ function projectTerminalTurnState( ...(header.branchOfTurnId ? { branchOfTurnId: header.branchOfTurnId } : {}), ...(header.parentSessionId ? { parentSessionId: header.parentSessionId } : {}), ...(status === 'aborted' ? { abortedAt: event.ts } : {}), + ...(abortSource ? { abortSource } : {}), ...(status === 'failed' ? { errorClass: header.failureClass ?? 'unknown' } : {}), partialOutputRetained, }); if (status === 'failed' && !header.failureClass) { diagnostic(state, event, 'incomplete_event', 'failed terminal event did not carry an exact AgentRunHeader.failureClass'); } - if (status === 'aborted') { + if (status === 'aborted' && !abortSource) { diagnostic(state, event, 'incomplete_event', 'abortSource is not present in RuntimeEvent or AgentRunHeader metadata'); } return true; } +function attachPendingThinking( + event: RuntimeEvent, + state: ProjectionState, + messages: StoredMessage[], +): void { + const key = thinkingKey(event); + const pending = state.thinkingByTurn.get(key); + if (!pending) return; + if (attachThinkingToAssistant(event, pending, messages)) { + state.thinkingByTurn.delete(key); + } +} + +function attachThinkingToAssistant( + event: RuntimeEvent, + pending: PendingThinking, + messages: StoredMessage[], +): boolean { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]!; + if (message.type !== 'assistant' || message.turnId !== event.turnId) continue; + message.thinking = { + text: pending.text, + ...(pending.signature !== undefined ? { signature: pending.signature } : {}), + }; + return true; + } + return false; +} + +function thinkingKey(event: RuntimeEvent): string { + return `${event.runId}:${event.turnId}`; +} + +function abortSourceFromRuntime(event: RuntimeEvent, header: AgentRunHeader): string | undefined { + return stringStateDelta(event, 'abortSource') + ?? stringStateDelta(event, 'source') + ?? stringRecordValue(event.refs, 'abortSource') + ?? stringRecordValue(event.refs, 'source') + ?? stringRecordValue(header as unknown as Record, 'abortSource'); +} + +function stringRecordValue(value: unknown, key: string): string | undefined { + if (!value || typeof value !== 'object') return undefined; + const result = (value as Record)[key]; + return typeof result === 'string' && result.length > 0 ? result : undefined; +} + function stableMessageId( event: RuntimeEvent, state: ProjectionState, diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 37820b8e5f..ea1946735f 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -45,8 +45,13 @@ import type { } from '@maka/core/runtime-inputs'; import type { PermissionResponse } from '@maka/core/permission'; import type { PermissionMode } from '@maka/core/permission'; -import { DEEP_RESEARCH_SESSION_LABEL, isDeepResearchSession } from '@maka/core'; -import type { AgentRunStore } from '@maka/core'; +import { DEEP_RESEARCH_SESSION_LABEL, deriveTurnRecords, isDeepResearchSession, isTerminalRuntimeEvent } from '@maka/core'; +import type { AgentRunHeader, AgentRunStore, RuntimeEvent } from '@maka/core'; +import { + compareRuntimeReadModelMessages, + projectRuntimeEventsToStoredMessages, + type RuntimeEventReadModelDiagnostic, +} from './runtime-event-read-model.js'; import type { AgentBackend } from './ai-sdk-backend.js'; import type { RunTraceRecorder } from './run-trace.js'; @@ -138,6 +143,15 @@ interface ActiveSession extends AgentRunActiveSession { turnToRunId: Map; } +type SessionReadSource = 'runtime_events' | 'legacy_session_store'; + +interface SessionReadView { + source: SessionReadSource; + messages: StoredMessage[]; + turns: TurnRecord[]; + diagnostics: RuntimeEventReadModelDiagnostic[]; +} + export class SessionManager { private readonly active = new Map(); @@ -157,11 +171,11 @@ export class SessionManager { } async getMessages(sessionId: string): Promise { - return this.deps.store.readMessages(sessionId); + return (await this.readSessionView(sessionId)).messages; } async listTurns(sessionId: string): Promise { - return this.deps.store.listTurns(sessionId); + return (await this.readSessionView(sessionId)).turns; } async recoverInterruptedSessions(): Promise { @@ -486,7 +500,7 @@ export class SessionManager { input: BranchFromTurnInput, ): Promise { const header = await this.deps.store.readHeader(sessionId); - const messages = await this.deps.store.readMessages(sessionId); + const { messages } = await this.readSessionView(sessionId); const copied = copyMessagesThroughTurnBoundary(messages, input.sourceTurnId); if (copied.length === 0) throw new Error(`Cannot branch from unknown turn ${input.sourceTurnId}`); const next = await this.deps.store.create({ @@ -639,7 +653,7 @@ export class SessionManager { allowed: readonly TurnRecord['status'][], action: string, ): Promise { - const turn = (await this.deps.store.listTurns(sessionId)).find((candidate) => candidate.turnId === turnId); + const turn = (await this.readSessionView(sessionId)).turns.find((candidate) => candidate.turnId === turnId); if (!turn) throw new Error(`Cannot ${action}: unknown turn ${turnId}`); if (!allowed.includes(turn.status)) { throw new Error(`Cannot ${action}: turn ${turnId} is ${turn.status}`); @@ -648,12 +662,109 @@ export class SessionManager { } private async requireUserMessageForTurn(sessionId: string, turnId: string): Promise { - const user = (await this.deps.store.readMessages(sessionId)) + const user = (await this.readSessionView(sessionId)).messages .find((message): message is UserMessage => message.type === 'user' && message.turnId === turnId); if (!user) throw new Error(`Turn ${turnId} has no user message`); return user; } + private async readSessionView(sessionId: string): Promise { + const legacy = await this.readLegacyMessages(sessionId); + const fallback = (diagnostics: RuntimeEventReadModelDiagnostic[] = []): SessionReadView => { + if (!legacy.readable) throw legacy.error; + return { + source: 'legacy_session_store', + messages: legacy.messages, + turns: deriveTurnRecords(legacy.messages), + diagnostics, + }; + }; + + if (runtimeReadSourceForcedLegacy()) { + return fallback([readDiagnostic('unsupported_event', 'RuntimeEvent read projection disabled by MAKA_RUNTIME_READ_SOURCE')]); + } + if (!this.deps.runStore) { + return fallback(); + } + + let runs: AgentRunHeader[]; + try { + runs = await this.deps.runStore.listSessionRuns(sessionId); + } catch { + return fallback([readDiagnostic('unsupported_event', 'AgentRunStore.listSessionRuns failed')]); + } + if (runs.length === 0) { + return fallback(); + } + + const events: RuntimeEvent[] = []; + const ordered: Array<{ event: RuntimeEvent; runIndex: number; eventIndex: number }> = []; + for (let runIndex = 0; runIndex < runs.length; runIndex += 1) { + const run = runs[runIndex]!; + if (!isTerminalRunStatus(run.status)) { + return fallback([readDiagnostic('incomplete_event', 'RuntimeEvent read projection is not used for active runs', run)]); + } + + let runEvents: RuntimeEvent[]; + try { + runEvents = await this.deps.runStore.readRuntimeEvents(sessionId, run.runId); + } catch { + return fallback([readDiagnostic('unsupported_event', 'AgentRunStore.readRuntimeEvents failed', { runId: run.runId })]); + } + + if (runEvents.length === 0) { + return fallback([readDiagnostic('incomplete_event', 'terminal run has no readable RuntimeEvent ledger', { runId: run.runId })]); + } + if (!runEvents.some(isTerminalRuntimeEvent)) { + return fallback([readDiagnostic('incomplete_event', 'terminal run has no terminal RuntimeEvent', { runId: run.runId })]); + } + + for (let eventIndex = 0; eventIndex < runEvents.length; eventIndex += 1) { + ordered.push({ event: runEvents[eventIndex]!, runIndex, eventIndex }); + } + } + + ordered.sort((a, b) => + a.event.ts - b.event.ts || + a.runIndex - b.runIndex || + a.eventIndex - b.eventIndex || + a.event.id.localeCompare(b.event.id) + ); + for (const item of ordered) events.push(item.event); + + const projected = projectRuntimeEventsToStoredMessages(events, { runHeaders: runs }); + if (hasHardProjectionDiagnostic(projected.diagnostics)) { + return fallback(projected.diagnostics); + } + + const diagnostics = [...projected.diagnostics]; + if (legacy.readable) { + const compatibility = compareRuntimeReadModelMessages(projected.messages, legacy.messages); + diagnostics.push(...compatibility.diagnostics); + if (hasHardCompatibilityDiagnostic(compatibility.diagnostics)) { + return fallback(diagnostics); + } + } + + return { + source: 'runtime_events', + messages: projected.messages, + turns: deriveTurnRecords(projected.messages), + diagnostics, + }; + } + + private async readLegacyMessages(sessionId: string): Promise< + | { readable: true; messages: StoredMessage[] } + | { readable: false; error: unknown } + > { + try { + return { readable: true, messages: await this.deps.store.readMessages(sessionId) }; + } catch (error) { + return { readable: false, error }; + } + } + private async recoverAgentRunsFromLedger( sessionId: string, messages: readonly StoredMessage[], @@ -873,6 +984,42 @@ function copyMessagesThroughTurnBoundary(messages: readonly StoredMessage[], tur .filter((message) => message.type !== 'turn_state'); } +function isTerminalRunStatus(status: AgentRunHeader['status']): boolean { + return status === 'completed' || status === 'failed' || status === 'cancelled'; +} + +function hasHardProjectionDiagnostic(diagnostics: readonly RuntimeEventReadModelDiagnostic[]): boolean { + return diagnostics.some((diagnostic) => + diagnostic.code === 'incomplete_event' || + diagnostic.code === 'unsupported_event' || + diagnostic.code === 'tool_use_id_mismatch' + ); +} + +function hasHardCompatibilityDiagnostic(diagnostics: readonly RuntimeEventReadModelDiagnostic[]): boolean { + // A projected view missing a legacy semantic row means RuntimeEvents cannot + // yet render the public session faithfully. Extra projected rows are allowed: + // for compatible runtime-ledger sessions they represent RuntimeEvents being + // fresher than stale legacy rows, while diagnostics keep that mismatch visible. + return diagnostics.some((diagnostic) => diagnostic.code === 'missing_legacy_message'); +} + +function readDiagnostic( + code: RuntimeEventReadModelDiagnostic['code'], + message: string, + detail?: unknown, +): RuntimeEventReadModelDiagnostic { + return { + code, + message, + ...(detail !== undefined ? { detail } : {}), + }; +} + +function runtimeReadSourceForcedLegacy(): boolean { + return process.env.MAKA_RUNTIME_READ_SOURCE === 'legacy'; +} + function blockedReasonFromErrorReason(reason: string | undefined): SessionBlockedReason { if (!reason) return 'unknown'; if (reason === 'permission_required') return 'permission_required'; From 6a27af44d7a02cca46ed5462cd314154db877f4d Mon Sep 17 00:00:00 2001 From: likun Date: Mon, 15 Jun 2026 16:06:35 +0800 Subject: [PATCH 2/2] Harden RuntimeEvent read cutover --- .../src/__tests__/session-manager.test.ts | 153 +++++++++++++++++- packages/runtime/src/session-manager.ts | 13 +- .../src/__tests__/agent-run-store.test.ts | 21 ++- packages/storage/src/agent-run-store.ts | 4 +- 4 files changed, 185 insertions(+), 6 deletions(-) diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 3a89a72145..39245afe17 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -360,6 +360,51 @@ describe('SessionManager permission mode updates', () => { expect(await manager.getMessages(session.id)).toEqual(legacyMessages); }); + test('getMessages falls back to legacy when runtime ledger read fails', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore({ failRuntimeEventReads: true }); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const seeded = await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'turn-1', + runId: 'run-1', + userText: 'legacy question', + assistantText: 'legacy answer', + legacyIdPrefix: 'legacy', + }); + + expect(await manager.getMessages(session.id)).toEqual(seeded.legacyMessages); + }); + + test('MAKA_RUNTIME_READ_SOURCE=legacy forces legacy reads even when RuntimeEvents are complete', async () => { + const previous = process.env.MAKA_RUNTIME_READ_SOURCE; + process.env.MAKA_RUNTIME_READ_SOURCE = 'legacy'; + try { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + const seeded = await seedRuntimeReadTurn({ + store, + runStore, + sessionId: session.id, + turnId: 'turn-1', + runId: 'run-1', + userText: 'legacy forced question', + assistantText: 'legacy forced answer', + legacyIdPrefix: 'legacy', + }); + + expect(await manager.getMessages(session.id)).toEqual(seeded.legacyMessages); + } finally { + if (previous === undefined) delete process.env.MAKA_RUNTIME_READ_SOURCE; + else process.env.MAKA_RUNTIME_READ_SOURCE = previous; + } + }); + test('listTurns derives from the RuntimeEvent-primary message view', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); @@ -417,6 +462,111 @@ describe('SessionManager permission mode updates', () => { expect(messages).toEqual([...seeded.legacyMessages, legacyNote]); }); + test('getMessages orders RuntimeEvent-primary reads by session event chronology across runs', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const manager = makeManagerForReadCutover(store, runStore); + const session = await manager.createSession(makeInput()); + await seedRuntimeRun(runStore, makeRunHeader({ + sessionId: session.id, + runId: 'slow-run', + turnId: 'slow', + status: 'completed', + createdAt: 100, + updatedAt: 107, + completedAt: 107, + }), [ + runtimeEvent({ + id: 'slow-user', + sessionId: session.id, + runId: 'slow-run', + turnId: 'slow', + ts: 101, + role: 'user', + author: 'user', + content: { kind: 'text', text: 'slow question' }, + refs: { storedMessageId: 'slow-user-message' }, + }), + runtimeEvent({ + id: 'slow-assistant', + sessionId: session.id, + runId: 'slow-run', + turnId: 'slow', + ts: 106, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'slow answer' }, + refs: { storedMessageId: 'slow-assistant-message' }, + }), + runtimeEvent({ + id: 'slow-complete', + sessionId: session.id, + runId: 'slow-run', + turnId: 'slow', + ts: 107, + role: 'system', + author: 'system', + status: 'completed', + actions: { endInvocation: true }, + }), + ]); + await seedRuntimeRun(runStore, makeRunHeader({ + sessionId: session.id, + runId: 'fast-run', + turnId: 'fast', + status: 'completed', + createdAt: 102, + updatedAt: 105, + completedAt: 105, + }), [ + runtimeEvent({ + id: 'fast-user', + sessionId: session.id, + runId: 'fast-run', + turnId: 'fast', + ts: 103, + role: 'user', + author: 'user', + content: { kind: 'text', text: 'fast question' }, + refs: { storedMessageId: 'fast-user-message' }, + }), + runtimeEvent({ + id: 'fast-assistant', + sessionId: session.id, + runId: 'fast-run', + turnId: 'fast', + ts: 104, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'fast answer' }, + refs: { storedMessageId: 'fast-assistant-message' }, + }), + runtimeEvent({ + id: 'fast-complete', + sessionId: session.id, + runId: 'fast-run', + turnId: 'fast', + ts: 105, + role: 'system', + author: 'system', + status: 'completed', + actions: { endInvocation: true }, + }), + ]); + store.failNextReadMessagesFor.set(session.id, 1); + + const messages = await manager.getMessages(session.id); + + expect(messages.map((message) => `${message.type}:${'turnId' in message ? message.turnId : 'none'}:${message.ts}`)).toEqual([ + 'user:slow:101', + 'user:fast:103', + 'assistant:fast:104', + 'turn_state:fast:105', + 'assistant:slow:106', + 'turn_state:slow:107', + ]); + }); + test('retry finds aborted source turns and user messages through the RuntimeEvent-primary view', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); @@ -1512,7 +1662,7 @@ class MemoryAgentRunStore implements AgentRunStore { private events = new Map(); private runtimeEvents = new Map(); - constructor(private readonly options: { failRuntimeEventAppends?: boolean } = {}) {} + constructor(private readonly options: { failRuntimeEventAppends?: boolean; failRuntimeEventReads?: boolean } = {}) {} async createRun(header: AgentRunHeader): Promise { this.headers.set(key(header.sessionId, header.runId), { ...header }); @@ -1555,6 +1705,7 @@ class MemoryAgentRunStore implements AgentRunStore { } async readRuntimeEvents(sessionId: string, runId: string): Promise { + if (this.options.failRuntimeEventReads) throw new Error('runtime event read failed'); return (this.runtimeEvents.get(key(sessionId, runId)) ?? []).map(copyRuntimeEvent); } } diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index ea1946735f..f0430d96e0 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -708,8 +708,13 @@ export class SessionManager { let runEvents: RuntimeEvent[]; try { runEvents = await this.deps.runStore.readRuntimeEvents(sessionId, run.runId); - } catch { - return fallback([readDiagnostic('unsupported_event', 'AgentRunStore.readRuntimeEvents failed', { runId: run.runId })]); + } catch (error) { + return fallback([ + readDiagnostic('unsupported_event', 'AgentRunStore.readRuntimeEvents failed', { + runId: run.runId, + error: error instanceof Error ? error.message : String(error), + }), + ]); } if (runEvents.length === 0) { @@ -724,6 +729,10 @@ export class SessionManager { } } + // RuntimeEvent-primary reads use session chronology across terminal runs, + // with stable run/ledger/id tie-breakers. This matches legacy message + // reads better than concatenating each run ledger in creation order when + // overlapping runs complete out of order. ordered.sort((a, b) => a.event.ts - b.event.ts || a.runIndex - b.runIndex || diff --git a/packages/storage/src/__tests__/agent-run-store.test.ts b/packages/storage/src/__tests__/agent-run-store.test.ts index 42a7ce76b9..6f314221b9 100644 --- a/packages/storage/src/__tests__/agent-run-store.test.ts +++ b/packages/storage/src/__tests__/agent-run-store.test.ts @@ -113,7 +113,7 @@ describe('AgentRunStore', () => { }); }); - it('skips corrupt runtime event lines and ignores a partial corrupt tail', async () => { + it('rejects durable corrupt runtime event lines instead of shortening the canonical ledger', async () => { await withStore(async (store, root) => { await store.createRun(makeHeader()); const runtimeEventsPath = join(root, 'sessions', 'session-1', 'runs', 'run-1', 'runtime-events.jsonl'); @@ -122,11 +122,28 @@ describe('AgentRunStore', () => { JSON.stringify(makeRuntimeEvent({ id: 'runtime-1' })) + '\n{"id":"corrupt"\n' + JSON.stringify(makeRuntimeEvent({ id: 'runtime-2' })) + + '\n', + ); + + await assert.rejects( + () => store.readRuntimeEvents('session-1', 'run-1'), + /Invalid RuntimeEvent JSONL line 2 for run run-1/, + ); + }); + }); + + it('ignores an unterminated partial runtime event tail', async () => { + await withStore(async (store, root) => { + await store.createRun(makeHeader()); + const runtimeEventsPath = join(root, 'sessions', 'session-1', 'runs', 'run-1', 'runtime-events.jsonl'); + await writeFile( + runtimeEventsPath, + JSON.stringify(makeRuntimeEvent({ id: 'runtime-1' })) + '\n{"id":"partial"', ); const events = await store.readRuntimeEvents('session-1', 'run-1'); - assert.deepEqual(events.map((event) => event.id), ['runtime-1', 'runtime-2']); + assert.deepEqual(events.map((event) => event.id), ['runtime-1']); }); }); }); diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index e97dc0da25..6b06315ee7 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -140,8 +140,10 @@ class FileAgentRunStore implements AgentRunStore { for (const entry of lines) { try { events.push(JSON.parse(entry.line) as RuntimeEvent); - } catch { + } catch (error) { if (!endsWithNewline && entry.lineNumber === lastLineNumber) continue; + const message = error instanceof Error ? error.message : 'Invalid JSON'; + throw new Error(`Invalid RuntimeEvent JSONL line ${entry.lineNumber} for run ${runId}: ${message}`); } } return events;