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..f75f1dcd08 100644 --- a/packages/core/src/tool-result-archive-evidence.ts +++ b/packages/core/src/tool-result-archive-evidence.ts @@ -24,8 +24,10 @@ 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; } | { 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..f939e91ec4 100644 --- a/packages/runtime-host/src/__tests__/execution-artifacts.test.ts +++ b/packages/runtime-host/src/__tests__/execution-artifacts.test.ts @@ -19,6 +19,15 @@ 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 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'; import { join } from 'node:path'; @@ -29,12 +38,222 @@ 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'; +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' }), + ); + 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, + ); + const projectedEvidence = await evidence.read({ + sessionId: 'session', + 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', + 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 +418,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/__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-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..139f129571 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -18,6 +18,10 @@ */ import assert from 'node:assert/strict'; +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'; @@ -32,6 +36,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'; @@ -62,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, @@ -79,6 +90,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 +2807,389 @@ function sourceProjectionTransition(input: { }); } +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, + 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 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-')); 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..5662a681be 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) { @@ -569,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) => { @@ -583,10 +654,9 @@ export async function cloneConversationRuntimeLedger( sourceCompactableEvents.get(plan.run.runId) ?? [], clonedEventBySourceId, checkpointIds, - transitionIds, - transitionState, providerTraceIds, logicalCallIds, + clonedTransitions, ); return clonedEvent ? [clonedEvent] : []; }); @@ -604,9 +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 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) => { @@ -661,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. */ @@ -873,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}`); @@ -938,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, @@ -958,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 @@ -1007,16 +1124,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 +1633,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 +1963,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 +1980,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/__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 17ac5c6d77..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' }; } @@ -114,7 +155,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 {