From 2340635c0a623ccca90da529efc73fd160d2bbeb Mon Sep 17 00:00:00 2001 From: likun Date: Mon, 7 Sep 2026 21:30:49 +0800 Subject: [PATCH 1/2] feat(runtime): archive tool results through durable ledger refs Publish v2 ledger placeholders after reconstructibility validation under Session admission, retain legacy Artifact reads, and rebuild archive identities and hashes across conversation copies. Advance Host compatibility to 125. Generated-by: Codex --- packages/core/src/events.ts | 1 + .../core/src/tool-result-archive-evidence.ts | 1 + .../core/src/tool-result-record-schema.ts | 3 +- .../src/__tests__/execution-artifacts.test.ts | 197 +++++++++++++++++- packages/runtime-host/src/protocol/index.ts | 3 +- .../src/server/execution-artifacts.ts | 82 ++++---- .../src/server/execution-composition.ts | 6 + .../server/session-revision-coordinator.ts | 2 + .../src/__tests__/ai-sdk-backend.test.ts | 3 +- .../src/__tests__/conversation-copy.test.ts | 146 +++++++++++++ .../ledger-tool-result-archive-reader.test.ts | 114 +++++++++- ...-result-archive-capability-backend.test.ts | 1 + packages/runtime/src/archive-read-tool.ts | 6 +- packages/runtime/src/conversation-copy.ts | 84 +++++++- .../src/ledger-tool-result-archive-reader.ts | 88 +++++++- .../runtime/src/runtime-event-read-model.ts | 4 + .../src/tool-result-archive-capability.ts | 17 +- .../src/tool-result-archive-resource.ts | 93 ++++++++- .../src/tool-result-archive-transition.ts | 82 +++++--- packages/runtime/src/tool-result-archive.ts | 55 ++++- .../src/tool-result-archive-evidence.ts | 2 +- 21 files changed, 890 insertions(+), 100 deletions(-) diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index 002bbcba0e..889069e08a 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -836,6 +836,7 @@ export type ToolResultContent = toolCallId: string; toolName: string; artifactId?: string; + resourceRef?: string; bodySha256?: string; originalEstimatedTokens: number; originalBytes: number; diff --git a/packages/core/src/tool-result-archive-evidence.ts b/packages/core/src/tool-result-archive-evidence.ts index d3d6a8a95e..9f39d96310 100644 --- a/packages/core/src/tool-result-archive-evidence.ts +++ b/packages/core/src/tool-result-archive-evidence.ts @@ -26,6 +26,7 @@ export type ToolResultArchiveEvidence = readonly ok: true; readonly event: RuntimeEvent; readonly transitions: readonly AgentRunEvent[]; + readonly storedBytes?: number; } | { readonly ok: false; readonly reason: 'not_found' | 'too_large' | 'corrupt' | 'unavailable' }; diff --git a/packages/core/src/tool-result-record-schema.ts b/packages/core/src/tool-result-record-schema.ts index dc0b090285..4f75d318f0 100644 --- a/packages/core/src/tool-result-record-schema.ts +++ b/packages/core/src/tool-result-record-schema.ts @@ -65,7 +65,7 @@ const ARCHIVED_SHAPE = defineObjectShape>()( 'rewriteVersion', 'reason', ], - ['artifactId', 'bodySha256'], + ['artifactId', 'resourceRef', 'bodySha256'], ); const IMAGE_SHAPE = defineObjectShape>()(['kind', 'mimeType', 'ref'], []); const SUMMARY_SHAPE = defineObjectShape>()( @@ -242,6 +242,7 @@ function isNonShellToolResultContent(value: unknown): value is ToolResultContent typeof value.toolCallId === 'string' && typeof value.toolName === 'string' && isOptionalString(value.artifactId) && + isOptionalString(value.resourceRef) && isOptionalString(value.bodySha256) && isFiniteNumber(value.originalEstimatedTokens) && isFiniteNumber(value.originalBytes) && diff --git a/packages/runtime-host/src/__tests__/execution-artifacts.test.ts b/packages/runtime-host/src/__tests__/execution-artifacts.test.ts index 1143f0c5de..ffd4ce9d60 100644 --- a/packages/runtime-host/src/__tests__/execution-artifacts.test.ts +++ b/packages/runtime-host/src/__tests__/execution-artifacts.test.ts @@ -19,6 +19,14 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; +import { DatabaseSync } from 'node:sqlite'; +import { openToolResultArchiveEvidenceReader } from '@maka/storage/tool-result-archive-evidence'; +import { + buildModelProjectionTransition, + durableToolResultProjectionDigest, +} from '@maka/core/model-projection-transition'; +import { buildLedgerArchivedToolResultPlaceholder } from '@maka/runtime/tool-result-archive'; +import { parseToolResultArchiveResourceRef } from '@maka/runtime/tool-result-archive-resource'; import { mkdir, mkdtemp, rm, stat, truncate, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -35,6 +43,179 @@ import { createHostExecutionArtifactServices } from '../server/execution-artifac import { restoreArtifactV1Shape } from './fixtures/artifact-v1.js'; import { SessionAdmissionGate } from '../server/session-admission-gate.js'; +test('production archive services prepare ledger writes, keep legacy reads, and replay after reopen', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-ledger-archive-host-')); + const owner = await tryAcquireInteractiveRootOwner( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + assert.ok(owner); + const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); + let evidence = await openToolResultArchiveEvidenceReader(owner.lease); + const db = new DatabaseSync(join(root, 'runtime.sqlite')); + try { + const projection = { version: 1 as const, kind: 'text' as const, text: 'durable ledger body' }; + const serializedResult = JSON.stringify(projection.text); + const bodySha256 = createHash('sha256').update(serializedResult).digest('hex'); + const event = { + id: 'response', + sessionId: 'session', + runId: 'run', + invocationId: 'invocation', + turnId: 'turn', + ts: 1, + partial: false, + author: 'tool', + role: 'tool', + content: { + kind: 'function_response', + id: 'call', + name: 'Read', + result: 'raw execution body', + modelProjection: projection, + }, + }; + db.prepare( + 'INSERT INTO runtime_events(event_id, session_id, invocation_id, run_id, turn_id, event_seq, event_kind, payload_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', + ).run( + 'response', + 'session', + 'invocation', + 'run', + 'turn', + 1, + 'function_response', + JSON.stringify(event), + 1, + ); + const old = await artifacts.create({ + id: 'legacy-archive', + sessionId: 'session', + turnId: 'turn', + name: 'legacy.json', + kind: 'file', + content: serializedResult, + source: 'tool_result_archive', + }); + const input = { + sessionId: 'session', + runtimeEventId: 'response', + turnId: 'turn', + toolCallId: 'call', + toolName: 'Read', + serializedResult, + bodySha256, + originalBytes: Buffer.byteLength(serializedResult), + originalEstimatedTokens: 10, + rewriteVersion: 1, + sourceProjectionDigest: durableToolResultProjectionDigest(projection), + reason: 'stale_tool_result_pruned_before_compact' as const, + }; + const make = () => + createHostExecutionArtifactServices({ + artifacts, + archiveEvidence: evidence, + sessionAdmission: new SessionAdmissionGate(), + sessions: { probeSessionRemoval: async () => ({ kind: 'present' }) }, + requestDrain: () => assert.fail('archive failure must not drain'), + }); + let services = make(); + const prepared = await services.toolResultArchive.services.archiveToolResult(input); + assert.ok(prepared?.ledger); + assert.equal(typeof prepared.commitTransition, 'function'); + assert.equal((await artifacts.listPage('session', { offset: 0, limit: 10 })).total, 1); + const placeholder = buildLedgerArchivedToolResultPlaceholder({ ...input, storage: 'ledger' }); + const transition = buildModelProjectionTransition({ + sessionId: 'session', + target: { + runtimeEventId: 'response', + part: 'tool_result', + toolCallId: 'call', + toolName: 'Read', + }, + sourceProjection: projection, + replacement: { version: 1, kind: 'json', value: placeholder as never }, + now: 2, + }); + db.prepare('INSERT INTO core_agent_runs(session_id, run_id, created_at) VALUES (?, ?, ?)').run( + 'session', + 'run', + 1, + ); + const persist = async () => { + db.prepare('INSERT INTO core_agent_run_events VALUES (?, ?, ?, ?, ?, ?, ?)').run( + 'session', + 'run', + 1, + transition.transitionId, + 'model_projection_transition_recorded', + 2, + JSON.stringify({ + id: transition.transitionId, + type: 'model_projection_transition_recorded', + sessionId: 'session', + runId: 'run', + turnId: 'turn', + ts: 2, + data: { runtimeEventId: 'response', part: 'tool_result', transition }, + }), + ); + }; + assert.equal(await prepared.commitTransition!(transition, persist), true); + assert.equal( + await prepared.commitTransition!(transition, async () => + assert.fail('stale preparation must not append'), + ), + false, + ); + evidence.close(); + evidence = await openToolResultArchiveEvidenceReader(owner.lease); + services = make(); + assert.deepEqual( + await services.toolResultArchive.services.readToolResultArchive({ + ...placeholder, + sessionId: 'session', + }), + { ok: true, serializedResult }, + ); + assert.equal( + ( + await services.toolResultArchive.services.readToolResultArchive({ + ...placeholder, + sessionId: 'other', + }) + ).ok, + false, + ); + const identity = parseToolResultArchiveResourceRef(placeholder.resourceRef!); + assert.ok(identity); + assert.deepEqual( + await services.toolResultArchive.services.readArchivedToolResultResource({ + ...identity, + sessionId: 'session', + maxBytes: input.originalBytes, + }), + { ok: true, serializedResult }, + ); + assert.deepEqual( + await services.toolResultArchive.services.readArchivedToolResultResource({ + artifactId: old.id, + bodySha256, + originalBytes: input.originalBytes, + sessionId: 'session', + maxBytes: input.originalBytes, + }), + { ok: true, serializedResult }, + ); + } finally { + db.close(); + evidence.close(); + artifacts.close(); + await owner.close(); + await rm(root, { recursive: true, force: true }); + await rm(owner.controlDirectory, { recursive: true, force: true }); + } +}); + test('a refused projection preserves a shared image until Session cleanup', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-shared-projection-')); const owner = await tryAcquireInteractiveRootOwner( @@ -199,12 +380,20 @@ test('Hosted execution publishes contained Tool Artifacts and durable result arc reason: 'stale_tool_result_pruned_before_compact' as const, bodySha256, }; - const archived = await services.toolResultArchive.services.archiveToolResult(archiveInput); - assert.ok(archived, 'the host archive writer always reports where it stored the body'); - assert.deepEqual( + assert.equal( await services.toolResultArchive.services.archiveToolResult(archiveInput), - archived, + undefined, + 'without ledger evidence the Host must not fall back to publishing an Artifact', ); + const legacy = await store.create({ + sessionId: archiveInput.sessionId, + turnId: archiveInput.turnId, + name: 'legacy.json', + kind: 'file', + content: serializedResult, + source: 'tool_result_archive', + }); + const archived = { artifactId: legacy.id }; assert.deepEqual( await services.toolResultArchive.services.readToolResultArchive({ ...archiveInput, diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 78e0241d89..ac0d00407f 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 131 as const; +// 132: new Tool Result archives use versioned ledger references, not Artifact payloads. +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 132 as const; // 131: Logical model steps bind durable Request Composition identities. // 130: Turn contributions carry the optional bounded `failureMessage` diagnostic. // Epoch-129 peers reject this added field on the strict contribution shape. diff --git a/packages/runtime-host/src/server/execution-artifacts.ts b/packages/runtime-host/src/server/execution-artifacts.ts index 43186e196f..fb856e6e64 100644 --- a/packages/runtime-host/src/server/execution-artifacts.ts +++ b/packages/runtime-host/src/server/execution-artifacts.ts @@ -24,10 +24,15 @@ import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments'; import { createToolResultArchiveCapability, type ToolResultArchiveCapability, - type ToolResultArchiveRecorderInput, + type ToolResultArchiveRecorder, } from '@maka/runtime/tool-result-archive-capability'; import { isPathInside } from '@maka/runtime/path-containment'; -import { stableToolResultArchiveArtifactId } from '@maka/runtime/tool-result-archive'; +import { + createLedgerArchivePreparer, + createLedgerArchiveResourceReader, + createLedgerToolResultArchiveReader, +} from '@maka/runtime/ledger-tool-result-archive-reader'; +import type { ToolResultArchiveEvidenceReader } from '@maka/core/tool-result-archive-evidence'; import { type ToolArtifactRecorderInput } from '@maka/runtime/tool-artifacts'; import { type ToolResultArchiveReaderInput, @@ -43,9 +48,8 @@ export interface HostExecutionArtifactServices { recordToolArtifacts(event: ToolArtifactRecorderInput): Promise; publishChildWorkspacePatch: NonNullable; /** - * One archive authority over the session artifact store (#2026). All three - * reads and the writer address the same store, so the host has no way to hand - * out half of it. + * New archives use the Session ledger. Legacy Artifact refs retain their + * scoped reader; the writer never falls back to publishing Artifact bytes. */ toolResultArchive: ToolResultArchiveCapability; } @@ -55,6 +59,7 @@ export function createHostExecutionArtifactServices(input: { requestDrain: () => void; sessionAdmission: SessionAdmissionGate; sessions: SessionPresenceReader; + archiveEvidence?: ToolResultArchiveEvidenceReader; }): HostExecutionArtifactServices { const runWrite = async (operation: () => Promise): Promise => { try { @@ -92,6 +97,32 @@ export function createHostExecutionArtifactServices(input: { } }; + const prepareLedger = input.archiveEvidence + ? createLedgerArchivePreparer(input.archiveEvidence) + : undefined; + const readLedger = input.archiveEvidence + ? createLedgerToolResultArchiveReader(input.archiveEvidence) + : undefined; + const readLedgerResource = input.archiveEvidence + ? createLedgerArchiveResourceReader(input.archiveEvidence) + : undefined; + const prepareLedgerForCommit: ToolResultArchiveRecorder = async (event) => { + const accepted = { ...event }; + if (!prepareLedger || !(await prepareLedger(accepted))) return; + return { + ledger: true, + commitTransition: (transition, persist) => + input.sessionAdmission.runOrJoin(accepted.sessionId, async () => { + if ( + (await input.sessions.probeSessionRemoval(accepted.sessionId)).kind !== 'present' || + !(await prepareLedger(accepted)) + ) + return false; + await persist(transition); + return true; + }), + }; + }; const services: HostExecutionArtifactServices = { recordToolArtifacts, publishChildWorkspacePatch: async ({ sessionId, turnId, binding, patch }) => { @@ -111,40 +142,15 @@ export function createHostExecutionArtifactServices(input: { return artifact; }, toolResultArchive: createToolResultArchiveCapability({ - archiveToolResult: (event: ToolResultArchiveRecorderInput) => - runWrite(async () => { - const artifactId = stableToolResultArchiveArtifactId(event); - const existing = await input.artifacts.getInSession(event.sessionId, artifactId); - if (existing.record) { - const read = await readArchive(input.artifacts, { - artifactId, - sessionId: event.sessionId, - bodySha256: event.bodySha256, - originalBytes: event.originalBytes, - maxBytes: event.originalBytes, - }); - if (!read.ok) { - throw new Error(`Tool result archive identity conflict: ${read.reason}`); - } - return { artifactId }; - } - const artifact = await input.artifacts.create({ - id: artifactId, - sessionId: event.sessionId, - turnId: event.turnId, - name: `archived-${event.toolName}-${event.runtimeEventId}.json`, - kind: 'file', - content: event.serializedResult, - mimeType: 'application/json', - source: 'tool_result_archive', - summary: `Archived ${event.toolName} tool result for context budget replay`, - }); - return { artifactId: artifact.id }; - }), + archiveToolResult: prepareLedgerForCommit, readToolResultArchive: (event: ToolResultArchiveReaderInput) => - readArchive(input.artifacts, event), + event.rewriteVersion === 2 + ? (readLedger?.(event) ?? { ok: false, reason: 'read_failed' }) + : readArchive(input.artifacts, event), readArchivedToolResultResource: (event: ToolResultArchiveResourceReadInput) => - readArchive(input.artifacts, event), + event.storage === 'ledger' + ? (readLedgerResource?.(event) ?? { ok: false, reason: 'read_failed' }) + : readArchive(input.artifacts, event), }), }; return Object.freeze(services); @@ -202,7 +208,7 @@ function contentBytes(content: string | Uint8Array): number { async function readArchive( artifacts: InteractiveArtifactStoreWriter, event: Pick< - ToolResultArchiveReaderInput, + Extract, 'artifactId' | 'sessionId' | 'bodySha256' | 'originalBytes' | 'maxBytes' >, ): Promise { diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index bd11b01cfc..9caabd01ad 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -208,6 +208,7 @@ import { } from './web-search-tool.js'; import { createHostWebFetchService, createHostWebFetchToolFromService } from './web-fetch-tool.js'; import { createHostExecutionArtifactServices } from './execution-artifacts.js'; +import { openToolResultArchiveEvidenceReader } from '@maka/storage/tool-result-archive-evidence'; import { createRuntimeHostWorkspaceExecutionComposition, RuntimeHostWorkspaceExecutionError, @@ -290,6 +291,7 @@ export async function createExecutionRuntimeHostComposition( let pluginPlatform: HostPluginPlatform | undefined; let manager: SessionManager | undefined; let modelMetadataRefresh: ReturnType | undefined; + let archiveEvidence: Awaited> | undefined; try { const pluginRoot = new Context(); const pluginTools = new PluginToolService(pluginRoot); @@ -426,7 +428,9 @@ export async function createExecutionRuntimeHostComposition( onProjectionChanged: (update) => requireContinuity(continuity).enqueueRuntimeResourceChanged(update), }); + archiveEvidence = await openToolResultArchiveEvidenceReader(context.owner.lease); const executionArtifacts = createHostExecutionArtifactServices({ + archiveEvidence, artifacts: openedArtifactStore, requestDrain: context.requestDrain, sessionAdmission, @@ -1900,6 +1904,7 @@ export async function createExecutionRuntimeHostComposition( () => oauth?.beginDrain(), ], close: [ + () => archiveEvidence?.close(), () => modelMetadataRefresh?.close(), () => connectionEffects.close(), () => @@ -2156,6 +2161,7 @@ export async function createExecutionRuntimeHostComposition( }; } catch (error) { const errors: unknown[] = [error]; + archiveEvidence?.close(); try { await modelMetadataRefresh?.close(); } catch (closeError) { diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index 1b92e210af..4f0344d0fa 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -955,6 +955,7 @@ function collectArchivedToolResultPlaceholders( const add = (value: unknown): boolean => { if (!isRecord(value) || value.kind !== 'maka.archived_tool_result') return true; if (!isArchivedToolResultPlaceholder(value)) return false; + if (value.rewriteVersion === 2) return true; addDescriptor(value); return true; }; @@ -990,6 +991,7 @@ function collectArchivedToolResultPlaceholders( continue; } if (message.content.kind === 'archived_tool_result') { + if (message.content.rewriteVersion === 2 && message.content.resourceRef) continue; if (!message.content.artifactId && !message.content.bodySha256) continue; if (!message.content.artifactId || !message.content.bodySha256) return null; addDescriptor({ diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 086e5b36e6..9e9f82bcdc 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -8977,7 +8977,8 @@ describe('AiSdkBackend usage telemetry', () => { }, readToolResultArchive: async () => ({ ok: false, reason: 'not_found' }), readArchivedToolResultResource: async (event) => { - const serializedResult = store.get(event.artifactId); + const serializedResult = + event.storage === 'ledger' ? undefined : store.get(event.artifactId); return serializedResult === undefined ? { ok: false, reason: 'not_found' } : { ok: true, serializedResult }; diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 37de5690e7..2e64f664bb 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -18,6 +18,7 @@ */ import assert from 'node:assert/strict'; +import { createLedgerToolResultArchiveReader } from '../ledger-tool-result-archive-reader.js'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -32,6 +33,7 @@ import { decodeModelCallAttempt } from '@maka/core/model-call-attempt'; import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; import { buildModelProjectionTransition, + durableToolResultProjectionDigest, MODEL_PROJECTION_TRANSITION_EVENT_TYPE, type ModelProjectionTransition, } from '@maka/core/model-projection-transition'; @@ -79,6 +81,8 @@ import { } from '../tool-result-archive-transition.js'; import { buildArchivedToolResultPlaceholder, + buildLedgerArchivedToolResultPlaceholder, + type ArchivedToolResultPlaceholder, isArchivedToolResultPlaceholder, } from '../tool-result-archive.js'; import { testInvocationOpening, testInvocationRecord } from './invocation-fixture.js'; @@ -2794,6 +2798,148 @@ function sourceProjectionTransition(input: { }); } +test('conversation copy rebuilds ledger archive hashes after remapping image refs', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-ledger-copy-')); + try { + const runStore = createSqliteAgentRunStore(root); + const runtimeEventStore = createWorkspaceRuntimeStore(root); + await seedRun(runtimeEventStore, { + runId: 'run-source', + invocationId: 'invocation-source', + turnId: 'turn-1', + cwd: root, + }); + const imageRef = { + kind: 'session_context' as const, + sessionId: 'session-source', + refId: 'image-source', + }; + const projection: DurableToolResultProjection = { + version: 1, + kind: 'content', + parts: [ + { kind: 'text', text: 'caption' }, + { kind: 'artifact', mediaType: 'image/png', ref: imageRef }, + ], + }; + const result = runtimeEvent({ + id: 'event-result-ledger', + ts: 2, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-1', + name: 'Read', + result: { kind: 'image', mimeType: 'image/png', ref: imageRef }, + modelProjection: projection, + }, + }); + for (const event of [ + runtimeEvent({ + id: 'event-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'copy image' }, + }), + runtimeEvent({ + id: 'event-call', + ts: 1.5, + role: 'model', + author: 'agent', + content: { kind: 'function_call', id: 'tool-1', name: 'Read', args: { path: 'image.png' } }, + }), + result, + runtimeEvent({ id: 'event-terminal', ts: 3, status: 'completed' }), + ]) + await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); + const serialized = serializedToolResultProjection(projection); + const placeholder = buildLedgerArchivedToolResultPlaceholder({ + storage: 'ledger', + runtimeEventId: result.id, + toolCallId: 'tool-1', + toolName: 'Read', + sourceProjectionDigest: durableToolResultProjectionDigest(projection), + bodySha256: sha256(serialized), + originalBytes: Buffer.byteLength(serialized), + originalEstimatedTokens: 100, + reason: 'stale_tool_result_pruned_before_compact', + }); + const transition = buildModelProjectionTransition({ + sessionId: 'session-source', + target: { + runtimeEventId: result.id, + part: 'tool_result', + toolCallId: 'tool-1', + toolName: 'Read', + }, + sourceProjection: projection, + replacement: archivedToolResultProjection(placeholder), + now: 4, + }); + await runStore.appendEvent('session-source', 'run-source', { + type: MODEL_PROJECTION_TRANSITION_EVENT_TYPE, + id: transition.transitionId, + runId: 'run-source', + sessionId: 'session-source', + turnId: 'turn-1', + ts: 4, + data: { runtimeEventId: result.id, part: 'tool_result', transition }, + }); + const source = await new RuntimeReadModel({ runtimeEventStore }).getSessionView( + 'session-source', + ); + await cloneConversationRuntimeLedger({ + plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), + copiedMessages: source.messages, + referenceMap: { + mode: 'exact', + linkedChildren: { mode: 'reject' }, + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map(), + relativePaths: new Map(), + contextRefs: new Map([['image-source', 'image-target']]), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const [targetRun] = await runtimeEventStore.listSessionInvocations('session-target'); + const targetEvents = await runtimeEventStore.readRuntimeEvents( + 'session-target', + targetRun!.runId, + ); + const target = targetEvents.find((event) => event.content?.kind === 'function_response')!; + const records = await runStore.readEvents('session-target', targetRun!.runId); + const transitions = await loadModelProjectionTransitionsFromRunLedger( + runStore, + 'session-target', + [targetRun!.runId], + ); + const copied = transitions.transitions[0]!; + assert.ok(copied.replacement.kind === 'json'); + assert.ok(isArchivedToolResultPlaceholder(copied.replacement.value)); + const copiedPlaceholder = copied.replacement.value as ArchivedToolResultPlaceholder; + assert.equal(copiedPlaceholder.rewriteVersion, 2); + assert.notEqual(copiedPlaceholder.bodySha256, placeholder.bodySha256); + assert.notEqual(copiedPlaceholder.resourceRef, placeholder.resourceRef); + const read = createLedgerToolResultArchiveReader({ + read: async () => ({ + ok: true, + event: target, + transitions: records.filter((row) => row.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE), + }), + }); + const body = await read({ ...copiedPlaceholder, sessionId: 'session-target' }); + assert.ok(body.ok); + assert.match(body.serializedResult, /image-target/); + assert.doesNotMatch(body.serializedResult, /image-source/); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('conversation copy rebuilds projection transitions against the copied events', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-conversation-transition-copy-')); try { diff --git a/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts b/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts index e5532c6d40..6d4f3af719 100644 --- a/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts +++ b/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts @@ -30,7 +30,17 @@ import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { AgentRunEvent } from '@maka/core/agent-run'; import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; import { buildModelProjectionTransition } from '@maka/core/model-projection-transition'; -import { createLedgerToolResultArchiveReader } from '../ledger-tool-result-archive-reader.js'; +import { + createLedgerToolResultArchiveReader, + createLedgerArchivePreparer, + createLedgerArchiveResourceReader, +} from '../ledger-tool-result-archive-reader.js'; +import { archiveToolResultAsTransition } from '../tool-result-archive-transition.js'; +import { + readToolResultArchiveResource, + parseToolResultArchiveResourceRef, +} from '../tool-result-archive-resource.js'; +import type { ModelProjectionTransition } from '@maka/core/model-projection-transition'; import { serializedToolResultProjection } from '../tool-result-archive-transition.js'; import { serializeToolResultProjectionV1 } from '../tool-result-archive-encoding.js'; import { @@ -294,6 +304,108 @@ test('unknown transition versions and incomplete evidence do not expose a source }); }); +test('new archive commits only a v2 ledger reference and is readable through the resource decoder', async () => { + const f = fixture(); + f.records.length = 0; + const evidence = { + read: async () => ({ + ok: true as const, + event: f.event, + transitions: f.records, + storedBytes: 1024, + }), + }; + const prepare = createLedgerArchivePreparer(evidence); + const outcome = await archiveToolResultAsTransition( + { + sessionId: 'session', + archiveToolResult: prepare, + recordTransition: async (transition) => { + f.records.push(envelope(transition)); + }, + loadTransitions: async () => ({ + transitions: f.records.map((row) => row.data!.transition as ModelProjectionTransition), + }), + now: () => 2, + }, + { + runtimeEventId: 'response', + turnId: 'turn', + toolCallId: 'call', + toolName: 'Read', + sourceProjection: f.source, + serializedResult: f.body, + originalBytes: Buffer.byteLength(f.body), + originalEstimatedTokens: 100, + reason: 'stale_tool_result_pruned_before_compact', + }, + ); + assert.ok(outcome); + assert.equal(outcome.placeholder.rewriteVersion, 2); + assert.equal(outcome.placeholder.artifactId, undefined); + assert.match(outcome.placeholder.resourceRef!, /^maka:\/\/archive-ledger\/v1\//); + const reader = createLedgerToolResultArchiveReader(evidence); + assert.deepEqual(await reader({ ...outcome.placeholder, sessionId: 'session' }), { + ok: true, + serializedResult: f.body, + }); + const resource = createLedgerArchiveResourceReader(evidence); + const read = await readToolResultArchiveResource( + { + readArchivedToolResultResource: (input) => + input.storage === 'ledger' ? resource(input) : { ok: false, reason: 'not_found' }, + }, + 'session', + { ref: outcome.placeholder.resourceRef!, operation: 'read' }, + ); + assert.match(JSON.stringify(read), /bounded model output/); + assert.ok(parseToolResultArchiveResourceRef(outcome.placeholder.resourceRef!)); + assert.equal( + parseToolResultArchiveResourceRef(outcome.placeholder.resourceRef! + '#extra'), + null, + ); +}); + +test('preflight and transition failure leave the source projection unchanged', async () => { + const f = fixture(); + f.records.length = 0; + let writes = 0; + const request = { + runtimeEventId: 'response', + turnId: 'turn', + toolCallId: 'call', + toolName: 'Read', + sourceProjection: f.source, + serializedResult: f.body, + originalBytes: Buffer.byteLength(f.body), + originalEstimatedTokens: 100, + reason: 'stale_tool_result_pruned_before_compact' as const, + }; + const services = { + sessionId: 'session', + archiveToolResult: createLedgerArchivePreparer({ + read: async () => ({ ok: true, event: f.event, transitions: f.records, storedBytes: 1024 }), + }), + recordTransition: async () => { + writes += 1; + throw new Error('commit failed'); + }, + now: () => 2, + }; + assert.equal( + await archiveToolResultAsTransition(services, { ...request, serializedResult: '"wrong"' }), + undefined, + ); + assert.equal(writes, 0); + assert.equal(await archiveToolResultAsTransition(services, request), undefined); + assert.equal(writes, 1); + assert.equal(f.records.length, 0); + assert.deepEqual( + f.event.content?.kind === 'function_response' ? f.event.content.modelProjection : null, + f.source, + ); +}); + test('availability maps to read_failed while corrupt evidence remains corrupt', async () => { const f = fixture(); for (const reason of ['unavailable', 'corrupt'] as const) { diff --git a/packages/runtime/src/__tests__/tool-result-archive-capability-backend.test.ts b/packages/runtime/src/__tests__/tool-result-archive-capability-backend.test.ts index 3dd86309b0..fc46670ae2 100644 --- a/packages/runtime/src/__tests__/tool-result-archive-capability-backend.test.ts +++ b/packages/runtime/src/__tests__/tool-result-archive-capability-backend.test.ts @@ -119,6 +119,7 @@ describe('AiSdkBackend tool-result archive capability', () => { reason: 'active_current_turn_tool_result_pruned_before_next_step', }); assert.ok(written, 'the writer must report where it archived the body'); + assert.ok(written.artifactId); const page = (await archive.archiveReadTool.impl( { diff --git a/packages/runtime/src/archive-read-tool.ts b/packages/runtime/src/archive-read-tool.ts index 642b9bd98a..d84cbc3c08 100644 --- a/packages/runtime/src/archive-read-tool.ts +++ b/packages/runtime/src/archive-read-tool.ts @@ -35,7 +35,9 @@ export function buildArchiveReadTool(reader: ToolResultArchiveResourceReader): M .object({ ref: z .string() - .describe('A maka://archive/... ref returned in an archived tool-result placeholder'), + .describe( + 'The exact resourceRef returned in an archived tool-result placeholder (ledger or legacy archive)', + ), operation: z .enum(['inspect', 'read', 'query', 'search']) .default('inspect') @@ -100,7 +102,7 @@ export function buildArchiveReadTool(reader: ToolResultArchiveResourceReader): M displayName: 'Read archived result', activityKind: 'read', description: - 'Inspect, search, or page through a tool-result archive returned as a maka://archive/... ref. Start with inspect for a preview and the char/line coordinate space. Use operation "search" with a pattern to locate text, operation "read" with unit "line" for line-oriented terminal output, or operation "query" with an itemId for one agent_swarm item. Results are strictly bounded so reading an archive cannot immediately trigger another archive.', + 'Inspect, search, or page through a tool-result archive using its exact resourceRef. Both ledger and legacy archive references are supported. Start with inspect for a preview and the char/line coordinate space. Use operation "search" with a pattern to locate text, operation "read" with unit "line" for line-oriented terminal output, or operation "query" with an itemId for one agent_swarm item. Results are strictly bounded so reading an archive cannot immediately trigger another archive.', parameters: jsonSchema(async () => await providerSchema.jsonSchema, { validate: async (value) => { const result = await parameters.safeParseAsync(value); diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 8cbccfdaeb..8db580dc3b 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -58,11 +58,14 @@ import { deserializeToolResultArchive, isArchivedToolResultPlaceholder, type ArchivedToolResultPlaceholder, + type LedgerArchivedToolResultPlaceholder, + buildLedgerArchivedToolResultPlaceholder, } from './tool-result-archive.js'; import { rewriteDurableToolResultProjectionArtifactRefs } from './durable-tool-result-projection.js'; import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; import { buildModelProjectionTransition, + durableToolResultProjectionDigest, decodeModelProjectionTransition, MODEL_PROJECTION_TRANSITION_EVENT_TYPE, type ModelProjectionTransition, @@ -73,6 +76,8 @@ import { reduceEffectiveModelProjections, } from './model-projection-transition-ledger.js'; import { archivedToolResultProjection } from './tool-result-archive-transition.js'; +import { serializeToolResultProjectionV1 } from './tool-result-archive-encoding.js'; +import { createHash } from 'node:crypto'; export interface ConversationCopySlice { readonly messages: readonly StoredMessage[]; @@ -122,6 +127,7 @@ export type ConversationCopyArtifactReferenceMap = }); export type ConversationCopyMessageReferenceMap = ConversationCopyArtifactReferenceMap & { + readonly ledgerArchives?: Map; readonly runIds: ReadonlyMap; readonly runtimeEventIds: ReadonlyMap; readonly providerTraceIds: ReadonlyMap; @@ -538,6 +544,7 @@ export async function cloneConversationRuntimeLedger( runtimeEventIds, providerTraceIds, agentRunEventIds: operationalEventIds, + ledgerArchives: new Map(), }; const clonedEventBySourceId = new Map(); for (const plan of flattenedPlans) { @@ -608,6 +615,34 @@ export async function cloneConversationRuntimeLedger( rewriteConversationCopyMessage(message, references), ); + const rewriteLedgerText = (text: string): string => { + for (const [source, target] of references.ledgerArchives ?? []) + text = text.split(source).join(target.resourceRef!); + return text; + }; + for (const message of copiedMessages) { + if (message.type === 'user' || message.type === 'assistant') + message.text = rewriteLedgerText(message.text); + } + for (const event of clonedEventBySourceId.values()) { + if (event.content?.kind === 'text') event.content.text = rewriteLedgerText(event.content.text); + if (event.content?.kind === 'function_call' && event.content.name === 'ArchiveRead') { + const args = event.content.args; + if (args && typeof args === 'object' && 'ref' in args && typeof args.ref === 'string') { + event.content.args = { ...args, ref: rewriteLedgerText(args.ref) }; + } + } + } + if (references.ledgerArchives?.size) { + // Frozen summaries may embed source archive addresses. Keep the copied raw + // ledger and transitions; the target can compact again with its own refs. + for (const plan of preparedPlans) { + plan.clonedOperationalEvents = plan.clonedOperationalEvents.filter( + (event) => event.type !== 'history_compact_checkpoint_recorded', + ); + } + } + const importedSourceEventIds = new Set(); const orderedBatches = input.plan.inlineRuntimeEvents.flatMap((event) => { const cloned = clonedEventBySourceId.get(event.id); @@ -1007,16 +1042,40 @@ function cloneModelProjectionTransition( } const clonedTarget = clonedRuntimeEvents.get(source.target.runtimeEventId); if (!clonedTarget) return null; - const placeholder = source.replacement.kind === 'json' ? source.replacement.value : undefined; - if (!isArchivedToolResultPlaceholder(placeholder)) { + const rawPlaceholder = source.replacement.kind === 'json' ? source.replacement.value : undefined; + if (!isArchivedToolResultPlaceholder(rawPlaceholder)) { throw new Error(`Cannot copy unsupported model projection transition ${event.id}`); } + const placeholder = rawPlaceholder as ArchivedToolResultPlaceholder; const existing = transitionState.get(clonedTarget.id); const sourceProjection = existing?.projection ?? baseToolResultProjection(clonedTarget); if (!sourceProjection) { throw new Error(`Cannot copy model projection transition ${event.id} onto its target`); } - const rewritten = rewriteArchivedToolResult(placeholder, references); + let rewritten: ArchivedToolResultPlaceholder; + if (placeholder.rewriteVersion === 2) { + const { previousTransitionId: _previous, ...rest } = placeholder; + const serialized = serializeToolResultProjectionV1(sourceProjection); + rewritten = buildLedgerArchivedToolResultPlaceholder({ + ...rest, + runtimeEventId: clonedTarget.id, + sourceProjectionDigest: durableToolResultProjectionDigest(sourceProjection), + bodySha256: createHash('sha256').update(serialized).digest('hex'), + originalBytes: Buffer.byteLength(serialized), + ...(source.previousTransitionId + ? { + previousTransitionId: requiredMappedId( + transitionIds, + source.previousTransitionId, + 'model projection transition', + ), + } + : {}), + }); + references.ledgerArchives?.set(buildToolResultArchiveResourceRef(placeholder), rewritten); + } else { + rewritten = rewriteArchivedToolResult(placeholder, references); + } const transition = buildModelProjectionTransition({ sessionId: references.targetSessionId, target: { @@ -1492,6 +1551,18 @@ function rewriteToolResultContent( return { ...content, ref: rewriteStorageRef(content.ref, references) }; } if (content.kind === 'archived_tool_result') { + if (content.rewriteVersion === 2 && content.resourceRef) { + const rewritten = references.ledgerArchives?.get(content.resourceRef); + if (!rewritten) + throw new Error('Cannot copy ledger archive without its committed transition'); + return { + ...content, + runtimeEventId: rewritten.runtimeEventId, + resourceRef: rewritten.resourceRef, + bodySha256: rewritten.bodySha256, + originalBytes: rewritten.originalBytes, + }; + } const snapshot = rewriteArchivedSnapshot(content, references); if (snapshot) return snapshot; return { @@ -1810,6 +1881,11 @@ function rewriteArchivedToolResult( value: ArchivedToolResultPlaceholder, references: ConversationCopyMessageReferenceMap, ): ArchivedToolResultPlaceholder { + if (value.rewriteVersion === 2) { + const rewritten = references.ledgerArchives?.get(buildToolResultArchiveResourceRef(value)); + if (!rewritten) throw new Error('Cannot copy ledger archive without its committed transition'); + return rewritten; + } const artifactId = rewriteOwnedArtifactId(value.artifactId, references); const resource = value.resourceRef ? parseToolResultArchiveResourceRef(value.resourceRef) @@ -1822,7 +1898,7 @@ function rewriteArchivedToolResult( 'RuntimeEvent', ), artifactId, - ...(resource && artifactId !== value.artifactId + ...(resource && resource.storage !== 'ledger' && artifactId !== value.artifactId ? { resourceRef: buildToolResultArchiveResourceRef({ ...resource, diff --git a/packages/runtime/src/ledger-tool-result-archive-reader.ts b/packages/runtime/src/ledger-tool-result-archive-reader.ts index 398ccd53a2..41f6c65b29 100644 --- a/packages/runtime/src/ledger-tool-result-archive-reader.ts +++ b/packages/runtime/src/ledger-tool-result-archive-reader.ts @@ -19,7 +19,10 @@ import { createHash } from 'node:crypto'; import { decodeDurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; -import type { ModelProjectionTransition } from '@maka/core/model-projection-transition'; +import { + durableToolResultProjectionDigest, + type ModelProjectionTransition, +} from '@maka/core/model-projection-transition'; import type { ToolResultArchiveEvidenceReader } from '@maka/core/tool-result-archive-evidence'; import { decodeLedgerTransition, @@ -30,6 +33,12 @@ import { type ToolResultArchiveReader, } from './tool-result-archive.js'; import { serializeToolResultProjectionV1 } from './tool-result-archive-encoding.js'; +import type { LedgerArchiveResourceIdentity } from './tool-result-archive-resource.js'; +import type { ToolResultArchiveRecorder } from './tool-result-archive-capability.js'; +import { + TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_BYTES, + TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_TRANSITIONS, +} from '@maka/core/tool-result-archive-evidence'; /** * Read-only v1 reconstruction. It never calls a live tool projector, reads an @@ -100,7 +109,13 @@ export function createLedgerToolResultArchiveReader( const placeholder = replacement.kind === 'json' ? replacement.value : undefined; if ( isArchivedToolResultPlaceholder(placeholder) && - placeholder.artifactId === request.artifactId + ((placeholder.rewriteVersion === 1 && + request.rewriteVersion === 1 && + placeholder.artifactId === request.artifactId) || + (placeholder.rewriteVersion === 2 && + request.rewriteVersion === 2 && + placeholder.sourceProjectionDigest === request.sourceProjectionDigest && + placeholder.previousTransitionId === request.previousTransitionId)) ) { if ( placeholder.runtimeEventId !== request.runtimeEventId || @@ -111,6 +126,12 @@ export function createLedgerToolResultArchiveReader( placeholder.rewriteVersion !== request.rewriteVersion ) return { ok: false, reason: 'source_mismatch' }; + if ( + placeholder.rewriteVersion === 2 && + (placeholder.sourceProjectionDigest !== transition.sourceProjectionDigest || + placeholder.previousTransitionId !== transition.previousTransitionId) + ) + return { ok: false, reason: 'corrupt' }; const serializedResult = serializeToolResultProjectionV1(source); if (Buffer.byteLength(serializedResult, 'utf8') !== request.originalBytes) return { ok: false, reason: 'size_mismatch' }; @@ -126,3 +147,66 @@ export function createLedgerToolResultArchiveReader( } }; } + +export function createLedgerArchiveResourceReader(evidence: ToolResultArchiveEvidenceReader) { + const reader = createLedgerToolResultArchiveReader(evidence); + return (input: LedgerArchiveResourceIdentity & { sessionId: string; maxBytes: number }) => + reader({ + ...input, + kind: 'maka.archived_tool_result', + rewriteVersion: 2, + originalEstimatedTokens: 1, + reason: 'stale_tool_result_pruned_before_compact', + }); +} + +/** Verify reconstructibility before committing a replacement; does not write any payload. */ +export function createLedgerArchivePreparer( + evidence: ToolResultArchiveEvidenceReader, +): ToolResultArchiveRecorder { + return async (input) => { + const request = { ...input }; + const loaded = await evidence.read({ + sessionId: request.sessionId, + runtimeEventId: request.runtimeEventId, + }); + if ( + !loaded.ok || + loaded.transitions.length >= TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_TRANSITIONS || + loaded.storedBytes === undefined || + loaded.storedBytes > TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_BYTES - 64 * 1024 + ) + return; + const content = loaded.event.content; + if ( + loaded.event.sessionId !== request.sessionId || + loaded.event.id !== request.runtimeEventId || + content?.kind !== 'function_response' || + content.providerExecuted || + !content.modelProjection || + content.id !== request.toolCallId || + content.name !== request.toolName + ) + return; + const transitions: ModelProjectionTransition[] = []; + for (const event of loaded.transitions) { + const decoded = decodeLedgerTransition(event, request.sessionId); + if (!decoded || decoded.target.runtimeEventId !== request.runtimeEventId) return; + transitions.push(decoded); + } + const reduced = reduceEffectiveModelProjections([loaded.event], transitions); + const previous = reduced.applied.at(-1); + if (previous?.transitionId !== request.previousTransitionId) return; + const source = previous?.replacement ?? content.modelProjection; + if (source.kind === 'json' && isArchivedToolResultPlaceholder(source.value)) return; + if (durableToolResultProjectionDigest(source) !== request.sourceProjectionDigest) return; + const body = serializeToolResultProjectionV1(source); + if ( + body !== request.serializedResult || + Buffer.byteLength(body) !== request.originalBytes || + createHash('sha256').update(body).digest('hex') !== request.bodySha256 + ) + return; + return { ledger: true }; + }; +} diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts index 61e79973dd..10a79aa96c 100644 --- a/packages/runtime/src/runtime-event-read-model.ts +++ b/packages/runtime/src/runtime-event-read-model.ts @@ -558,6 +558,7 @@ export function applyArchivedToolResultReadModelStatuses( toolCallId: placeholder.toolCallId, toolName: placeholder.toolName, artifactId: placeholder.artifactId, + ...(placeholder.rewriteVersion === 2 ? { resourceRef: placeholder.resourceRef } : {}), bodySha256: placeholder.bodySha256, originalEstimatedTokens: placeholder.originalEstimatedTokens, originalBytes: placeholder.originalBytes, @@ -970,6 +971,9 @@ function projectFunctionResponse( toolName: archivedPlaceholder.toolName, artifactId: archivedPlaceholder.artifactId, bodySha256: archivedPlaceholder.bodySha256, + ...(archivedPlaceholder.rewriteVersion === 2 + ? { resourceRef: archivedPlaceholder.resourceRef } + : {}), originalEstimatedTokens: archivedPlaceholder.originalEstimatedTokens, originalBytes: archivedPlaceholder.originalBytes, rewriteVersion: archivedPlaceholder.rewriteVersion, diff --git a/packages/runtime/src/tool-result-archive-capability.ts b/packages/runtime/src/tool-result-archive-capability.ts index 6004c830a1..106dbd8fa6 100644 --- a/packages/runtime/src/tool-result-archive-capability.ts +++ b/packages/runtime/src/tool-result-archive-capability.ts @@ -38,6 +38,7 @@ import type { ArchivedToolResultReason } from './tool-result-archive.js'; import type { ToolResultArchiveReader } from './tool-result-archive.js'; import type { ToolResultArchiveResourceReader } from './tool-result-archive-resource.js'; import type { MakaTool } from './tool-runtime.js'; +import type { ModelProjectionTransition } from '@maka/core/model-projection-transition'; export { ARCHIVE_READ_TOOL_NAME }; @@ -63,14 +64,26 @@ export interface ToolResultArchiveRecorderInput { originalEstimatedTokens: number; rewriteVersion: number; reason: ArchivedToolResultReason; + sourceProjectionDigest?: `sha256:${string}`; + previousTransitionId?: string; } +export type ToolResultArchiveLocation = + | { artifactId: string; ledger?: never } + | { + ledger: true; + artifactId?: never; + commitTransition?: ( + transition: ModelProjectionTransition, + persist: (transition: ModelProjectionTransition) => Promise, + ) => Promise; + }; export type ToolResultArchiveRecorder = ( input: ToolResultArchiveRecorderInput, -) => Promise<{ artifactId: string } | void> | { artifactId: string } | void; +) => Promise | ToolResultArchiveLocation | void; /** Host-owned storage for one archive authority. */ export interface ToolResultArchiveServices { - /** Durably archives a pruned tool result body. */ + /** Persists legacy bytes or verifies ledger reconstruction before the transition commit. */ archiveToolResult: ToolResultArchiveRecorder; /** Replay hydration, addressed by the originating runtime event. */ readToolResultArchive: ToolResultArchiveReader; diff --git a/packages/runtime/src/tool-result-archive-resource.ts b/packages/runtime/src/tool-result-archive-resource.ts index c7134935ed..d3bc63df23 100644 --- a/packages/runtime/src/tool-result-archive-resource.ts +++ b/packages/runtime/src/tool-result-archive-resource.ts @@ -45,15 +45,54 @@ const MAX_SEARCH_PATTERN_CHARS = 256; /** Preview length surfaced by inspect for text/object payloads. */ const INSPECT_PREVIEW_CHARS = 600; -export interface ToolResultArchiveResourceIdentity { +export interface LegacyArchiveResourceIdentity { artifactId: string; + storage?: never; bodySha256: string; originalBytes: number; } -export interface ToolResultArchiveResourceReadInput extends ToolResultArchiveResourceIdentity { +export interface LedgerArchiveResourceIdentity { + storage: 'ledger'; + artifactId?: never; + runtimeEventId: string; + toolCallId: string; + toolName: string; + sourceProjectionDigest: `sha256:${string}`; + previousTransitionId?: string; + bodySha256: string; + originalBytes: number; +} + +export type ToolResultArchiveResourceIdentity = + | LegacyArchiveResourceIdentity + | LedgerArchiveResourceIdentity; +export type ToolResultArchiveResourceReadInput = ToolResultArchiveResourceIdentity & { sessionId: string; maxBytes: number; +}; + +export function isLedgerArchiveIdentity(value: unknown): value is LedgerArchiveResourceIdentity { + if (!isRecord(value)) return false; + return ( + value.storage === 'ledger' && + value.artifactId === undefined && + ['runtimeEventId', 'toolCallId', 'toolName'].every( + (key) => + typeof value[key] === 'string' && + (value[key] as string).length > 0 && + (value[key] as string).length <= 512, + ) && + typeof value.sourceProjectionDigest === 'string' && + /^sha256:[a-f0-9]{64}$/.test(value.sourceProjectionDigest) && + (value.previousTransitionId === undefined || + (typeof value.previousTransitionId === 'string' && + /^mptransition-[a-f0-9]{32}$/.test(value.previousTransitionId))) && + typeof value.bodySha256 === 'string' && + /^[a-f0-9]{64}$/.test(value.bodySha256) && + Number.isSafeInteger(value.originalBytes) && + Number(value.originalBytes) > 0 + ); } export interface ToolResultArchiveResourceReader { @@ -77,6 +116,19 @@ export interface ToolResultArchiveResourceRequest { export function buildToolResultArchiveResourceRef( input: ToolResultArchiveResourceIdentity, ): string { + if (input.storage === 'ledger') { + if (!isLedgerArchiveIdentity(input)) throw new Error('Invalid ledger archive identity'); + const value = [ + input.runtimeEventId, + input.toolCallId, + input.toolName, + input.sourceProjectionDigest, + input.previousTransitionId ?? null, + input.bodySha256, + input.originalBytes, + ]; + return `maka://archive-ledger/v1/${encodeURIComponent(JSON.stringify(value))}`; + } const artifactId = encodeURIComponent(input.artifactId); const sha256 = encodeURIComponent(input.bodySha256); return `maka://archive/${artifactId}/${sha256}/${input.originalBytes}`; @@ -85,6 +137,39 @@ export function buildToolResultArchiveResourceRef( export function parseToolResultArchiveResourceRef( ref: string, ): ToolResultArchiveResourceIdentity | null { + if (ref.startsWith('maka://archive-ledger/')) { + try { + const prefix = 'maka://archive-ledger/v1/'; + if (!ref.startsWith(prefix) || ref.length > 16384) return null; + const values = JSON.parse(decodeURIComponent(ref.slice(prefix.length))); + if (!Array.isArray(values) || values.length !== 7) return null; + const [ + runtimeEventId, + toolCallId, + toolName, + sourceProjectionDigest, + previous, + bodySha256, + originalBytes, + ] = values; + const identity = { + storage: 'ledger' as const, + runtimeEventId, + toolCallId, + toolName, + sourceProjectionDigest, + ...(previous === null ? {} : { previousTransitionId: previous }), + bodySha256, + originalBytes, + }; + return isLedgerArchiveIdentity(identity) && + buildToolResultArchiveResourceRef(identity) === ref + ? identity + : null; + } catch { + return null; + } + } let url: URL; try { url = new URL(ref); @@ -229,7 +314,9 @@ function inspectArchive( kind: 'tool_result_archive', operation: 'inspect', ref, - artifactId: identity.artifactId, + ...(identity.storage === 'ledger' + ? { storage: 'ledger', runtimeEventId: identity.runtimeEventId } + : { artifactId: identity.artifactId }), originalBytes: identity.originalBytes, }; const readHint = diff --git a/packages/runtime/src/tool-result-archive-transition.ts b/packages/runtime/src/tool-result-archive-transition.ts index 50ceb7a06d..dafc89b46e 100644 --- a/packages/runtime/src/tool-result-archive-transition.ts +++ b/packages/runtime/src/tool-result-archive-transition.ts @@ -45,6 +45,7 @@ import type { DurableToolResultProjection } from '@maka/core/durable-tool-result import { DURABLE_TOOL_RESULT_PROJECTION_VERSION } from '@maka/core/durable-tool-result-projection'; import { buildModelProjectionTransition, + durableToolResultProjectionDigest, type ModelProjectionTransition, } from '@maka/core/model-projection-transition'; import type { RuntimeEvent } from '@maka/core/runtime-event'; @@ -57,21 +58,23 @@ import { turnKey, utf8ByteLength, } from './context-budget-helpers.js'; -import { - durableProjectionToToolResultOutput, - projectionArtifactMedia, -} from './durable-tool-result-projection.js'; +import { projectionArtifactMedia } from './durable-tool-result-projection.js'; import { baseToolResultProjection, nextInChain } from './model-projection-transition-ledger.js'; import { ARCHIVED_TOOL_RESULT_REWRITE_VERSION, buildArchivedToolResultPlaceholder, + buildLedgerArchivedToolResultPlaceholder, isArchivedToolResultPlaceholder, - serializeToolResultForArchive, type ArchivedToolResultPlaceholder, type ArchivedToolResultReason, type StaleToolResultArchiveCandidate, type StaleToolResultPrunePolicy, } from './tool-result-archive.js'; +import type { + ToolResultArchiveRecorder, + ToolResultArchiveLocation, +} from './tool-result-archive-capability.js'; +import { serializeToolResultProjectionV1 } from './tool-result-archive-encoding.js'; const DEFAULT_MAX_TOOL_RESULT_ESTIMATED_TOKENS = 2048; @@ -88,10 +91,7 @@ export type ModelProjectionTransitionRecorder = ( * body that is not the one removed from the model's view. */ export function serializedToolResultProjection(projection: DurableToolResultProjection): string { - const output = durableProjectionToToolResultOutput(projection); - return serializeToolResultForArchive( - output.type === 'execution-denied' ? { kind: 'text', text: output.reason ?? '' } : output.value, - ); + return serializeToolResultProjectionV1(projection); } /** The replacement a pruned Tool Result projects to. */ @@ -107,20 +107,7 @@ export function archivedToolResultProjection( export interface ToolResultArchiveTransitionServices { sessionId: string; - archiveToolResult: (input: { - sessionId: string; - runtimeEventId: string; - turnId: string; - toolCallId: string; - toolName: string; - result: unknown; - serializedResult: string; - bodySha256: string; - originalBytes: number; - originalEstimatedTokens: number; - rewriteVersion: typeof ARCHIVED_TOOL_RESULT_REWRITE_VERSION; - reason: ArchivedToolResultReason; - }) => Promise<{ artifactId: string } | void> | { artifactId: string } | void; + archiveToolResult: ToolResultArchiveRecorder; recordTransition: ModelProjectionTransitionRecorder; /** * Re-read the durable ledger after an append. @@ -168,7 +155,7 @@ export async function archiveToolResultAsTransition( request: ToolResultArchiveTransitionRequest, ): Promise { const bodySha256 = sha256(request.serializedResult); - let archived: { artifactId: string } | void; + let archived: ToolResultArchiveLocation | void; try { archived = await Promise.resolve( services.archiveToolResult({ @@ -184,16 +171,22 @@ export async function archiveToolResultAsTransition( originalEstimatedTokens: request.originalEstimatedTokens, rewriteVersion: ARCHIVED_TOOL_RESULT_REWRITE_VERSION, reason: request.reason, + sourceProjectionDigest: durableToolResultProjectionDigest(request.sourceProjection), + ...(request.previousTransitionId + ? { previousTransitionId: request.previousTransitionId } + : {}), }), ); } catch { return undefined; } - const artifactId = archived?.artifactId; - if (typeof artifactId !== 'string' || artifactId.trim().length === 0) return undefined; - - const placeholder = buildArchivedToolResultPlaceholder({ - artifactId, + if (!archived) return undefined; + if ( + !archived.ledger && + (typeof archived.artifactId !== 'string' || archived.artifactId.trim().length === 0) + ) + return undefined; + const common = { runtimeEventId: request.runtimeEventId, toolCallId: request.toolCallId, toolName: request.toolName, @@ -202,7 +195,22 @@ export async function archiveToolResultAsTransition( originalBytes: request.originalBytes, reason: request.reason, ...(request.supersession ? { supersession: request.supersession } : {}), - }); + }; + let placeholder: ArchivedToolResultPlaceholder; + try { + placeholder = archived.ledger + ? buildLedgerArchivedToolResultPlaceholder({ + ...common, + storage: 'ledger', + sourceProjectionDigest: durableToolResultProjectionDigest(request.sourceProjection), + ...(request.previousTransitionId + ? { previousTransitionId: request.previousTransitionId } + : {}), + }) + : buildArchivedToolResultPlaceholder({ ...common, artifactId: archived.artifactId }); + } catch { + return undefined; + } let transition: ModelProjectionTransition; try { @@ -224,7 +232,17 @@ export async function archiveToolResultAsTransition( : {}), now: services.now(), }); - await services.recordTransition(transition); + if ( + placeholder.rewriteVersion === 2 && + Buffer.byteLength(JSON.stringify(transition)) > 32 * 1024 + ) + return undefined; + if (archived.ledger && archived.commitTransition) { + if (!(await archived.commitTransition(transition, services.recordTransition))) + return undefined; + } else { + await services.recordTransition(transition); + } const winner = await winningTransition(services, transition); if (winner && winner.transitionId !== transition.transitionId) { // The rival won. Show what the ledger says, not what this writer wrote; @@ -338,7 +356,7 @@ export function collectReachableArchiveArtifactIds(events: readonly RuntimeEvent for (const event of events) { const content = event.content; if (content?.kind !== 'function_response') continue; - if (isArchivedToolResultPlaceholder(content.result)) { + if (isArchivedToolResultPlaceholder(content.result) && content.result.rewriteVersion === 1) { reachable.add(content.result.artifactId); } } diff --git a/packages/runtime/src/tool-result-archive.ts b/packages/runtime/src/tool-result-archive.ts index 8290aaca5b..faba2ebec8 100644 --- a/packages/runtime/src/tool-result-archive.ts +++ b/packages/runtime/src/tool-result-archive.ts @@ -22,6 +22,8 @@ import { createHash } from 'node:crypto'; import { buildToolResultArchiveResourceRef, TOOL_RESULT_ARCHIVE_READ_INSTRUCTIONS, + isLedgerArchiveIdentity, + type LedgerArchiveResourceIdentity, } from './tool-result-archive-resource.js'; import type { ActiveToolResultSupersession } from './active-tool-result-working-set.js'; @@ -50,7 +52,7 @@ export const ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND = 'maka.archived_tool_result' export const ARCHIVED_TOOL_RESULT_REWRITE_VERSION = 1; -export interface ArchivedToolResultPlaceholder { +export interface LegacyArchivedToolResultPlaceholder { kind: typeof ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND; rewriteVersion: typeof ARCHIVED_TOOL_RESULT_REWRITE_VERSION; artifactId: string; @@ -69,6 +71,40 @@ export interface ArchivedToolResultPlaceholder { supersession?: ActiveToolResultSupersession; } +export type LedgerArchivedToolResultPlaceholder = Omit< + LegacyArchivedToolResultPlaceholder, + 'artifactId' | 'rewriteVersion' +> & + LedgerArchiveResourceIdentity & { rewriteVersion: 2 }; +export type ArchivedToolResultPlaceholder = + | LegacyArchivedToolResultPlaceholder + | LedgerArchivedToolResultPlaceholder; + +export function buildLedgerArchivedToolResultPlaceholder( + input: Omit< + LedgerArchivedToolResultPlaceholder, + 'kind' | 'rewriteVersion' | 'resourceRef' | 'readInstructions' + >, +): LedgerArchivedToolResultPlaceholder { + return { + storage: 'ledger', + runtimeEventId: input.runtimeEventId, + toolCallId: input.toolCallId, + toolName: input.toolName, + sourceProjectionDigest: input.sourceProjectionDigest, + ...(input.previousTransitionId ? { previousTransitionId: input.previousTransitionId } : {}), + bodySha256: input.bodySha256, + originalBytes: input.originalBytes, + originalEstimatedTokens: input.originalEstimatedTokens, + reason: input.reason, + ...(input.supersession ? { supersession: input.supersession } : {}), + kind: ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND, + rewriteVersion: 2, + resourceRef: buildToolResultArchiveResourceRef(input), + readInstructions: TOOL_RESULT_ARCHIVE_READ_INSTRUCTIONS, + }; +} + export interface StaleToolResultArchiveCandidate { runtimeEventId: string; turnId: string; @@ -95,10 +131,10 @@ export type ToolResultArchiveReadFailureReason = | 'size_mismatch' | 'corrupt'; -export interface ToolResultArchiveReaderInput extends ArchivedToolResultPlaceholder { +export type ToolResultArchiveReaderInput = ArchivedToolResultPlaceholder & { sessionId: string; maxBytes?: number; -} +}; export type ToolResultArchiveReadResult = | { ok: true; serializedResult: string } @@ -153,12 +189,14 @@ export function isArchivedToolResultPlaceholder( value: unknown, ): value is ArchivedToolResultPlaceholder { if (!value || typeof value !== 'object') return false; - const candidate = value as Partial; + const candidate = value as Record; + const ledgerIdentity = isLedgerArchiveIdentity(value); return ( candidate.kind === ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND && - candidate.rewriteVersion === ARCHIVED_TOOL_RESULT_REWRITE_VERSION && - typeof candidate.artifactId === 'string' && - candidate.artifactId.length > 0 && + ((candidate.rewriteVersion === 1 && + typeof candidate.artifactId === 'string' && + candidate.artifactId.length > 0) || + (candidate.rewriteVersion === 2 && ledgerIdentity)) && typeof candidate.runtimeEventId === 'string' && candidate.runtimeEventId.length > 0 && typeof candidate.toolCallId === 'string' && @@ -200,6 +238,7 @@ function isValidSupersession(value: unknown): boolean { /** Add the canonical ArchiveRead address to persisted v1 placeholders. */ export function withToolResultArchiveResourceRef(value: unknown): unknown { if (!isArchivedToolResultPlaceholder(value)) return value; + if (value.rewriteVersion === 2) return buildLedgerArchivedToolResultPlaceholder(value); return { ...value, resourceRef: buildToolResultArchiveResourceRef({ @@ -221,7 +260,7 @@ export function buildArchivedToolResultPlaceholder(input: { originalBytes: number; reason: ArchivedToolResultReason; supersession?: ActiveToolResultSupersession; -}): ArchivedToolResultPlaceholder { +}): LegacyArchivedToolResultPlaceholder { return { kind: ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND, rewriteVersion: ARCHIVED_TOOL_RESULT_REWRITE_VERSION, diff --git a/packages/storage/src/tool-result-archive-evidence.ts b/packages/storage/src/tool-result-archive-evidence.ts index 17ac5c6d77..ebfff21cff 100644 --- a/packages/storage/src/tool-result-archive-evidence.ts +++ b/packages/storage/src/tool-result-archive-evidence.ts @@ -114,7 +114,7 @@ export async function openToolResultArchiveEvidenceReader( if (transition.sessionId !== sessionId) return { ok: false, reason: 'corrupt' }; transitions.push(transition); } - return { ok: true, event, transitions }; + return { ok: true, event, transitions, storedBytes: bytes }; }); }); } catch { From db8e5296bd63ff36e5df1f27fde4e29e478a3fd3 Mon Sep 17 00:00:00 2001 From: likun Date: Tue, 8 Sep 2026 09:41:50 +0800 Subject: [PATCH 2/2] fix(runtime): preserve archive projection and copy replay contracts Read only reconstruction fields instead of budgeting raw Tool Result payloads. Rebuild typed ArchiveRead results and dependent transitions in dependency order, then preserve and rebuild valid text checkpoints against final copied content. Cover large MCP image archives, returned-ref reads and checkpoint lineage. Generated-by: Codex --- .../core/src/tool-result-archive-evidence.ts | 1 + .../src/__tests__/execution-artifacts.test.ts | 364 ++++++------ .../execution-model-composition.test.ts | 3 +- .../src/__tests__/conversation-copy.test.ts | 532 +++++++++++++----- packages/runtime/src/conversation-copy.ts | 164 ++++-- .../tool-result-archive-evidence.test.ts | 15 +- .../src/tool-result-archive-evidence.ts | 47 +- 7 files changed, 775 insertions(+), 351 deletions(-) diff --git a/packages/core/src/tool-result-archive-evidence.ts b/packages/core/src/tool-result-archive-evidence.ts index 9f39d96310..f75f1dcd08 100644 --- a/packages/core/src/tool-result-archive-evidence.ts +++ b/packages/core/src/tool-result-archive-evidence.ts @@ -24,6 +24,7 @@ import type { AgentRunEvent } from './agent-run.js'; export type ToolResultArchiveEvidence = | { readonly ok: true; + /** Reconstruction view: required event identity/projection only; raw result is null. */ readonly event: RuntimeEvent; readonly transitions: readonly AgentRunEvent[]; readonly storedBytes?: number; diff --git a/packages/runtime-host/src/__tests__/execution-artifacts.test.ts b/packages/runtime-host/src/__tests__/execution-artifacts.test.ts index ffd4ce9d60..f939e91ec4 100644 --- a/packages/runtime-host/src/__tests__/execution-artifacts.test.ts +++ b/packages/runtime-host/src/__tests__/execution-artifacts.test.ts @@ -26,6 +26,7 @@ import { durableToolResultProjectionDigest, } from '@maka/core/model-projection-transition'; import { buildLedgerArchivedToolResultPlaceholder } from '@maka/runtime/tool-result-archive'; +import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; import { parseToolResultArchiveResourceRef } from '@maka/runtime/tool-result-archive-resource'; import { mkdir, mkdtemp, rm, stat, truncate, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; @@ -37,184 +38,221 @@ import { createReadImageSnapshotPlanner, } from '@maka/storage/artifact-stores'; import { encodeDurableToolResultOutputWithArtifacts } from '@maka/runtime/durable-tool-result-projection'; +import { durableProjectionToToolResultOutput } from '@maka/runtime/durable-tool-result-projection'; import { deferred } from '@maka/core/test-only/async-primitives'; import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; import { createHostExecutionArtifactServices } from '../server/execution-artifacts.js'; import { restoreArtifactV1Shape } from './fixtures/artifact-v1.js'; import { SessionAdmissionGate } from '../server/session-admission-gate.js'; -test('production archive services prepare ledger writes, keep legacy reads, and replay after reopen', async () => { - const root = await mkdtemp(join(tmpdir(), 'maka-ledger-archive-host-')); - const owner = await tryAcquireInteractiveRootOwner( - await resolveStorageRoot({ path: root, kind: 'interactive' }), - ); - assert.ok(owner); - const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); - let evidence = await openToolResultArchiveEvidenceReader(owner.lease); - const db = new DatabaseSync(join(root, 'runtime.sqlite')); - try { - const projection = { version: 1 as const, kind: 'text' as const, text: 'durable ledger body' }; - const serializedResult = JSON.stringify(projection.text); - const bodySha256 = createHash('sha256').update(serializedResult).digest('hex'); - const event = { - id: 'response', - sessionId: 'session', - runId: 'run', - invocationId: 'invocation', - turnId: 'turn', - ts: 1, - partial: false, - author: 'tool', - role: 'tool', - content: { - kind: 'function_response', - id: 'call', - name: 'Read', - result: 'raw execution body', - modelProjection: projection, - }, - }; - db.prepare( - 'INSERT INTO runtime_events(event_id, session_id, invocation_id, run_id, turn_id, event_seq, event_kind, payload_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', - ).run( - 'response', - 'session', - 'invocation', - 'run', - 'turn', - 1, - 'function_response', - JSON.stringify(event), - 1, - ); - const old = await artifacts.create({ - id: 'legacy-archive', - sessionId: 'session', - turnId: 'turn', - name: 'legacy.json', - kind: 'file', - content: serializedResult, - source: 'tool_result_archive', - }); - const input = { - sessionId: 'session', - runtimeEventId: 'response', - turnId: 'turn', - toolCallId: 'call', - toolName: 'Read', - serializedResult, - bodySha256, - originalBytes: Buffer.byteLength(serializedResult), - originalEstimatedTokens: 10, - rewriteVersion: 1, - sourceProjectionDigest: durableToolResultProjectionDigest(projection), - reason: 'stale_tool_result_pruned_before_compact' as const, - }; - const make = () => - createHostExecutionArtifactServices({ - artifacts, - archiveEvidence: evidence, - sessionAdmission: new SessionAdmissionGate(), - sessions: { probeSessionRemoval: async () => ({ kind: 'present' }) }, - requestDrain: () => assert.fail('archive failure must not drain'), - }); - let services = make(); - const prepared = await services.toolResultArchive.services.archiveToolResult(input); - assert.ok(prepared?.ledger); - assert.equal(typeof prepared.commitTransition, 'function'); - assert.equal((await artifacts.listPage('session', { offset: 0, limit: 10 })).total, 1); - const placeholder = buildLedgerArchivedToolResultPlaceholder({ ...input, storage: 'ledger' }); - const transition = buildModelProjectionTransition({ - sessionId: 'session', - target: { - runtimeEventId: 'response', - part: 'tool_result', - toolCallId: 'call', - toolName: 'Read', - }, - sourceProjection: projection, - replacement: { version: 1, kind: 'json', value: placeholder as never }, - now: 2, - }); - db.prepare('INSERT INTO core_agent_runs(session_id, run_id, created_at) VALUES (?, ?, ?)').run( - 'session', - 'run', - 1, +for (const largeImage of [false, true]) { + test(`production archives survive reopen (large raw MCP image: ${largeImage})`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-ledger-archive-host-')); + const owner = await tryAcquireInteractiveRootOwner( + await resolveStorageRoot({ path: root, kind: 'interactive' }), ); - const persist = async () => { - db.prepare('INSERT INTO core_agent_run_events VALUES (?, ?, ?, ?, ?, ?, ?)').run( + assert.ok(owner); + const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); + let evidence = await openToolResultArchiveEvidenceReader(owner.lease); + const db = new DatabaseSync(join(root, 'runtime.sqlite')); + try { + const projection: DurableToolResultProjection = largeImage + ? { + version: 1, + kind: 'content', + parts: [ + { + kind: 'artifact', + mediaType: 'image/png', + ref: { kind: 'session_file', sessionId: 'session', relativePath: 'mcp-image' }, + }, + ], + } + : { version: 1, kind: 'text', text: 'durable ledger body' }; + const output = durableProjectionToToolResultOutput(projection); + assert.ok('value' in output); + const serializedResult = JSON.stringify(output.value); + const bodySha256 = createHash('sha256').update(serializedResult).digest('hex'); + const event = { + id: 'response', + sessionId: 'session', + runId: 'run', + invocationId: 'invocation', + turnId: 'turn', + ts: 1, + partial: false, + author: 'tool', + role: 'tool', + content: { + kind: 'function_response', + id: 'call', + name: 'Read', + result: largeImage + ? { + content: [ + { + type: 'image', + mimeType: 'image/png', + data: Buffer.alloc(2 * 1024 * 1024).toString('base64'), + }, + ], + } + : 'raw execution body', + modelProjection: projection, + }, + }; + db.prepare( + 'INSERT INTO runtime_events(event_id, session_id, invocation_id, run_id, turn_id, event_seq, event_kind, payload_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', + ).run( + 'response', 'session', + 'invocation', 'run', + 'turn', + 1, + 'function_response', + JSON.stringify(event), 1, - transition.transitionId, - 'model_projection_transition_recorded', - 2, - JSON.stringify({ - id: transition.transitionId, - type: 'model_projection_transition_recorded', - sessionId: 'session', - runId: 'run', - turnId: 'turn', - ts: 2, - data: { runtimeEventId: 'response', part: 'tool_result', transition }, - }), ); - }; - assert.equal(await prepared.commitTransition!(transition, persist), true); - assert.equal( - await prepared.commitTransition!(transition, async () => - assert.fail('stale preparation must not append'), - ), - false, - ); - evidence.close(); - evidence = await openToolResultArchiveEvidenceReader(owner.lease); - services = make(); - assert.deepEqual( - await services.toolResultArchive.services.readToolResultArchive({ - ...placeholder, + const projectedEvidence = await evidence.read({ sessionId: 'session', - }), - { ok: true, serializedResult }, - ); - assert.equal( - ( - await services.toolResultArchive.services.readToolResultArchive({ - ...placeholder, - sessionId: 'other', - }) - ).ok, - false, - ); - const identity = parseToolResultArchiveResourceRef(placeholder.resourceRef!); - assert.ok(identity); - assert.deepEqual( - await services.toolResultArchive.services.readArchivedToolResultResource({ - ...identity, + runtimeEventId: 'response', + }); + assert.ok(projectedEvidence.ok); + assert.ok(projectedEvidence.storedBytes! < 4096); + assert.equal( + projectedEvidence.event.content?.kind === 'function_response' + ? projectedEvidence.event.content.result + : undefined, + null, + ); + const old = await artifacts.create({ + id: 'legacy-archive', sessionId: 'session', - maxBytes: input.originalBytes, - }), - { ok: true, serializedResult }, - ); - assert.deepEqual( - await services.toolResultArchive.services.readArchivedToolResultResource({ - artifactId: old.id, + turnId: 'turn', + name: 'legacy.json', + kind: 'file', + content: serializedResult, + source: 'tool_result_archive', + }); + const input = { + sessionId: 'session', + runtimeEventId: 'response', + turnId: 'turn', + toolCallId: 'call', + toolName: 'Read', + serializedResult, bodySha256, - originalBytes: input.originalBytes, + originalBytes: Buffer.byteLength(serializedResult), + originalEstimatedTokens: 10, + rewriteVersion: 1, + sourceProjectionDigest: durableToolResultProjectionDigest(projection), + reason: 'stale_tool_result_pruned_before_compact' as const, + }; + const make = () => + createHostExecutionArtifactServices({ + artifacts, + archiveEvidence: evidence, + sessionAdmission: new SessionAdmissionGate(), + sessions: { probeSessionRemoval: async () => ({ kind: 'present' }) }, + requestDrain: () => assert.fail('archive failure must not drain'), + }); + let services = make(); + const prepared = await services.toolResultArchive.services.archiveToolResult(input); + assert.ok(prepared?.ledger); + assert.equal(typeof prepared.commitTransition, 'function'); + assert.equal((await artifacts.listPage('session', { offset: 0, limit: 10 })).total, 1); + const placeholder = buildLedgerArchivedToolResultPlaceholder({ ...input, storage: 'ledger' }); + const transition = buildModelProjectionTransition({ sessionId: 'session', - maxBytes: input.originalBytes, - }), - { ok: true, serializedResult }, - ); - } finally { - db.close(); - evidence.close(); - artifacts.close(); - await owner.close(); - await rm(root, { recursive: true, force: true }); - await rm(owner.controlDirectory, { recursive: true, force: true }); - } -}); + target: { + runtimeEventId: 'response', + part: 'tool_result', + toolCallId: 'call', + toolName: 'Read', + }, + sourceProjection: projection, + replacement: { version: 1, kind: 'json', value: placeholder as never }, + now: 2, + }); + db.prepare( + 'INSERT INTO core_agent_runs(session_id, run_id, created_at) VALUES (?, ?, ?)', + ).run('session', 'run', 1); + const persist = async () => { + db.prepare('INSERT INTO core_agent_run_events VALUES (?, ?, ?, ?, ?, ?, ?)').run( + 'session', + 'run', + 1, + transition.transitionId, + 'model_projection_transition_recorded', + 2, + JSON.stringify({ + id: transition.transitionId, + type: 'model_projection_transition_recorded', + sessionId: 'session', + runId: 'run', + turnId: 'turn', + ts: 2, + data: { runtimeEventId: 'response', part: 'tool_result', transition }, + }), + ); + }; + assert.equal(await prepared.commitTransition!(transition, persist), true); + assert.equal( + await prepared.commitTransition!(transition, async () => + assert.fail('stale preparation must not append'), + ), + false, + ); + evidence.close(); + evidence = await openToolResultArchiveEvidenceReader(owner.lease); + services = make(); + assert.deepEqual( + await services.toolResultArchive.services.readToolResultArchive({ + ...placeholder, + sessionId: 'session', + }), + { ok: true, serializedResult }, + ); + assert.equal( + ( + await services.toolResultArchive.services.readToolResultArchive({ + ...placeholder, + sessionId: 'other', + }) + ).ok, + false, + ); + const identity = parseToolResultArchiveResourceRef(placeholder.resourceRef!); + assert.ok(identity); + assert.deepEqual( + await services.toolResultArchive.services.readArchivedToolResultResource({ + ...identity, + sessionId: 'session', + maxBytes: input.originalBytes, + }), + { ok: true, serializedResult }, + ); + assert.deepEqual( + await services.toolResultArchive.services.readArchivedToolResultResource({ + artifactId: old.id, + bodySha256, + originalBytes: input.originalBytes, + sessionId: 'session', + maxBytes: input.originalBytes, + }), + { ok: true, serializedResult }, + ); + } finally { + db.close(); + evidence.close(); + artifacts.close(); + await owner.close(); + await rm(root, { recursive: true, force: true }); + await rm(owner.controlDirectory, { recursive: true, force: true }); + } + }); +} test('a refused projection preserves a shared image until Session cleanup', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-shared-projection-')); diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts index 6284bd75d8..915dbe5803 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -149,7 +149,8 @@ const MAX_IMPLEMENTATION_CHILD_REQUESTS = const HEADLESS_CODING_V1_PROMPT_HASH = 'sha256:b2773282ac4755dc8d8a663eafdec68c3fa6f5680ec8557d261b5f723672b467'; const HEADLESS_CODING_V1_TOOLS_HASH = - 'sha256:aa3ab56a7b67dde133fffe885f4def81735c93015202e31ecb339a84863f6d03'; + // ArchiveRead now describes both ledger and legacy resource references. + 'sha256:22809de022f9c46186cae986eda23438efe9dbe6856b57abb0613ea48b51ad9c'; const execFileAsync = promisify(execFile); test('backend creation resolves a bound Session by immutable Connection identity', async () => { let observedRef: unknown; diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 2e64f664bb..139f129571 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -18,7 +18,10 @@ */ import assert from 'node:assert/strict'; -import { createLedgerToolResultArchiveReader } from '../ledger-tool-result-archive-reader.js'; +import { + createLedgerToolResultArchiveReader, + createLedgerArchiveResourceReader, +} from '../ledger-tool-result-archive-reader.js'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -64,10 +67,16 @@ import { matchHistoryCompactCheckpointPrefix, validateHistoryCompactCheckpointShape, } from '../history-compact-checkpoint.js'; -import { isHistoryCompactContentEvent } from '../history-compaction.js'; +import { + isHistoryCompactContentEvent, + applyRuntimeEventHistoryCompact, +} from '../history-compaction.js'; import { RuntimeReadModel, type RuntimeReadModelSessionView } from '../runtime-read-model.js'; import { buildToolOperationId } from '../runtime-commit-sink.js'; -import { buildToolResultArchiveResourceRef } from '../tool-result-archive-resource.js'; +import { + buildToolResultArchiveResourceRef, + readToolResultArchiveResource, +} from '../tool-result-archive-resource.js'; import { sha256 } from '../context-budget-helpers.js'; import { baseToolResultProjection, @@ -2798,147 +2807,388 @@ function sourceProjectionTransition(input: { }); } -test('conversation copy rebuilds ledger archive hashes after remapping image refs', async () => { - const root = await mkdtemp(join(tmpdir(), 'maka-ledger-copy-')); - try { - const runStore = createSqliteAgentRunStore(root); - const runtimeEventStore = createWorkspaceRuntimeStore(root); - await seedRun(runtimeEventStore, { - runId: 'run-source', - invocationId: 'invocation-source', - turnId: 'turn-1', - cwd: root, - }); - const imageRef = { - kind: 'session_context' as const, - sessionId: 'session-source', - refId: 'image-source', - }; - const projection: DurableToolResultProjection = { - version: 1, - kind: 'content', - parts: [ - { kind: 'text', text: 'caption' }, - { kind: 'artifact', mediaType: 'image/png', ref: imageRef }, - ], - }; - const result = runtimeEvent({ - id: 'event-result-ledger', - ts: 2, - role: 'tool', - author: 'tool', - content: { - kind: 'function_response', - id: 'tool-1', - name: 'Read', - result: { kind: 'image', mimeType: 'image/png', ref: imageRef }, - modelProjection: projection, - }, - }); - for (const event of [ - runtimeEvent({ - id: 'event-user', - role: 'user', - author: 'user', - content: { kind: 'text', text: 'copy image' }, - }), - runtimeEvent({ - id: 'event-call', - ts: 1.5, - role: 'model', - author: 'agent', - content: { kind: 'function_call', id: 'tool-1', name: 'Read', args: { path: 'image.png' } }, - }), - result, - runtimeEvent({ id: 'event-terminal', ts: 3, status: 'completed' }), - ]) - await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); - const serialized = serializedToolResultProjection(projection); - const placeholder = buildLedgerArchivedToolResultPlaceholder({ - storage: 'ledger', - runtimeEventId: result.id, - toolCallId: 'tool-1', - toolName: 'Read', - sourceProjectionDigest: durableToolResultProjectionDigest(projection), - bodySha256: sha256(serialized), - originalBytes: Buffer.byteLength(serialized), - originalEstimatedTokens: 100, - reason: 'stale_tool_result_pruned_before_compact', - }); - const transition = buildModelProjectionTransition({ - sessionId: 'session-source', - target: { +for (const inspectFirst of [false, true]) { + test(`conversation copy retains archive reference closure and checkpoints (inspect: ${inspectFirst})`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-ledger-copy-')); + try { + const runStore = createSqliteAgentRunStore(root); + const runtimeEventStore = createWorkspaceRuntimeStore(root); + await seedRun(runtimeEventStore, { + runId: 'run-source', + invocationId: 'invocation-source', + turnId: 'turn-1', + cwd: root, + }); + const imageRef = { + kind: 'session_context' as const, + sessionId: 'session-source', + refId: 'image-source', + }; + const projection: DurableToolResultProjection = { + version: 1, + kind: 'content', + parts: [ + { kind: 'text', text: 'caption' }, + { kind: 'artifact', mediaType: 'image/png', ref: imageRef }, + ], + }; + const result = runtimeEvent({ + id: 'event-result-ledger', + ts: 2, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-1', + name: 'Read', + result: { kind: 'image', mimeType: 'image/png', ref: imageRef }, + modelProjection: projection, + }, + }); + for (const event of [ + runtimeEvent({ + id: 'event-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'copy image' }, + }), + runtimeEvent({ + id: 'event-call', + ts: 1.5, + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'tool-1', + name: 'Read', + args: { path: 'image.png' }, + }, + }), + result, + ]) + await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); + const serialized = serializedToolResultProjection(projection); + const placeholder = buildLedgerArchivedToolResultPlaceholder({ + storage: 'ledger', runtimeEventId: result.id, - part: 'tool_result', toolCallId: 'tool-1', toolName: 'Read', - }, - sourceProjection: projection, - replacement: archivedToolResultProjection(placeholder), - now: 4, - }); - await runStore.appendEvent('session-source', 'run-source', { - type: MODEL_PROJECTION_TRANSITION_EVENT_TYPE, - id: transition.transitionId, - runId: 'run-source', - sessionId: 'session-source', - turnId: 'turn-1', - ts: 4, - data: { runtimeEventId: result.id, part: 'tool_result', transition }, - }); - const source = await new RuntimeReadModel({ runtimeEventStore }).getSessionView( - 'session-source', - ); - await cloneConversationRuntimeLedger({ - plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), - copiedMessages: source.messages, - referenceMap: { - mode: 'exact', - linkedChildren: { mode: 'reject' }, - sourceSessionId: 'session-source', - targetSessionId: 'session-target', - artifactIds: new Map(), - relativePaths: new Map(), - contextRefs: new Map([['image-source', 'image-target']]), - }, - runStore, - runtimeEventStore, - newId: () => crypto.randomUUID(), - }); - const [targetRun] = await runtimeEventStore.listSessionInvocations('session-target'); - const targetEvents = await runtimeEventStore.readRuntimeEvents( - 'session-target', - targetRun!.runId, - ); - const target = targetEvents.find((event) => event.content?.kind === 'function_response')!; - const records = await runStore.readEvents('session-target', targetRun!.runId); - const transitions = await loadModelProjectionTransitionsFromRunLedger( - runStore, - 'session-target', - [targetRun!.runId], - ); - const copied = transitions.transitions[0]!; - assert.ok(copied.replacement.kind === 'json'); - assert.ok(isArchivedToolResultPlaceholder(copied.replacement.value)); - const copiedPlaceholder = copied.replacement.value as ArchivedToolResultPlaceholder; - assert.equal(copiedPlaceholder.rewriteVersion, 2); - assert.notEqual(copiedPlaceholder.bodySha256, placeholder.bodySha256); - assert.notEqual(copiedPlaceholder.resourceRef, placeholder.resourceRef); - const read = createLedgerToolResultArchiveReader({ - read: async () => ({ - ok: true, - event: target, - transitions: records.filter((row) => row.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE), - }), - }); - const body = await read({ ...copiedPlaceholder, sessionId: 'session-target' }); - assert.ok(body.ok); - assert.match(body.serializedResult, /image-target/); - assert.doesNotMatch(body.serializedResult, /image-source/); - } finally { - await rm(root, { recursive: true, force: true }); - } -}); + sourceProjectionDigest: durableToolResultProjectionDigest(projection), + bodySha256: sha256(serialized), + originalBytes: Buffer.byteLength(serialized), + originalEstimatedTokens: 100, + reason: 'stale_tool_result_pruned_before_compact', + }); + const transition = buildModelProjectionTransition({ + sessionId: 'session-source', + target: { + runtimeEventId: result.id, + part: 'tool_result', + toolCallId: 'tool-1', + toolName: 'Read', + }, + sourceProjection: projection, + replacement: archivedToolResultProjection(placeholder), + now: 4, + }); + await runStore.appendEvent('session-source', 'run-source', { + type: MODEL_PROJECTION_TRANSITION_EVENT_TYPE, + id: transition.transitionId, + runId: 'run-source', + sessionId: 'session-source', + turnId: 'turn-1', + ts: 4, + data: { runtimeEventId: result.id, part: 'tool_result', transition }, + }); + const sourceRecords = await runStore.readEvents('session-source', 'run-source'); + if (inspectFirst) { + const resource = createLedgerArchiveResourceReader({ + read: async () => ({ ok: true, event: result, transitions: sourceRecords }), + }); + const inspected = await readToolResultArchiveResource( + { + readArchivedToolResultResource: (input) => { + assert.equal(input.storage, 'ledger'); + return input.storage === 'ledger' + ? resource(input) + : { ok: false, reason: 'not_found' }; + }, + }, + 'session-source', + { ref: placeholder.resourceRef!, operation: 'inspect' }, + ); + await runtimeEventStore.appendRuntimeEvent( + 'session-source', + 'run-source', + runtimeEvent({ + id: 'archive-call', + ts: 5, + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'archive-tool', + name: 'ArchiveRead', + args: { ref: placeholder.resourceRef, operation: 'inspect' }, + }, + }), + ); + await runtimeEventStore.appendRuntimeEvent( + 'session-source', + 'run-source', + runtimeEvent({ + id: 'archive-result', + ts: 6, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'archive-tool', + name: 'ArchiveRead', + result: { kind: 'json', value: inspected }, + modelProjection: { version: 1, kind: 'json', value: inspected as never }, + }, + }), + ); + const inspectedProjection: DurableToolResultProjection = { + version: 1, + kind: 'json', + value: inspected as never, + }; + const inspectedBody = serializedToolResultProjection(inspectedProjection); + const dependentPlaceholder = buildLedgerArchivedToolResultPlaceholder({ + storage: 'ledger', + runtimeEventId: 'archive-result', + toolCallId: 'archive-tool', + toolName: 'ArchiveRead', + sourceProjectionDigest: durableToolResultProjectionDigest(inspectedProjection), + bodySha256: sha256(inspectedBody), + originalBytes: Buffer.byteLength(inspectedBody), + originalEstimatedTokens: 100, + reason: 'stale_tool_result_pruned_before_compact', + }); + const dependent = buildModelProjectionTransition({ + sessionId: 'session-source', + target: { + runtimeEventId: 'archive-result', + part: 'tool_result', + toolCallId: 'archive-tool', + toolName: 'ArchiveRead', + }, + sourceProjection: inspectedProjection, + replacement: archivedToolResultProjection(dependentPlaceholder), + now: 7, + }); + await runStore.appendEvent('session-source', 'run-source', { + type: MODEL_PROJECTION_TRANSITION_EVENT_TYPE, + id: dependent.transitionId, + sessionId: 'session-source', + runId: 'run-source', + turnId: 'turn-1', + ts: 7, + data: { runtimeEventId: 'archive-result', part: 'tool_result', transition: dependent }, + }); + } + await runtimeEventStore.appendRuntimeEvent( + 'session-source', + 'run-source', + runtimeEvent({ id: 'event-terminal', ts: 7, status: 'completed' }), + ); + const compactable = ( + await runtimeEventStore.readRuntimeEvents('session-source', 'run-source') + ).filter(isHistoryCompactContentEvent); + const checkpoint = buildHistoryCompactCheckpoint({ + sessionId: 'session-source', + coveredRuntimeEvents: inspectFirst ? compactable : compactable.slice(0, 1), + summary: sectionedSummary( + inspectFirst ? `Use ${placeholder.resourceRef}` : 'Unrelated user request summary.', + ), + highWaterName: 'copy-test', + highWaterSeq: 1, + now: 8, + }); + await runStore.appendEvent('session-source', 'run-source', { + type: 'history_compact_checkpoint_recorded', + id: 'archive-checkpoint', + sessionId: 'session-source', + runId: 'run-source', + turnId: 'turn-1', + ts: 8, + data: { checkpoint }, + }); + const successor = buildHistoryCompactCheckpoint({ + sessionId: 'session-source', + coveredRuntimeEvents: compactable, + summary: sectionedSummary(`Continue with ${placeholder.resourceRef}`), + highWaterName: 'copy-test', + highWaterSeq: 2, + now: 9, + previousCheckpointId: checkpoint.checkpointId, + }); + await runStore.appendEvent('session-source', 'run-source', { + type: 'history_compact_checkpoint_recorded', + id: 'archive-checkpoint-next', + sessionId: 'session-source', + runId: 'run-source', + turnId: 'turn-1', + ts: 9, + data: { checkpoint: successor }, + }); + const source = await new RuntimeReadModel({ runtimeEventStore }).getSessionView( + 'session-source', + ); + await cloneConversationRuntimeLedger({ + plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), + copiedMessages: source.messages, + referenceMap: { + mode: 'exact', + linkedChildren: { mode: 'reject' }, + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map(), + relativePaths: new Map(), + contextRefs: new Map([['image-source', 'image-target']]), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const [targetRun] = await runtimeEventStore.listSessionInvocations('session-target'); + const targetEvents = await runtimeEventStore.readRuntimeEvents( + 'session-target', + targetRun!.runId, + ); + const target = targetEvents.find((event) => event.content?.kind === 'function_response')!; + const records = await runStore.readEvents('session-target', targetRun!.runId); + const transitions = await loadModelProjectionTransitionsFromRunLedger( + runStore, + 'session-target', + [targetRun!.runId], + ); + const copied = transitions.transitions[0]!; + assert.ok(copied.replacement.kind === 'json'); + assert.ok(isArchivedToolResultPlaceholder(copied.replacement.value)); + const copiedPlaceholder = copied.replacement.value as ArchivedToolResultPlaceholder; + assert.equal(copiedPlaceholder.rewriteVersion, 2); + assert.notEqual(copiedPlaceholder.bodySha256, placeholder.bodySha256); + assert.notEqual(copiedPlaceholder.resourceRef, placeholder.resourceRef); + const read = createLedgerToolResultArchiveReader({ + read: async () => ({ + ok: true, + event: target, + transitions: records.filter( + (row) => + row.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE && + row.data?.runtimeEventId === target.id, + ), + }), + }); + const body = await read({ ...copiedPlaceholder, sessionId: 'session-target' }); + assert.ok(body.ok); + assert.match(body.serializedResult, /image-target/); + assert.doesNotMatch(body.serializedResult, /image-source/); + if (inspectFirst) { + const response = targetEvents.find( + (event) => + event.content?.kind === 'function_response' && event.content.name === 'ArchiveRead', + ); + assert.ok(response?.content?.kind === 'function_response'); + const output = ( + response.content.result as { value: { ref: string; runtimeEventId: string } } + ).value; + assert.equal(output.ref, copiedPlaceholder.resourceRef); + assert.equal(output.runtimeEventId, target.id); + assert.equal( + response.content.modelProjection?.kind === 'json' + ? (response.content.modelProjection.value as { ref: string }).ref + : undefined, + output.ref, + ); + const resource = createLedgerArchiveResourceReader({ + read: async () => ({ + ok: true, + event: target, + transitions: records.filter( + (row) => + row.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE && + row.data?.runtimeEventId === target.id, + ), + }), + }); + const reread = await readToolResultArchiveResource( + { + readArchivedToolResultResource: (input) => + input.storage === 'ledger' ? resource(input) : { ok: false, reason: 'not_found' }, + }, + 'session-target', + { ref: output.ref, operation: 'read' }, + ); + assert.equal((reread as { ok: boolean }).ok, true); + const dependent = transitions.transitions.find( + (transition) => transition.target.runtimeEventId === response.id, + )!; + assert.ok(dependent?.replacement.kind === 'json'); + assert.ok(isArchivedToolResultPlaceholder(dependent.replacement.value)); + const dependentReader = createLedgerToolResultArchiveReader({ + read: async () => ({ + ok: true, + event: response, + transitions: records.filter( + (record) => + record.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE && + record.data?.runtimeEventId === response.id, + ), + }), + }); + const dependentBody = await dependentReader({ + ...(dependent.replacement.value as ArchivedToolResultPlaceholder), + sessionId: 'session-target', + }); + assert.ok(dependentBody.ok); + assert.equal(JSON.parse(dependentBody.serializedResult).ref, copiedPlaceholder.resourceRef); + } + const checkpointEvent = records.find( + (record) => record.type === 'history_compact_checkpoint_recorded', + ); + const copiedCheckpoint = checkpointEvent?.data?.checkpoint; + assert.ok(validateHistoryCompactCheckpointShape(copiedCheckpoint, 'session-target')); + assert.equal(copiedCheckpoint.version, 2); + assert.equal( + matchHistoryCompactCheckpointPrefix( + copiedCheckpoint, + targetEvents.filter(isHistoryCompactContentEvent), + ).reason, + undefined, + ); + const replay = applyRuntimeEventHistoryCompact(targetEvents, { + enabled: true, + checkpoint: copiedCheckpoint, + }); + assert.notDeepEqual(replay.events, targetEvents); + if (copiedCheckpoint.version === 2) + assert.equal( + copiedCheckpoint.summary, + inspectFirst + ? sectionedSummary(`Use ${copiedPlaceholder.resourceRef}`) + : checkpoint.summary, + ); + const copiedSuccessor = records.filter( + (record) => record.type === 'history_compact_checkpoint_recorded', + )[1]?.data?.checkpoint; + assert.ok(validateHistoryCompactCheckpointShape(copiedSuccessor, 'session-target')); + assert.equal(copiedSuccessor.previousCheckpointId, copiedCheckpoint.checkpointId); + assert.equal( + matchHistoryCompactCheckpointPrefix( + copiedSuccessor, + targetEvents.filter(isHistoryCompactContentEvent), + ).reason, + undefined, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); +} test('conversation copy rebuilds projection transitions against the copied events', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-conversation-transition-copy-')); diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 8db580dc3b..5662a681be 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -576,6 +576,70 @@ export async function cloneConversationRuntimeLedger( string, { projection: DurableToolResultProjection; transitionId: string } >(); + const clonedTransitions = new Map(); + const visiting = new Set(); + const finished = new Set(); + const transitionsByTarget = new Map(); + for (const plan of flattenedPlans) { + for (const record of plan.operationalEvents) { + if (record.type !== MODEL_PROJECTION_TRANSITION_EVENT_TYPE) continue; + const transition = decodeModelProjectionTransition(record.data?.transition, record.sessionId); + const group = transitionsByTarget.get(transition.target.runtimeEventId) ?? []; + group.push(record); + transitionsByTarget.set(transition.target.runtimeEventId, group); + } + } + // ArchiveRead can itself be archived. Resolve its referenced target first, + // then rebuild this result's transitions against its final copied projection. + const finishTarget = (sourceId: string): void => { + if (finished.has(sourceId)) return; + if (visiting.has(sourceId)) throw new Error('Cyclic copied archive reference'); + visiting.add(sourceId); + const target = clonedEventBySourceId.get(sourceId); + if (target?.content?.kind === 'function_response' && target.content.name === 'ArchiveRead') { + const resolveResult = (value: unknown): unknown => + rewriteArchiveReadResult(value, references, (ref) => { + const identity = parseToolResultArchiveResourceRef(ref); + if (identity?.storage === 'ledger' && clonedEventBySourceId.has(identity.runtimeEventId)) + finishTarget(identity.runtimeEventId); + }); + target.content.result = resolveResult(target.content.result); + const projection = target.content.modelProjection; + if (projection?.kind === 'json') + target.content.modelProjection = { + ...projection, + value: resolveResult(projection.value) as typeof projection.value, + }; + } + for (const record of transitionsByTarget.get(sourceId) ?? []) { + clonedTransitions.set( + record, + cloneModelProjectionTransition( + record, + references, + clonedEventBySourceId, + transitionIds, + transitionState, + ), + ); + } + visiting.delete(sourceId); + finished.add(sourceId); + }; + for (const sourceId of clonedEventBySourceId.keys()) finishTarget(sourceId); + for (const event of clonedEventBySourceId.values()) { + if (event.content?.kind === 'text') + event.content.text = rewriteLedgerArchiveText(event.content.text, references); + if (event.content?.kind === 'function_call' && event.content.name === 'ArchiveRead') { + const args = event.content.args; + if (args && typeof args === 'object' && 'ref' in args && typeof args.ref === 'string') { + event.content = { + ...event.content, + args: { ...args, ref: rewriteLedgerArchiveText(args.ref, references) }, + }; + } + } + } const preparedPlans = flattenedPlans.map((plan) => { const runId = runIds.get(plan.run.runId)!; const clonedOperationalEvents = plan.operationalEvents.flatMap((event) => { @@ -590,10 +654,9 @@ export async function cloneConversationRuntimeLedger( sourceCompactableEvents.get(plan.run.runId) ?? [], clonedEventBySourceId, checkpointIds, - transitionIds, - transitionState, providerTraceIds, logicalCallIds, + clonedTransitions, ); return clonedEvent ? [clonedEvent] : []; }); @@ -611,37 +674,30 @@ export async function cloneConversationRuntimeLedger( terminalEvent, }; }); - const copiedMessages = input.copiedMessages.map((message) => - rewriteConversationCopyMessage(message, references), + const archiveReadCalls = new Set( + flattenedPlans.flatMap((plan) => + plan.events.flatMap((event) => + event.content?.kind === 'function_response' && event.content.name === 'ArchiveRead' + ? [`${event.turnId}:${event.content.id}`] + : [], + ), + ), ); - - const rewriteLedgerText = (text: string): string => { - for (const [source, target] of references.ledgerArchives ?? []) - text = text.split(source).join(target.resourceRef!); - return text; - }; - for (const message of copiedMessages) { - if (message.type === 'user' || message.type === 'assistant') - message.text = rewriteLedgerText(message.text); - } - for (const event of clonedEventBySourceId.values()) { - if (event.content?.kind === 'text') event.content.text = rewriteLedgerText(event.content.text); - if (event.content?.kind === 'function_call' && event.content.name === 'ArchiveRead') { - const args = event.content.args; - if (args && typeof args === 'object' && 'ref' in args && typeof args.ref === 'string') { - event.content.args = { ...args, ref: rewriteLedgerText(args.ref) }; - } - } - } - if (references.ledgerArchives?.size) { - // Frozen summaries may embed source archive addresses. Keep the copied raw - // ledger and transitions; the target can compact again with its own refs. - for (const plan of preparedPlans) { - plan.clonedOperationalEvents = plan.clonedOperationalEvents.filter( - (event) => event.type !== 'history_compact_checkpoint_recorded', - ); + const copiedMessages = input.copiedMessages.map((message) => { + const copied = rewriteConversationCopyMessage(message, references); + if (copied.type === 'user' || copied.type === 'assistant') + return { ...copied, text: rewriteLedgerArchiveText(copied.text, references) }; + if ( + copied.type === 'tool_result' && + archiveReadCalls.has(`${copied.turnId}:${copied.toolUseId}`) + ) { + return { + ...copied, + content: rewriteArchiveReadResult(copied.content, references) as ToolResultContent, + }; } - } + return copied; + }); const importedSourceEventIds = new Set(); const orderedBatches = input.plan.inlineRuntimeEvents.flatMap((event) => { @@ -696,6 +752,39 @@ export async function cloneConversationRuntimeLedger( }; } +function rewriteLedgerArchiveText( + text: string, + references: ConversationCopyMessageReferenceMap, +): string { + for (const [source, target] of references.ledgerArchives ?? []) + text = text.split(source).join(target.resourceRef!); + return text; +} + +/** Only ArchiveRead's defined envelope is rewritten, never its opaque content/items. */ +function rewriteArchiveReadResult( + value: unknown, + references: ConversationCopyMessageReferenceMap, + before?: (ref: string) => void, +): unknown { + if (!value || typeof value !== 'object' || Array.isArray(value)) return value; + const record = value as Record; + if (record.kind === 'json') + return { ...record, value: rewriteArchiveReadResult(record.value, references, before) }; + if (record.kind !== 'tool_result_archive' || typeof record.ref !== 'string') return value; + before?.(record.ref); + const mapped = references.ledgerArchives?.get(record.ref); + if (!mapped) return value; + return { + ...record, + ref: mapped.resourceRef, + ...(record.storage === 'ledger' && typeof record.runtimeEventId === 'string' + ? { runtimeEventId: mapped.runtimeEventId } + : {}), + ...(typeof record.originalBytes === 'number' ? { originalBytes: mapped.originalBytes } : {}), + }; +} + interface ConversationCopyRunEvents { readonly run: RuntimeInvocationRecord; /** The run's events, beginning with its opening. */ @@ -908,10 +997,9 @@ function cloneAgentRunEvent( sourceCompactableEvents: readonly RuntimeEvent[], clonedRuntimeEvents: ReadonlyMap, checkpointIds: Map, - transitionIds: Map, - transitionState: Map, providerTraceIds: ReadonlyMap, logicalCallIds: ReadonlyMap, + clonedTransitions: ReadonlyMap, ): EmittedAgentRunEvent | null { if (event.type === 'event_corrupt') { throw new Error(`Cannot copy corrupt AgentRun event ${event.id}`); @@ -973,7 +1061,7 @@ function cloneAgentRunEvent( const checkpoint = buildHistoryCompactCheckpoint({ sessionId: references.targetSessionId, coveredRuntimeEvents, - summary: sourceCheckpoint.summary, + summary: rewriteLedgerArchiveText(sourceCheckpoint.summary, references), highWaterName: sourceCheckpoint.highWaterName, highWaterSeq: sourceCheckpoint.highWaterSeq, now: sourceCheckpoint.createdAt, @@ -993,13 +1081,7 @@ function cloneAgentRunEvent( checkpoint, }; } else if (event.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE) { - const cloned = cloneModelProjectionTransition( - event, - references, - clonedRuntimeEvents, - transitionIds, - transitionState, - ); + const cloned = clonedTransitions.get(event); // Every transition whose target is in the copied slice was gathered into // this run's ledger, wherever it was recorded. So a transition that finds no // cloned target has genuinely lost its target as well, and dropping it diff --git a/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts b/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts index 7d83b7850c..600a38e204 100644 --- a/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts +++ b/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts @@ -146,7 +146,7 @@ test('checks byte and record budgets before fetching any ledger JSON', async (t) const get = statement.get.bind(statement); const all = statement.all.bind(statement); const count = (row: Record | undefined) => { - for (const key of ['payload_json', 'record_json']) + for (const key of ['payload_json', 'evidence_json', 'record_json']) if (typeof row?.[key] === 'string') materialized += Buffer.byteLength(row[key]); }; t.mock.method(statement, 'get', (...args: Parameters) => { @@ -173,6 +173,17 @@ test('checks byte and record budgets before fetching any ledger JSON', async (t) reason: 'too_large', }); assert.equal(materialized, 0); + f.db.exec('DELETE FROM core_agent_run_events'); + f.db + .prepare( + "UPDATE runtime_events SET payload_json = json_set(payload_json, '$.content.modelProjection.text', ?) WHERE event_id = 'response'", + ) + .run('x'.repeat(3 * 1024 * 1024)); + assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), { + ok: false, + reason: 'too_large', + }); + assert.equal(materialized, 0, 'oversized projection is refused before returning evidence JSON'); }); test('unscoped malformed transitions prevent a false complete history', async (t) => { @@ -238,7 +249,7 @@ test('database read failures are unavailable, not corrupt evidence', async (t) = DatabaseSync.prototype, 'prepare', function (this: DatabaseSync, sql: string) { - if (sql.includes('length(CAST(payload_json AS BLOB))')) + if (sql.includes('AS bytes FROM runtime_events')) throw new Error('injected database unavailable'); return prepare.call(this, sql); }, diff --git a/packages/storage/src/tool-result-archive-evidence.ts b/packages/storage/src/tool-result-archive-evidence.ts index ebfff21cff..e3dfa9d792 100644 --- a/packages/storage/src/tool-result-archive-evidence.ts +++ b/packages/storage/src/tool-result-archive-evidence.ts @@ -34,6 +34,12 @@ import { decodeAgentRunEvent } from '@maka/core/agent-run'; import { MODEL_PROJECTION_TARGET_SQL as TARGET } from './sqlite-core-execution-schema.js'; const TRANSITIONS = 'core_agent_run_events INDEXED BY core_model_projection_target'; const KIND = "event_type = 'model_projection_transition_recorded'"; +// SQLite may parse the containing JSON, but only these reconstruction fields +// cross into JS or count toward the evidence budget. Raw tool bytes never do. +const EVENT_EVIDENCE = `CASE WHEN json_valid(payload_json) THEN json_extract(payload_json, + '$.id', '$.sessionId', '$.runId', '$.invocationId', '$.turnId', '$.ts', '$.partial', + '$.author', '$.role', '$.content.kind', '$.content.id', '$.content.name', + '$.content.modelProjection', '$.content.providerExecuted') END`; /** Reader-first foundation; no archive writer or fallback is activated by opening it. */ export async function openToolResultArchiveEvidenceReader( @@ -64,10 +70,11 @@ export async function openToolResultArchiveEvidenceReader( const { sessionId, runtimeEventId } = accepted; const eventSize = db .prepare( - 'SELECT length(CAST(payload_json AS BLOB)) AS bytes FROM runtime_events WHERE event_id = ? AND session_id = ?', + `SELECT length(CAST(${EVENT_EVIDENCE} AS BLOB)) AS bytes FROM runtime_events WHERE event_id = ? AND session_id = ?`, ) .get(runtimeEventId, sessionId); if (!eventSize) return { ok: false, reason: 'not_found' }; + if (eventSize.bytes === null) return { ok: false, reason: 'corrupt' }; // An unscoped unreadable transition prevents proving completeness. if ( db @@ -88,12 +95,46 @@ export async function openToolResultArchiveEvidenceReader( return { ok: false, reason: 'too_large' }; const raw = db .prepare( - 'SELECT payload_json FROM runtime_events WHERE event_id = ? AND session_id = ?', + `SELECT ${EVENT_EVIDENCE} AS evidence_json FROM runtime_events WHERE event_id = ? AND session_id = ?`, ) .get(runtimeEventId, sessionId); let event: ReturnType; try { - event = decodeRuntimeEvent(JSON.parse(String(raw?.payload_json))); + const [ + id, + sessionId, + runId, + invocationId, + turnId, + ts, + partial, + author, + role, + kind, + callId, + name, + modelProjection, + providerExecuted, + ] = JSON.parse(String(raw?.evidence_json)); + event = decodeRuntimeEvent({ + id, + sessionId, + runId, + invocationId, + turnId, + ts, + partial, + author, + role, + content: { + kind, + id: callId, + name, + result: null, + ...(modelProjection === null ? {} : { modelProjection }), + ...(providerExecuted === null ? {} : { providerExecuted }), + }, + }); } catch { return { ok: false, reason: 'corrupt' }; }