From bf95515e49d3d8e40b1404ff3f0ef36e19b45572 Mon Sep 17 00:00:00 2001 From: Yang Wu <1425326620@qq.com> Date: Wed, 23 Sep 2026 00:12:18 +0800 Subject: [PATCH] fix(runtime-host): preserve transcript projection diagnostics Refs #5572 Generated-by: Codex --- .../session-transcript-reader.test.ts | 135 ++++++++++++++++++ .../src/server/session-transcript-reader.ts | 23 ++- 2 files changed, 156 insertions(+), 2 deletions(-) diff --git a/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts b/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts index f9d0886628..837f4d0314 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts @@ -33,6 +33,7 @@ import { type StoredMessage, } from '@maka/core/session'; import { projectRuntimeEventsToStoredMessages } from '@maka/runtime/runtime-event-read-model'; +import { RuntimeReadModelError } from '@maka/runtime/runtime-read-model'; import { encodeDurableToolResultOutput } from '@maka/runtime/durable-tool-result-projection'; import { shapeTerminalResult } from '@maka/runtime/shell-tools'; import { createLedgerArchiveResourceReader } from '@maka/runtime/ledger-tool-result-archive-reader'; @@ -43,6 +44,7 @@ import { foldTurnContribution } from '@maka/storage/session-message-projection'; import type { SessionTurnContribution } from '@maka/storage/execution-stores'; import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import { boundedFailureDiagnostic } from '../server/failure-diagnostic.js'; import { createSessionTranscriptReader, TRANSCRIPT_TURN_MAX_BYTES, @@ -62,8 +64,12 @@ for (const coordination of [false, true]) const owner = await tryAcquireInteractiveRootOwner(capability); assert.ok(owner); if (!owner) assert.fail('expected the interactive root owner'); + let openedStores: + | Awaited> + | undefined; try { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + openedStores = stores; const input = { cwd: capability.canonicalPath, llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', @@ -424,6 +430,7 @@ for (const coordination of [false, true]) ); } } finally { + await openedStores?.sessionStore.close?.(); await owner.close(); await rm(base, { recursive: true, force: true }); } @@ -434,8 +441,10 @@ test('pages the ledger without materializing Turns it takes no rows from', async const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); const owner = await tryAcquireInteractiveRootOwner(capability); assert.ok(owner); + let openedStores: Awaited> | undefined; try { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + openedStores = stores; const session = await stores.sessionStore.create({ cwd: capability.canonicalPath, llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', @@ -694,6 +703,7 @@ test('pages the ledger without materializing Turns it takes no rows from', async }); assert.deepEqual(frozen.fragments, tail.fragments); } finally { + await openedStores?.sessionStore.close?.(); await owner.close(); await rm(base, { recursive: true, force: true }); } @@ -915,6 +925,122 @@ test('does not end a page where a handoff resumes the same Turn', async () => { }); }); +for (const scenario of [ + { + code: 'unsupported_event', + message: 'thinking content has no assistant text row with a matching message id', + event: { + role: 'model', + author: 'agent', + content: { kind: 'thinking', text: 'private transcript content' }, + refs: { providerEventId: 'orphan-assistant' }, + } satisfies Partial, + }, + { + code: 'incomplete_event', + message: 'permission decision requires refs.toolCallId or a paired permission request', + event: { + actions: { + permissionDecision: { + requestId: 'missing-request', + decision: 'deny', + hint: 'private transcript content', + }, + }, + } satisfies Partial, + }, + { + code: 'tool_use_id_mismatch', + message: 'function_call content.id differs from refs.toolCallId', + event: { + role: 'model', + author: 'agent', + content: { kind: 'function_call', id: '', name: 'Read', args: {} }, + refs: { toolCallId: 'private-diagnostic-detail' }, + } satisfies Partial, + }, +]) { + test(`preserves ${scenario.code} in the durable transcript failure diagnostic`, async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-transcript-diagnostic-')); + await withNestedTranscript(base, async (read, sessionId) => { + await seed(read.stores, sessionId, 'run-1', { turnId: 'turn-1' }); + const eventId = 'failed-event-api_key=sk-secretvalue123'; + await read.stores.runtimeEventStore.appendRuntimeEvent( + sessionId, + 'run-1', + runtimeEvent(sessionId, { id: eventId, ...scenario.event }), + ); + await read.end('run-1', 'end', 'turn-1'); + for (const direction of ['older', 'newer'] as const) { + await assert.rejects( + read.readDurablePage(sessionId, { direction, maxBytes: 64 * 1024, maxMessages: 64 }), + (error: unknown) => { + assert.ok(error instanceof RuntimeReadModelError); + assert.equal(error.diagnostics.length, 1, 'retain hard diagnostics only'); + const diagnostic = error.diagnostics[0]!; + assert.equal(diagnostic.code, scenario.code); + assert.equal(diagnostic.eventId, eventId); + assert.equal(diagnostic.runId, 'run-1'); + assert.equal(diagnostic.turnId, 'turn-1'); + assert.equal(diagnostic.message, scenario.message); + if (scenario.code === 'tool_use_id_mismatch') { + assert.deepEqual(diagnostic.detail, { + contentId: '', + refToolCallId: 'private-diagnostic-detail', + }); + } + const logged = boundedFailureDiagnostic(error); + assert.ok(logged.includes(scenario.code)); + assert.ok(logged.includes(scenario.message)); + assert.ok(logged.includes(`"sessionId":"${sessionId}"`)); + assert.ok(logged.includes('"invocationId":"run-1"')); + assert.ok(logged.includes('"runId":"run-1"')); + assert.ok(logged.includes('"turnId":"turn-1"')); + assert.ok(logged.includes('failed-event-')); + assert.match(logged, /\[redacted\]/i); + assert.doesNotMatch(logged, /sk-secretvalue123/); + assert.doesNotMatch( + error.message, + /private transcript content|private-diagnostic-detail/, + ); + assert.doesNotMatch(logged, /unclaimed_control_fact/); + return true; + }, + ); + } + }); + }); +} + +test('continues serving durable transcript rows with only soft projection diagnostics', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-transcript-soft-diagnostic-')); + await withNestedTranscript(base, async (read, sessionId) => { + await seed(read.stores, sessionId, 'run-1'); + const control = runtimeEvent(sessionId, { + id: 'unclaimed-control', + turnId: 'turn-run-1', + actions: { stateDelta: { unclaimed: true } }, + }); + const projection = projectRuntimeEventsToStoredMessages([control], { + invocations: await read.stores.runtimeEventStore.listSessionInvocations(sessionId), + }); + assert.deepEqual( + projection.diagnostics.map(({ code }) => code), + ['unclaimed_control_fact'], + ); + await read.stores.runtimeEventStore.appendRuntimeEvent(sessionId, 'run-1', control); + await read.text('run-1', 'still readable'); + const page = await read.readDurableRecords(sessionId, { + direction: 'newer', + maxMessages: 64, + maxStoredBytes: 64 * 1024, + }); + assert.equal(page.records.length, 1); + assert.ok(page.records[0]!.message.type === 'assistant'); + assert.equal(page.records[0]!.message.text, 'still readable'); + }); +}); + const seed = ( stores: Awaited>, sessionId: string, @@ -945,8 +1071,10 @@ async function withNestedTranscript( const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); const owner = await tryAcquireInteractiveRootOwner(capability); assert.ok(owner); + let openedStores: Awaited> | undefined; try { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + openedStores = stores; const session = await stores.sessionStore.create({ cwd: capability.canonicalPath, llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', @@ -1014,6 +1142,7 @@ async function withNestedTranscript( session.id, ); } finally { + await openedStores?.sessionStore.close?.(); await owner.close(); await rm(base, { recursive: true, force: true }); } @@ -1024,8 +1153,10 @@ test('cuts a byte-sized page back to the last whole Turn on it', async () => { const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); const owner = await tryAcquireInteractiveRootOwner(capability); assert.ok(owner); + let openedStores: Awaited> | undefined; try { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + openedStores = stores; const session = await stores.sessionStore.create({ cwd: capability.canonicalPath, llmConnectionId: 'dddddddd-dddd-4ddd-8ddd-dddddddddddd', @@ -1082,6 +1213,7 @@ test('cuts a byte-sized page back to the last whole Turn on it', async () => { ); } } finally { + await openedStores?.sessionStore.close?.(); await owner.close(); await rm(base, { recursive: true, force: true }); } @@ -1092,8 +1224,10 @@ test('cuts a guest page where it cuts an owner page', async () => { const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); const owner = await tryAcquireInteractiveRootOwner(capability); assert.ok(owner); + let openedStores: Awaited> | undefined; try { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + openedStores = stores; const session = await stores.sessionStore.create({ cwd: capability.canonicalPath, llmConnectionId: 'eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee', @@ -1156,6 +1290,7 @@ test('cuts a guest page where it cuts an owner page', async () => { assert.ok(pages > 1, `${projection} needs more than one page to be worth cutting`); } } finally { + await openedStores?.sessionStore.close?.(); await owner.close(); await rm(base, { recursive: true, force: true }); } diff --git a/packages/runtime-host/src/server/session-transcript-reader.ts b/packages/runtime-host/src/server/session-transcript-reader.ts index a4d80a2311..a808ca89e2 100644 --- a/packages/runtime-host/src/server/session-transcript-reader.ts +++ b/packages/runtime-host/src/server/session-transcript-reader.ts @@ -27,6 +27,7 @@ import { isHardRuntimeEventReadModelDiagnostic, projectRuntimeEventUserMessage, } from '@maka/runtime/runtime-event-read-model'; +import { RuntimeReadModelError } from '@maka/runtime/runtime-read-model'; import { type CanonicalPermissionOutcomeReader, type CanonicalPermissionOutcomeRecord, @@ -157,8 +158,26 @@ function createDurableLedgerTranscriptReader(input: { turn: PendingTranscriptRun, ): Promise<{ sequence: number; message: StoredMessage }[]> => { const projected = await turn.projection.finish(input.canonicalPermissionOutcomes); - if (projected.diagnostics.some(isHardRuntimeEventReadModelDiagnostic)) { - throw new Error('Durable RuntimeEvent transcript projection is incomplete'); + const hardDiagnostics = projected.diagnostics.filter(isHardRuntimeEventReadModelDiagnostic); + if (hardDiagnostics.length > 0) { + // Host diagnostics format the Error stack, so keep the failure's location + // and reasons in its message without serializing raw diagnostic detail. + throw new RuntimeReadModelError( + `Durable RuntimeEvent transcript projection is incomplete: ${JSON.stringify({ + sessionId: turn.invocation.sessionId, + invocationId: turn.invocation.invocationId, + runId: turn.invocation.runId, + turnId: turn.invocation.turnId, + diagnostics: hardDiagnostics.map(({ code, eventId, runId, turnId, message }) => ({ + code, + eventId, + runId, + turnId, + message, + })), + })}`, + hardDiagnostics, + ); } const admission = turn.invocation.sessionId === WORKHUB_COORDINATION_SESSION_ID