From e5a7877c9d1dac23c32a8c81f7ffcc5ff159a925 Mon Sep 17 00:00:00 2001 From: chinawch007 Date: Sat, 19 Sep 2026 23:09:27 +0800 Subject: [PATCH] fix(runtime): preserve Agent Graph results across session revisions Recognize child references in persisted JSON agent_output responses, including diagnostic views and archived results, so editing a follow-up after Graph completion can create a revision without losing child results. Validate retained child ownership, Graph and terminal Run identities, and allow historical polls only alongside a terminal result for the same Run. Preserve validated references within the revision family. Copy independent Side Conversations as static snapshots with rewritten Artifact links and reconstructible ledger archives, including subsequent copies. Cover the complete Graph completion, follow-up, edit-and-resend workflow through the production Host, along with reference and archive regressions. Fixes #5331 Generated-by: OpenAI Codex --- .../execution-model-composition.test.ts | 132 ++ .../fixtures/agent-graph-provider-scenario.ts | 24 + .../session-revision-graph-references.test.ts | 212 +++- .../session-revision-graph-references.ts | 21 +- .../conversation-copy-agent-output.test.ts | 1060 +++++++++++++++++ .../src/conversation-copy-agent-output.ts | 294 +++++ packages/runtime/src/conversation-copy.ts | 240 +++- 7 files changed, 1929 insertions(+), 54 deletions(-) create mode 100644 packages/runtime/src/__tests__/conversation-copy-agent-output.test.ts create mode 100644 packages/runtime/src/conversation-copy-agent-output.ts 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 662afd5eb9..3d7f8c0bfb 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -3249,6 +3249,138 @@ test('production Host executes and durably supervises an Agent Graph over a real JSON.stringify(request.body).includes('child_session_run'), ), ); + + // Reproduce edit-and-resend after the asynchronous Graph has finished. + // The real tools above persist agent_output as JSON, not agent_swarm. + const followUpTurnId = 'graph-follow-up'; + const followUp = await composition.handlers['turn.start']( + { + sessionId: session.id, + turnId: followUpTurnId, + content: { text: 'Graph follow-up: explain the result' }, + }, + context, + ); + assert.ok(followUp.ok && followUp.result.kind === 'started'); + const followUpTerminal = await waitForTerminal( + composition, + session.id, + followUpTurnId, + followUp.result.turn, + context, + ); + assert.equal(followUpTerminal.status, 'completed'); + await waitFor(() => liveResidencies === 0, { + timeoutMs: 5_000, + pollMs: 10, + message: 'follow-up turn did not release its residency', + }); + const sourceRecord = await execution.sessionStore.readHeaderRecordSnapshot(session.id); + const revisionId = 'graph-follow-up-revision'; + const revisionInput = { + sourceSessionId: session.id, + targetSessionId: revisionId, + sourceTurnId: followUpTurnId, + expectedSourceRevision: sourceRecord.revision, + }; + let revision = await composition.handlers['session.revision.create'](revisionInput, context); + // Final turn projections may advance metadata after the terminal snapshot. + // Retry only the explicit optimistic conflict, as the Desktop client does. + for ( + let attempt = 0; + attempt < 3 && revision.ok && revision.result.kind === 'source_revision_conflict'; + attempt++ + ) { + revisionInput.expectedSourceRevision = revision.result.actualRevision; + revision = await composition.handlers['session.revision.create'](revisionInput, context); + } + assert.ok(revision.ok, JSON.stringify(revision)); + assert.equal(revision.result.kind, 'committed'); + const revisedRuns = await execution.runtimeEventStore.listSessionInvocations(revisionId); + assert.ok(!revisedRuns.some((run) => run.turnId === followUpTurnId)); + const copiedEvents = ( + await Promise.all( + revisedRuns.map((run) => + execution.runtimeEventStore.readRuntimeEvents(revisionId, run.runId), + ), + ) + ).flat(); + const sourceEvents = ( + await Promise.all( + runs.map((run) => execution.runtimeEventStore.readRuntimeEvents(session.id, run.runId)), + ) + ).flat(); + const outputResults = (events: RuntimeEvent[]) => + events.flatMap((event) => + event.content?.kind === 'function_response' && event.content.name === 'agent_output' + ? [event.content.result] + : [], + ); + assert.ok(outputResults(sourceEvents).length > 0); + assert.deepEqual(outputResults(copiedEvents), outputResults(sourceEvents)); + + const editedTurnId = 'edited-graph-follow-up'; + const edited = await composition.handlers['turn.start']( + { + sessionId: revisionId, + turnId: editedTurnId, + content: { text: 'Graph follow-up: explain the result in more detail' }, + }, + context, + ); + assert.ok(edited.ok && edited.result.kind === 'started'); + assert.equal( + (await waitForTerminal(composition, revisionId, editedTurnId, edited.result.turn, context)) + .status, + 'completed', + ); + assert.equal((await execution.runtimeEventStore.listSessionInvocations(child!.id)).length, 1); + assert.ok( + (await execution.runtimeEventStore.listSessionInvocations(session.id)).some( + (run) => run.turnId === followUpTurnId, + ), + ); + // The shared copier must not grant an independent Side Conversation the + // original child's identities through its model-visible JSON projection. + const sideSource = await execution.sessionStore.readHeaderRecordSnapshot(session.id); + const sideId = 'graph-follow-up-side-conversation'; + const side = await composition.handlers['session.branch.create']( + { + sourceSessionId: session.id, + targetSessionId: sideId, + sourceTurnId: followUpTurnId, + expectedSourceRevision: sideSource.revision, + intent: 'side_conversation', + }, + context, + ); + assert.ok(side.ok, JSON.stringify(side)); + assert.equal(side.result.kind, 'committed'); + const sideRuns = await execution.runtimeEventStore.listSessionInvocations(sideId); + const sideEvents = ( + await Promise.all( + sideRuns.map((run) => execution.runtimeEventStore.readRuntimeEvents(sideId, run.runId)), + ) + ).flat(); + const sideOutputs = sideEvents.flatMap((event) => + event.content?.kind === 'function_response' && event.content.name === 'agent_output' + ? [event.content] + : [], + ); + assert.equal(sideOutputs.length, outputResults(sourceEvents).length); + assert.equal(sideOutputs.length, 2, 'Copy both the result and diagnostic views'); + assert.ok( + sideOutputs.some((output) => JSON.stringify(output.result).includes(CHILD_AGENT_RESULT_TEXT)), + ); + for (const output of sideOutputs) { + const decoded = decodeCanonicalToolResultContent(output.result); + assert.equal(decoded.kind, 'json'); + for (const payload of [output.result, output.modelProjection]) { + assert.ok(payload, 'Side Conversation must retain a model projection'); + assert.ok(!JSON.stringify(payload).includes(child!.id)); + assert.ok(!JSON.stringify(payload).includes(childRuns[0]!.runId)); + } + } } finally { graphStore?.close(); try { diff --git a/packages/runtime-host/src/__tests__/fixtures/agent-graph-provider-scenario.ts b/packages/runtime-host/src/__tests__/fixtures/agent-graph-provider-scenario.ts index 8e5037ffaa..9f6b6ce27c 100644 --- a/packages/runtime-host/src/__tests__/fixtures/agent-graph-provider-scenario.ts +++ b/packages/runtime-host/src/__tests__/fixtures/agent-graph-provider-scenario.ts @@ -25,6 +25,7 @@ type ScenarioPhase = | 'await_checkpoint' | 'await_view_result' | 'await_agent_output' + | 'await_agent_diagnostics' | 'await_finish_result' | 'completed'; @@ -41,6 +42,14 @@ export class AgentGraphProviderScenario { constructor(private readonly childResultText: string) {} respond(body: Record, reply: AgentGraphProviderReply): void { + if (this.#phase === 'completed' && latestUserText(body).startsWith('Graph follow-up:')) { + assert.ok( + JSON.stringify(body).includes(this.childResultText), + 'Follow-up lost the child result', + ); + reply.text(`Answered ${latestUserText(body)}`); + return; + } const names = toolNames(body); // Read covers both files and the child's Session-scoped tool results. if (names.join(',') === 'Glob,Grep,Read') { @@ -148,6 +157,21 @@ export class AgentGraphProviderScenario { assert.equal(result.status, 'completed'); assert.equal(result.text, this.childResultText); this.#resultRecordId = requireString(result.resultRecordId, 'Graph result record id'); + this.#phase = 'await_agent_diagnostics'; + reply.toolCall('agent_output', { + locator: 'child_session_run', + child_session_id: requireString(invocation.sessionId, 'child Session id'), + run_id: requireString(invocation.runId, 'child Run id'), + view: 'all', + max_events: 2, + max_bytes: 4096, + }); + return; + } + case 'await_agent_diagnostics': { + const output = requireRecord(toolResult, 'agent diagnostics'); + assert.equal(requireRecord(output.budget, 'agent diagnostics budget').view, 'all'); + assert.ok(requireArray(output.runtimeEvents, 'child RuntimeEvents').length > 0); this.#phase = 'await_finish_result'; reply.toolCall('update_agent_graph', { operation: 'finish', diff --git a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts index e8fb45f78a..4cd24ccfe9 100644 --- a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts @@ -19,7 +19,10 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; -import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation'; +import { + runtimeInvocationOutcome, + type RuntimeInvocationRecord, +} from '@maka/core/runtime-invocation'; import { testInvocationRecord } from '@maka/runtime/test-only/invocation-fixture'; import type { SessionHeader, StoredMessage } from '@maka/core/session'; import { agentGraphIdForRootSession } from '@maka/runtime/stream-graph-coordinator'; @@ -36,6 +39,207 @@ const CHILD_TURN_ID = 'child-turn'; const CHILD_RUN_ID = 'child-run'; const CHILD_ARTIFACT_ID = 'child-artifact'; +function agentOutputValue(run = agentRun(), artifactIds = [CHILD_ARTIFACT_ID]) { + return { + execution: { kind: 'child_session', sessionId: CHILD_SESSION_ID, currentRunId: run.runId }, + invocation: run, + result: { + schemaVersion: 1, + status: runtimeInvocationOutcome(run) ?? 'running', + text: 'Review complete', + textTruncated: false, + artifactIds, + omittedArtifactIds: 0, + graph: childHeader().subagentParent!.graph, + ...(run.terminalEvent ? { terminalRuntimeEventId: run.terminalEvent.id } : {}), + }, + budget: { view: 'result' }, + events: [], + runtimeEvents: [], + diagnostics: [], + artifacts: [], + }; +} + +test('revision validates historical JSON agent_output against the retained child authority', async () => { + const run = agentRun(); + const value = agentOutputValue(run); + const message: StoredMessage = { ...linkedResult(), content: { kind: 'json', value } }; + const accepted = await prepare({ messages: [message, message] }); + assert.ok(accepted.ok); + assert.equal(accepted.references.size, 1); + assert.deepEqual([...accepted.references.get(CHILD_SESSION_ID)!.runIds], [CHILD_RUN_ID]); + const archived = await prepare({ + messages: [], + archivedResults: [JSON.stringify(message.content)], + }); + assert.ok(archived.ok); + + for (const input of [ + { childActive: true }, + { graphState: 'live' as const }, + { runs: [agentRun({ status: 'running' })] }, + { runs: [agentRun({ status: 'failed' })] }, + { runs: [{ ...run, terminalEvent: { ...run.terminalEvent!, id: 'different-terminal' } }] }, + { artifactTurnId: 'another-turn' }, + { + sessionHeaders: [ + sessionHeader(ROOT_SESSION_ID), + childHeader({ parentSessionId: 'another-family' }), + ], + }, + { + sessionHeaders: [ + sessionHeader(ROOT_SESSION_ID), + childHeader({ parentTurnId: 'excluded-turn' }), + ], + }, + ]) { + const rejected = await prepare({ messages: [message], ...input }); + assert.equal(rejected.ok, false, JSON.stringify(input)); + } + for (const key of ['graphId', 'workId', 'operatorId'] as const) { + const rejected = await prepare({ + messages: [ + { + ...message, + content: { + kind: 'json', + value: { + ...value, + result: { ...value.result, graph: { ...value.result.graph, [key]: 'unrelated' } }, + }, + }, + }, + ], + }); + assert.equal(rejected.ok, false, key); + } +}); + +test('mixed agent_output views validate and retain diagnostic-only Artifacts', async () => { + const result = agentOutputValue(); + const { result: _result, ...envelope } = result; + for (const view of ['events', 'runtime_events', 'all'] as const) { + const diagnostic = { + ...envelope, + budget: { view }, + artifacts: [ + { id: 'diagnostic-artifact', sessionId: CHILD_SESSION_ID, turnId: CHILD_TURN_ID }, + ], + }; + for (const kind of ['revision', 'side_conversation'] as const) { + for (const archived of [false, true]) { + const messages: StoredMessage[] = [ + { ...linkedResult(), content: { kind: 'json', value: result } }, + ...(!archived + ? [{ ...linkedResult(), content: { kind: 'json' as const, value: diagnostic } }] + : []), + ]; + const input = { + kind, + messages, + archivedResults: archived ? [JSON.stringify(diagnostic)] : [], + }; + const accepted = await prepare(input); + assert.ok(accepted.ok); + assert.deepEqual( + [...accepted.references.get(CHILD_SESSION_ID)!.artifactIds].sort(), + [CHILD_ARTIFACT_ID, 'diagnostic-artifact'].sort(), + ); + const rejected = await prepare({ + ...input, + artifactTurns: new Map([['diagnostic-artifact', 'unrelated-turn']]), + }); + assert.equal(rejected.ok, false); + } + } + const { terminalEvent: _terminal, ...runningInvocation } = diagnostic.invocation; + const running = { ...diagnostic, invocation: runningInvocation }; + const poll = { ...linkedResult(), content: { kind: 'json' as const, value: running } }; + assert.equal((await prepare({ messages: [poll] })).ok, false); + assert.equal( + ( + await prepare({ + messages: [poll, { ...linkedResult(), content: { kind: 'json', value: result } }], + }) + ).ok, + true, + ); + } +}); + +test('historical polls require a retained terminal result for the same child run and turn', async () => { + const message = ( + value: ReturnType, + ): Extract => ({ + ...linkedResult(), + content: { kind: 'json', value }, + }); + const poll = agentOutputValue(agentRun({ status: 'running' }), ['poll-artifact']); + const done = agentOutputValue(); + for (const kind of ['revision', 'side_conversation'] as const) { + for (const messages of [ + [message(poll), message(done)], + [message(done), message(poll)], + ]) { + const accepted = await prepare({ kind, messages }); + assert.ok(accepted.ok, JSON.stringify(accepted)); + assert.deepEqual( + [...accepted.references.get(CHILD_SESSION_ID)!.artifactIds].sort(), + [CHILD_ARTIFACT_ID, 'poll-artifact'].sort(), + ); + } + const archived = await prepare({ + kind, + messages: [message(poll)], + archivedResults: [JSON.stringify(message(done).content)], + }); + assert.ok(archived.ok); + for (const overrides of [ + { messages: [message(poll)] }, // The ledger's completion is outside the retained history. + { childActive: true }, + { graphState: 'live' as const }, + { runs: [agentRun({ status: 'running' })] }, + { + messages: [ + message(agentOutputValue(agentRun({ status: 'running', runId: 'other-run' }))), + message(done), + ], + }, + { + messages: [ + message(agentOutputValue(agentRun({ status: 'running', turnId: 'other-turn' }))), + message(done), + ], + }, + { + messages: [ + message({ + ...poll, + result: { ...poll.result, graph: { ...poll.result.graph!, workId: 'wrong-work' } }, + }), + message(done), + ], + }, + { + messages: [ + message(poll), + message({ ...done, result: { ...done.result, status: 'failed' as const } }), + ], + }, + { artifactTurns: new Map([['poll-artifact', 'unrelated-turn']]) }, + ] satisfies PrepareOverrides[]) { + const rejected = await prepare({ + kind, + messages: [message(poll), message(done)], + ...overrides, + }); + assert.equal(rejected.ok, false, JSON.stringify(overrides)); + } + } +}); + test('Agent Graph revision references accept absent and reject live control state', async () => { const noGraph = await prepare({ graphState: 'absent', @@ -328,6 +532,7 @@ interface PrepareOverrides { readonly sessionGraphState?: 'absent' | 'live' | 'terminal'; readonly graphState?: 'absent' | 'live' | 'terminal'; readonly artifactTurnId?: string; + readonly artifactTurns?: ReadonlyMap; readonly artifactMissing?: boolean; readonly childActive?: boolean; } @@ -360,7 +565,10 @@ async function prepare(overrides: PrepareOverrides = {}) { : { id: artifactId, sessionId, - turnId: overrides.artifactTurnId ?? CHILD_TURN_ID, + turnId: + overrides.artifactTurns?.get(artifactId) ?? + overrides.artifactTurnId ?? + CHILD_TURN_ID, createdAt: 1, name: 'result.txt', kind: 'file', diff --git a/packages/runtime-host/src/server/session-revision-graph-references.ts b/packages/runtime-host/src/server/session-revision-graph-references.ts index cd2cd79155..768181c9b2 100644 --- a/packages/runtime-host/src/server/session-revision-graph-references.ts +++ b/packages/runtime-host/src/server/session-revision-graph-references.ts @@ -180,6 +180,9 @@ export async function prepareAgentGraphRevisionReferences( const references = new Map(); const runsByChildSession = new Map>(); + const terminalResults = new Set( + requests.filter((request) => isTerminalRunStatus(request.status)).map(linkedRunKey), + ); for (const request of requests) { const childSessionId = request.childSessionId; const child = headersById.get(childSessionId); @@ -189,6 +192,10 @@ export async function prepareAgentGraphRevisionReferences( !parent || !familySessionIds.has(parent.parentSessionId) || (input.kind === 'revision' && !parent.graph) || + (request.graph !== undefined && + (request.graph.graphId !== parent.graph?.graphId || + request.graph.workId !== parent.graph?.workId || + request.graph.operatorId !== parent.graph?.operatorId)) || (parent.graph !== undefined && !referencedGraphs.get(parent.parentSessionId)?.has(parent.graph.graphId)) || !retainedTurnIds.has(parent.spawnedBy.parentTurnId) @@ -201,7 +208,11 @@ export async function prepareAgentGraphRevisionReferences( if (dependencies.isSessionActive(childSessionId)) { return failure('session_busy', 'A retained Agent Graph child is still active'); } - if (!isTerminalRunStatus(request.status)) { + // A poll describes the run at read time. It may precede a retained terminal + // result for that exact run, but cannot borrow one from another execution. + // Keep validating every observation, including its Artifact references. + const terminalResult = isTerminalRunStatus(request.status); + if (!terminalResult && !terminalResults.has(linkedRunKey(request))) { return failure('session_busy', 'A retained Agent Graph result is not terminal'); } @@ -230,7 +241,9 @@ export async function prepareAgentGraphRevisionReferences( !currentRun || currentRun.sessionId !== childSessionId || currentRun.turnId !== request.turnId || - !linkedResultStatusMatchesRun(request, currentRun) + (request.terminalEventId !== undefined && + request.terminalEventId !== currentRun.terminalEvent?.id) || + (terminalResult && !linkedResultStatusMatchesRun(request, currentRun)) ) { return failure('operation_unavailable', 'Retained Agent Graph run reference is unavailable'); } @@ -272,6 +285,10 @@ export async function prepareAgentGraphRevisionReferences( return { ok: true, references }; } +function linkedRunKey(reference: ConversationCopyLinkedChildReference): string { + return JSON.stringify([reference.childSessionId, reference.runId, reference.turnId]); +} + export function agentGraphRevisionAdmissionSessionIds(input: { readonly sourceSessionId: string; readonly sessionHeaders: readonly SessionHeader[]; diff --git a/packages/runtime/src/__tests__/conversation-copy-agent-output.test.ts b/packages/runtime/src/__tests__/conversation-copy-agent-output.test.ts new file mode 100644 index 0000000000..9c97ec3752 --- /dev/null +++ b/packages/runtime/src/__tests__/conversation-copy-agent-output.test.ts @@ -0,0 +1,1060 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { createHash, randomUUID } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import type { RuntimeEvent } from '@maka/core/runtime-event'; +import type { AgentRunEvent, EmittedAgentRunEvent } from '@maka/core/agent-run'; +import { + buildModelProjectionTransition, + MODEL_PROJECTION_TRANSITION_EVENT_TYPE, +} from '@maka/core/model-projection-transition'; +import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection'; +import { buildInvocationOpenedEvent } from '@maka/core/runtime-invocation'; +import type { StoredMessage } from '@maka/core/session'; +import { parseAttachmentResourceRef } from '@maka/core/attachments'; +import { openInteractiveArtifactStoreForWrite } from '@maka/storage/artifact-stores'; +import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import { + archivedToolResultContainsConversationOwnedReferences, + archivedToolResultContainsLinkedChildReferences, + collectConversationCopyLinkedChildReferences, + collectConversationCopySessionFileRefs, + cloneConversationRuntimeLedger, + prepareConversationRuntimeLedgerCopy, + rewriteConversationCopyMessage, + type ConversationCopyMessageReferenceMap, +} from '../conversation-copy.js'; +import { testInvocationRecord } from './invocation-fixture.js'; +import { isConversationCopyAgentOutputSnapshot } from '../conversation-copy-agent-output.js'; +import { + buildArchivedToolResultPlaceholder, + isArchivedToolResultPlaceholder, +} from '../tool-result-archive.js'; +import { archivedToolResultProjection } from '../tool-result-archive-transition.js'; +import { serializeToolResultProjectionV1 } from '../tool-result-archive-encoding.js'; +import { + loadModelProjectionTransitionsFromRunLedger, + reduceEffectiveModelProjections, +} from '../model-projection-transition-ledger.js'; +import { createLedgerArchiveResourceReader } from '../ledger-tool-result-archive-reader.js'; +import { readToolResultArchiveResource } from '../tool-result-archive-resource.js'; +import { readPageSchema, readToolResultPage, type ReadInput } from '../read-page.js'; + +function output() { + const invocation = testInvocationRecord({ + sessionId: 'child', + runId: 'child-run', + turnId: 'child-turn', + outcome: 'completed', + opening: { + lineage: { + parentSessionId: 'parent', + parentRunId: 'parent-run', + parentTurnId: 'parent-turn', + }, + }, + }); + return { + execution: { kind: 'child_session', sessionId: 'child', currentRunId: 'child-run' }, + invocation, + result: { + schemaVersion: 1, + status: 'completed', + graph: { graphId: 'graph', workId: 'work', operatorId: 'operator' }, + resultRecordId: 'result-record', + terminalRecordId: 'terminal-record', + terminalRuntimeEventId: invocation.terminalEvent!.id, + text: 'Reviewed all packages.', + textTruncated: false, + artifactIds: ['child-artifact'], + omittedArtifactIds: 0, + }, + events: [], + runtimeEvents: [], + artifacts: [], + diagnostics: [], + sourceHealth: { kind: 'healthy' }, + budget: { view: 'result', maxBytes: 32768, projectedBytes: 1000 }, + truncated: { + events: true, + runtimeEvents: true, + artifacts: true, + diagnostics: false, + bytes: false, + }, + }; +} + +function diagnosticOutput(view: 'events' | 'runtime_events' | 'all') { + const { result: _result, ...raw } = output(); + const { invocation } = raw; + const { status: _status, ...terminalEnvelope } = invocation.terminalEvent!; + const runtimeEvents: RuntimeEvent[] = [ + { + ...terminalEnvelope, + id: 'child-text', + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'Diagnostic model text' }, + }, + { + ...terminalEnvelope, + id: 'child-read-call-event', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'child-read-call', + name: 'Read', + args: { path: 'maka://runtime/attachments/child-artifact' }, + }, + }, + { + ...terminalEnvelope, + id: 'child-read-response-event', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'child-read-call', + name: 'Read', + result: { + kind: 'text', + text: 'Expected 12, received 34.\nReport: maka://runtime/attachments/child-artifact\nEnd of report.', + }, + }, + }, + { + ...terminalEnvelope, + id: 'child-bash-call-event', + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'child-bash-call', + name: 'Bash', + args: { command: 'npm test' }, + }, + }, + { + ...terminalEnvelope, + id: 'child-bash-response-event', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'child-bash-call', + name: 'Bash', + result: { stdout: '1 test failed', stderr: 'Assertion failed', exitCode: 1 }, + isError: true, + providerOutput: { opaque: 'provider-only-output' }, + }, + }, + invocation.terminalEvent!, + ]; + const events: AgentRunEvent[] = [ + { + id: 'child-event', + sessionId: invocation.sessionId, + runId: invocation.runId, + turnId: invocation.turnId, + ts: 2, + type: 'model_stream_completed', + message: + 'Diagnostic operational message: [report](maka://runtime/attachments/child-artifact)', + }, + ]; + return { + ...raw, + events: view === 'runtime_events' ? [] : events, + runtimeEvents: view === 'events' ? [] : runtimeEvents, + artifacts: ['child-artifact', 'diagnostic-artifact'].map((id) => ({ + id, + sessionId: 'child', + turnId: 'child-turn', + createdAt: 1, + name: `${id}.txt`, + kind: 'file' as const, + source: 'tool_result' as const, + relativePath: `${id}.txt`, + sizeBytes: 1, + })), + diagnostics: [ + { + code: 'runtime_terminal_missing', + runId: 'child-run', + turnId: 'child-turn', + message: 'Diagnostic warning', + }, + ], + budget: { ...raw.budget, view }, + }; +} + +for (const view of ['events', 'runtime_events', 'all'] as const) { + test(`agent_output ${view} collects bounded diagnostic Artifacts and snapshots its text`, () => { + const raw = diagnosticOutput(view); + for (const input of [ + { messages: [message(output()), message(raw)], runtimeEvents: [], archivedResults: [] }, + { messages: [message(output())], runtimeEvents: [], archivedResults: [JSON.stringify(raw)] }, + ]) { + const refs = collectConversationCopyLinkedChildReferences(input); + assert.equal(refs.length, 2); + assert.deepEqual(refs[1]?.artifactIds, ['child-artifact', 'diagnostic-artifact']); + assert.equal(refs[1]?.terminalEventId, raw.invocation.terminalEvent!.id); + } + const external = { + ...references(), + linkedChildren: { + mode: 'preserve_validated' as const, + references: new Map([ + [ + 'child', + { + runIds: new Set(['child-run']), + artifactIds: new Set(['child-artifact', 'diagnostic-artifact']), + }, + ], + ]), + }, + }; + assert.deepEqual(rewriteConversationCopyMessage(message(raw), external), message(raw)); + const copied = rewriteConversationCopyMessage(message(raw), { + ...external, + artifactIds: new Map([ + ['child-artifact', 'copied-result'], + ['diagnostic-artifact', 'copied-diagnostic'], + ]), + linkedChildren: { mode: 'snapshot', archivedResults: new Map() }, + }); + assert.ok(copied.type === 'tool_result' && copied.content.kind === 'json'); + assert.ok(isConversationCopyAgentOutputSnapshot(copied.content.value)); + assert.deepEqual(copied.content.value.artifactIds, ['copied-result', 'copied-diagnostic']); + assert.match(copied.content.value.text!, /Diagnostic warning/); + assert.match( + copied.content.value.text!, + view === 'events' ? /operational message/ : /model text/, + ); + assert.ok(!JSON.stringify(copied.content).includes('child')); + for (const invalid of [ + { ...raw, artifacts: [{ ...raw.artifacts[0], sessionId: 'unrelated' }] }, + { ...raw, artifacts: [{ ...raw.artifacts[0], turnId: 'unrelated' }] }, + { ...raw, diagnostics: [{ ...raw.diagnostics[0], runId: 'unrelated' }] }, + { + ...raw, + invocation: { + ...raw.invocation, + terminalEvent: { ...raw.invocation.terminalEvent, runId: 'unrelated' }, + }, + }, + ]) { + assert.deepEqual( + collectConversationCopyLinkedChildReferences({ + messages: [message(invalid)], + runtimeEvents: [], + archivedResults: [], + }), + [], + ); + } + }); +} + +for (const view of ['runtime_events', 'all'] as const) { + test(`Side Conversation retains ${view} tool evidence without execution identities`, () => { + const raw = diagnosticOutput(view); + // A bounded diagnostic read can contain only tool evidence, with no model + // text or operational messages to stand in for the tool result. + const toolEvents = raw.runtimeEvents.filter( + (event) => + event.content?.kind === 'function_call' || event.content?.kind === 'function_response', + ); + for (const runtimeEvents of [toolEvents, [toolEvents[1]!]]) { + const source = message({ ...raw, runtimeEvents, events: [], diagnostics: [] }); + const copied = rewriteConversationCopyMessage(source, { + ...references(), + artifactIds: new Map([ + ['child-artifact', 'copied-result'], + ['diagnostic-artifact', 'copied-diagnostic'], + ]), + linkedChildren: { mode: 'snapshot', archivedResults: new Map() }, + }); + assert.ok(copied.type === 'tool_result' && copied.content.kind === 'json'); + assert.ok(isConversationCopyAgentOutputSnapshot(copied.content.value)); + const text = copied.content.value.text!; + assert.match(text, /Read/); + assert.match(text, /Expected 12, received 34\./); + if (runtimeEvents.length > 1) { + assert.match(text, /maka:\/\/runtime\/attachments\/copied-result/); + assert.match(text, /npm test/); + assert.match(text, /Bash/); + assert.match(text, /1 test failed/); + assert.match(text, /Assertion failed/); + assert.match(text, /"isError":true/); + assert.ok(text.indexOf('npm test') < text.indexOf('Assertion failed')); + } + for (const excluded of ['child', 'provider-only-output', 'terminal', 'invocation']) { + assert.ok(!JSON.stringify(copied.content).includes(excluded), excluded); + } + } + }); +} + +function message(value: unknown = output()): Extract { + return { + type: 'tool_result', + id: 'output', + turnId: 'parent-turn', + ts: 3, + toolUseId: 'output-call', + isError: false, + content: { kind: 'json', value }, + }; +} + +function references(): Extract { + return { + mode: 'exact', + sourceSessionId: 'parent', + targetSessionId: 'revision', + artifactIds: new Map(), + relativePaths: new Map(), + runIds: new Map(), + runtimeEventIds: new Map(), + providerTraceIds: new Map(), + linkedChildren: { + mode: 'preserve_validated', + references: new Map([ + ['child', { runIds: new Set(['child-run']), artifactIds: new Set(['child-artifact']) }], + ]), + }, + }; +} + +test('revision collects historical JSON agent_output references from messages, events and archives', () => { + const msg = message(); + const event: RuntimeEvent = { + id: 'output-event', + sessionId: 'parent', + runId: 'parent-run', + invocationId: 'parent-run', + turnId: 'parent-turn', + ts: 3, + partial: false, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'output-call', + name: 'agent_output', + result: msg.content, + }, + }; + for (const input of [ + { messages: [msg], runtimeEvents: [], archivedResults: [] }, + { messages: [], runtimeEvents: [event], archivedResults: [] }, + { messages: [], runtimeEvents: [], archivedResults: [JSON.stringify(msg.content)] }, + { messages: [], runtimeEvents: [], archivedResults: [JSON.stringify(output())] }, + ]) { + assert.deepEqual(collectConversationCopyLinkedChildReferences(input), [ + { + childSessionId: 'child', + runId: 'child-run', + turnId: 'child-turn', + status: 'completed', + artifactIds: ['child-artifact'], + terminalEventId: 'child-run-terminal', + graph: { graphId: 'graph', workId: 'work', operatorId: 'operator' }, + }, + ]); + } + const archived = JSON.stringify(msg.content); + assert.equal(archivedToolResultContainsLinkedChildReferences(archived), true); + assert.equal(archivedToolResultContainsLinkedChildReferences(JSON.stringify(output())), true); + assert.equal( + archivedToolResultContainsConversationOwnedReferences(JSON.stringify(output()), 'parent'), + true, + ); + assert.equal(archivedToolResultContainsConversationOwnedReferences(archived, 'parent'), true); + const refs = references(); + assert.ok(refs.mode === 'exact' && refs.linkedChildren.mode === 'preserve_validated'); + assert.equal( + archivedToolResultContainsConversationOwnedReferences( + archived, + 'parent', + refs.linkedChildren.references, + ), + false, + ); +}); + +test('agent_output JSON recognition rejects malformed identities and unrelated JSON', () => { + for (const mutate of [ + (v: ReturnType) => { + v.execution.sessionId = 'other'; + }, + (v: ReturnType) => { + v.execution.currentRunId = 'other'; + }, + (v: ReturnType) => { + v.invocation.terminalEvent!.turnId = 'other'; + }, + (v: ReturnType) => { + v.result.status = 'failed'; + }, + (v: ReturnType) => { + v.result.terminalRuntimeEventId = 'other'; + }, + (v: ReturnType) => { + delete v.invocation.terminalEvent; + }, + (v: ReturnType) => { + v.result.schemaVersion = 2; + }, + (v: ReturnType) => { + v.budget.view = 'all'; + }, + (v: ReturnType) => { + v.result.graph.graphId = ''; + }, + (v: ReturnType) => { + v.result.artifactIds = ['']; + }, + ]) { + const value = output(); + mutate(value); + assert.deepEqual( + collectConversationCopyLinkedChildReferences({ + messages: [message(value)], + runtimeEvents: [], + archivedResults: [], + }), + [], + ); + } + const unrelated = message({ childSessionId: 'child', runId: 'child-run', result: output() }); + assert.deepEqual( + collectConversationCopyLinkedChildReferences({ + messages: [unrelated], + runtimeEvents: [], + archivedResults: [], + }), + [], + ); + assert.deepEqual(rewriteConversationCopyMessage(unrelated, references()), unrelated); +}); + +test('revision preserves the exact child output only with validated external references', () => { + const source = message(); + assert.deepEqual(rewriteConversationCopyMessage(source, references()), source); + for (const linkedChildren of [ + { mode: 'reject' as const }, + { mode: 'preserve_validated' as const, references: new Map() }, + { + mode: 'preserve_validated' as const, + references: new Map([ + ['child', { runIds: new Set(['other-run']), artifactIds: new Set(['child-artifact']) }], + ]), + }, + { + mode: 'preserve_validated' as const, + references: new Map([ + ['child', { runIds: new Set(['child-run']), artifactIds: new Set() }], + ]), + }, + ]) { + assert.throws( + () => + rewriteConversationCopyMessage(source, { + ...references(), + mode: 'exact', + artifactIds: new Map(), + relativePaths: new Map(), + linkedChildren, + }), + /linked child|external/, + ); + } +}); + +test('Side Conversation keeps only the child result snapshot and copied artifacts', () => { + const source = message(); + const result = rewriteConversationCopyMessage(source, { + ...references(), + mode: 'exact', + artifactIds: new Map([['child-artifact', 'snapshot-artifact']]), + relativePaths: new Map(), + linkedChildren: { mode: 'snapshot', archivedResults: new Map() }, + }); + assert.equal(result.type, 'tool_result'); + if (result.type !== 'tool_result') assert.fail(); + assert.deepEqual(result.content, { + kind: 'json', + value: { + kind: 'maka.agent_output_snapshot', + schemaVersion: 1, + status: 'completed', + text: 'Reviewed all packages.', + textTruncated: false, + artifactIds: ['snapshot-artifact'], + omittedArtifactIds: 0, + }, + }); + assert.ok(!JSON.stringify(result.content).includes('child-run')); + assert.deepEqual( + collectConversationCopyLinkedChildReferences({ + messages: [result], + runtimeEvents: [], + archivedResults: [], + }), + [], + ); +}); + +test('agent_output snapshot attachment links follow each copy while revision references stay external', () => { + const raw = output(); + raw.result.text = + 'Report: [download](maka://runtime/attachments/child-artifact). Unrelated: maka://runtime/attachments/unmapped-artifact'; + assert.deepEqual(rewriteConversationCopyMessage(message(raw), references()), message(raw)); + let source: StoredMessage = message(raw); + let artifactId = 'child-artifact'; + for (const [targetId, mode] of [ + ['side-1-artifact', 'snapshot'], + ['side-2-artifact', 'snapshot'], + ['revision-artifact', 'reject'], + ] as const) { + source = rewriteConversationCopyMessage(source, { + ...references(), + artifactIds: new Map([[artifactId, targetId]]), + linkedChildren: mode === 'snapshot' ? { mode, archivedResults: new Map() } : { mode }, + }); + assert.ok(source.type === 'tool_result' && source.content.kind === 'json'); + assert.ok(isConversationCopyAgentOutputSnapshot(source.content.value)); + assert.deepEqual(source.content.value.artifactIds, [targetId]); + assert.equal( + source.content.value.text, + `Report: [download](maka://runtime/attachments/${targetId}). Unrelated: maka://runtime/attachments/unmapped-artifact`, + ); + artifactId = targetId; + } +}); + +test('agent_output snapshots remain copyable through messages, events and archives', () => { + const first = rewriteConversationCopyMessage(message(), { + ...references(), + mode: 'exact', + artifactIds: new Map([['child-artifact', 'side-1-artifact']]), + linkedChildren: { mode: 'snapshot', archivedResults: new Map() }, + }); + assert.ok(first.type === 'tool_result' && first.content.kind === 'json'); + const value = first.content.value; + assert.ok(isConversationCopyAgentOutputSnapshot(value)); + const event = { + content: { kind: 'function_response', name: 'agent_output', result: first.content }, + } as RuntimeEvent; + for (const input of [ + { messages: [first], runtimeEvents: [], archivedResults: [] }, + { messages: [], runtimeEvents: [event], archivedResults: [] }, + { messages: [], runtimeEvents: [], archivedResults: [JSON.stringify(first.content)] }, + { messages: [], runtimeEvents: [], archivedResults: [JSON.stringify(value)] }, + ]) { + assert.deepEqual( + [...collectConversationCopySessionFileRefs({ sourceSessionId: 'side-1', ...input })], + ['side-1-artifact'], + ); + assert.deepEqual(collectConversationCopyLinkedChildReferences(input), []); + } + for (const archived of [JSON.stringify(value), JSON.stringify(first.content)]) { + assert.equal(archivedToolResultContainsConversationOwnedReferences(archived, 'side-1'), true); + const restored = rewriteConversationCopyMessage( + { + ...first, + content: { + kind: 'archived_tool_result', + status: 'not_loaded', + artifactId: 'archive', + runtimeEventId: 'event', + toolCallId: 'call', + toolName: 'agent_output', + originalEstimatedTokens: 20, + originalBytes: archived.length, + rewriteVersion: 1, + reason: 'stale_tool_result_pruned_before_compact', + }, + }, + { + ...references(), + mode: 'exact', + sourceSessionId: 'side-1', + targetSessionId: 'side-2', + artifactIds: new Map([['side-1-artifact', 'side-2-artifact']]), + linkedChildren: { mode: 'snapshot', archivedResults: new Map([['archive', archived]]) }, + }, + ); + assert.ok(restored.type === 'tool_result'); + assert.deepEqual(restored.content, { + kind: 'json', + value: { ...value, artifactIds: ['side-2-artifact'] }, + }); + } + // Untagged and malformed JSON must not acquire Session-owned reference semantics. + const { kind: _kind, ...untagged } = value; + for (const opaque of [ + untagged, + { ...value, schemaVersion: 2 }, + { ...value, artifactIds: [''] }, + { ...value, execution: {} }, + ]) { + assert.equal(isConversationCopyAgentOutputSnapshot(opaque), false); + assert.deepEqual( + rewriteConversationCopyMessage(message(opaque), references()), + message(opaque), + ); + } + const reclaimed = rewriteConversationCopyMessage(first, { + ...references(), + mode: 'exact', + linkedChildren: { mode: 'reject' }, + }); + assert.ok(reclaimed.type === 'tool_result'); + assert.deepEqual(reclaimed.content, { kind: 'json', value: { ...value, artifactIds: [] } }); +}); + +for (const { view, archiveDepth } of [ + { view: 'result', archiveDepth: 0 }, + { view: 'result', archiveDepth: 1 }, + { view: 'result', archiveDepth: 2 }, + { view: 'result', archiveDepth: 3 }, + { view: 'events', archiveDepth: 0 }, + { view: 'runtime_events', archiveDepth: 0 }, + { view: 'all', archiveDepth: 0 }, + { view: 'all', archiveDepth: 1 }, +] as const) { + test(`successive copies retain ${view} snapshots with ${archiveDepth} legacy archives`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-output-snapshot-copy-')); + const owner = await tryAcquireInteractiveRootOwner( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + assert.ok(owner); + const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); + try { + const { agentRunStore: runStore, runtimeEventStore } = + await openInteractiveExecutionStoresForWrite(owner.lease); + for (const id of ['child-artifact', 'diagnostic-artifact']) + await artifacts.create({ + id, + sessionId: 'child', + turnId: 'child-turn', + name: 'result.txt', + kind: 'file', + content: `${id} bytes`, + source: 'tool_result', + }); + const parent = testInvocationRecord({ + sessionId: 'parent', + runId: 'parent-run', + turnId: 'parent-turn', + outcome: 'completed', + closedAt: 5, + }); + const raw = view === 'result' ? output() : diagnosticOutput(view); + if ('result' in raw) { + // The migrated snapshot must be read through multiple public Read pages. + raw.result.text = + raw.result.text.repeat(1000) + ' [report](maka://runtime/attachments/child-artifact)'; + } + const sourceEvents: RuntimeEvent[] = [ + buildInvocationOpenedEvent({ + id: 'parent-opening', + run: parent, + opening: parent.opening, + openedAt: 1, + }), + { + id: 'call-event', + sessionId: 'parent', + runId: parent.runId, + invocationId: parent.invocationId, + turnId: parent.turnId, + ts: 2, + partial: false, + role: 'model', + author: 'agent', + content: { kind: 'function_call', id: 'output-call', name: 'agent_output', args: {} }, + }, + { + id: 'output-event', + sessionId: 'parent', + runId: parent.runId, + invocationId: parent.invocationId, + turnId: parent.turnId, + ts: 3, + partial: false, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'output-call', + name: 'agent_output', + result: message(raw).content, + modelProjection: { version: 1, kind: 'json', value: JSON.parse(JSON.stringify(raw)) }, + }, + }, + parent.terminalEvent!, + ]; + let sourceSessionId = 'parent'; + let messages: readonly StoredMessage[] = [message(raw)]; + const archives = new Map(); + const archiveRecords: EmittedAgentRunEvent[] = []; + let projection: DurableToolResultProjection = { + version: 1, + kind: 'json', + value: JSON.parse(JSON.stringify(raw)), + }; + let previousTransitionId: string | undefined; + let legacyRef: string | undefined; + for (let index = 0; index < archiveDepth; index++) { + const serialized = serializeToolResultProjectionV1(projection); + const artifactId = `legacy-archive-${index}`; + archives.set(artifactId, serialized); + await artifacts.create({ + id: artifactId, + sessionId: 'parent', + turnId: parent.turnId, + name: `${artifactId}.json`, + kind: 'file', + content: serialized, + source: 'tool_result_archive', + }); + const placeholder = buildArchivedToolResultPlaceholder({ + artifactId, + runtimeEventId: 'output-event', + toolCallId: 'output-call', + toolName: 'agent_output', + bodySha256: createHash('sha256').update(serialized).digest('hex'), + originalEstimatedTokens: 1000, + originalBytes: Buffer.byteLength(serialized), + reason: 'stale_tool_result_pruned_before_compact', + }); + placeholder.page = readToolResultPage(serialized, { path: placeholder.resourceRef! }, 800); + assert.ok(placeholder.page.next, 'Exercise legacy archive pagination'); + const transition = buildModelProjectionTransition({ + sessionId: 'parent', + target: { + runtimeEventId: 'output-event', + part: 'tool_result', + toolCallId: 'output-call', + toolName: 'agent_output', + }, + sourceProjection: projection, + replacement: archivedToolResultProjection(placeholder), + ...(previousTransitionId ? { previousTransitionId } : {}), + now: 4 + index, + }); + archiveRecords.push({ + id: transition.transitionId, + sessionId: 'parent', + runId: parent.runId, + turnId: parent.turnId, + ts: 4 + index, + type: MODEL_PROJECTION_TRANSITION_EVENT_TYPE, + data: { runtimeEventId: 'output-event', part: 'tool_result', transition }, + }); + previousTransitionId = transition.transitionId; + projection = transition.replacement; + messages = [message(placeholder)]; + legacyRef = placeholder.page.next.path; + } + if (legacyRef) { + const envelope = { + sessionId: 'parent', + runId: parent.runId, + invocationId: parent.invocationId, + turnId: parent.turnId, + partial: false, + }; + sourceEvents.splice( + -1, + 0, + { + ...envelope, + id: 'archive-read-call', + ts: 3.1, + role: 'model', + author: 'agent', + content: { + kind: 'function_call', + id: 'read-archive', + name: 'Read', + args: { path: legacyRef }, + }, + }, + { + ...envelope, + id: 'archive-read-result', + ts: 3.2, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'read-archive', + name: 'Read', + result: { kind: 'text', text: 'Archived output' }, + }, + }, + ); + } + for (const event of sourceEvents) + await runtimeEventStore.appendRuntimeEvent('parent', parent.runId, event); + for (const event of archiveRecords) await runStore.appendEvent('parent', parent.runId, event); + let previousArtifactIds = ['child-artifact', 'diagnostic-artifact']; + for (const targetSessionId of ['side-1', 'side-2', 'revision-3']) { + const events = ( + await Promise.all( + ( + await runtimeEventStore.listSessionInvocations(sourceSessionId) + ).map((run) => runtimeEventStore.readRuntimeEvents(sourceSessionId, run.runId)), + ) + ).flat(); + const plan = await prepareConversationRuntimeLedgerCopy({ + sourceSessionId, + sourceEvents: events, + copiedMessages: messages, + runStore, + runtimeEventStore, + }); + const artifactCopy = await artifacts.copyConversationArtifacts({ + sourceSessionId, + targetSessionId, + turnIds: plan.copyTurnIds, + includeArtifactIds: [ + ...collectConversationCopySessionFileRefs({ + sourceSessionId, + messages, + runtimeEvents: events, + archivedResults: sourceSessionId === 'parent' ? [...archives.values()] : [], + }), + ], + ...(sourceSessionId === 'parent' + ? { + excludeArtifactIds: [...archives.keys()], + linkedArtifacts: collectConversationCopyLinkedChildReferences({ + messages: [message(raw)], + runtimeEvents: [], + archivedResults: [], + }).map((reference) => ({ + sessionId: reference.childSessionId, + artifactIds: reference.artifactIds, + })), + } + : {}), + }); + const copied = await cloneConversationRuntimeLedger({ + plan, + copiedMessages: messages, + runStore, + runtimeEventStore, + newId: randomUUID, + referenceMap: { + mode: 'exact', + sourceSessionId, + targetSessionId, + ...artifactCopy, + linkedChildren: + targetSessionId === 'revision-3' + ? { mode: 'reject' } + : { + mode: 'snapshot', + archivedResults: sourceSessionId === 'parent' ? archives : new Map(), + }, + }, + }); + const copiedEvents = ( + await Promise.all( + copied.runIdMap.map(({ targetRunId }) => + runtimeEventStore.readRuntimeEvents(targetSessionId, targetRunId), + ), + ) + ).flat(); + const responseEvent = copiedEvents.find( + (event) => + event.content?.kind === 'function_response' && event.content.name === 'agent_output', + )!; + const response = responseEvent.content; + assert.ok(response?.kind === 'function_response'); + const snapshot = response.result; + assert.ok( + snapshot && + typeof snapshot === 'object' && + 'kind' in snapshot && + snapshot.kind === 'json' && + 'value' in snapshot && + isConversationCopyAgentOutputSnapshot(snapshot.value), + ); + const expectedArtifacts = + view === 'result' ? ['child-artifact'] : ['child-artifact', 'diagnostic-artifact']; + assert.equal(snapshot.value.artifactIds.length, expectedArtifacts.length); + for (const [index, artifactId] of snapshot.value.artifactIds.entries()) { + assert.ok(!previousArtifactIds.includes(artifactId)); + const read = await artifacts.readTextInSession(targetSessionId, artifactId); + assert.ok(read.ok); + assert.equal(read.text, `${expectedArtifacts[index]} bytes`); + } + const attachmentLinks = snapshot.value.text?.match( + /maka:\/\/runtime\/attachments\/[A-Za-z0-9_-]+/g, + ); + assert.ok(attachmentLinks?.length, 'Retain the attachment links in the snapshot text'); + for (const link of attachmentLinks) { + const ref = parseAttachmentResourceRef(link); + assert.ok(ref); + const read = await artifacts.readTextInSession(targetSessionId, ref.artifactId); + assert.ok(read.ok, `Copied link must resolve in ${targetSessionId}: ${link}`); + assert.equal(read.text, 'child-artifact bytes'); + } + if (view === 'runtime_events' || view === 'all') { + assert.match(snapshot.value.text!, /Expected 12, received 34\./); + assert.match(snapshot.value.text!, /npm test/); + assert.match(snapshot.value.text!, /Assertion failed/); + assert.ok(!snapshot.value.text!.includes('child-read-call')); + assert.ok(!snapshot.value.text!.includes('provider-only-output')); + } + assert.deepEqual(response.modelProjection, { + version: 1, + kind: 'json', + value: snapshot.value, + }); + const runIds = copied.runIdMap.map(({ targetRunId }) => targetRunId); + const records = ( + await Promise.all(runIds.map((runId) => runStore.readEvents(targetSessionId, runId))) + ).flat(); + const transitions = await loadModelProjectionTransitionsFromRunLedger( + runStore, + targetSessionId, + runIds, + ); + const reduction = reduceEffectiveModelProjections(copiedEvents, transitions.transitions); + assert.equal(reduction.applied.length, Math.min(archiveDepth, 1)); + assert.equal(reduction.rejected.length, 0); + let expectedProjection: DurableToolResultProjection = response.modelProjection!; + for (const transition of reduction.applied) { + assert.ok(transition.replacement.kind === 'json'); + const placeholder: unknown = transition.replacement.value; + assert.ok( + isArchivedToolResultPlaceholder(placeholder) && placeholder.rewriteVersion === 2, + ); + const reader = createLedgerArchiveResourceReader({ + read: async () => ({ + ok: true, + event: responseEvent, + transitions: records.filter( + (record) => + record.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE && + record.data?.runtimeEventId === responseEvent.id, + ), + }), + }); + const archived = await reader({ + ...placeholder, + sessionId: targetSessionId, + maxBytes: 1024 * 1024, + }); + assert.ok(archived.ok, JSON.stringify({ archived, targetSessionId })); + assert.equal( + archived.serializedResult, + serializeToolResultProjectionV1(expectedProjection), + ); + assert.ok(!archived.serializedResult.includes('child-run')); + assert.ok(!JSON.stringify(placeholder.page).includes('child-run')); + const read = async (input: ReadInput) => + readPageSchema.parse( + await readToolResultArchiveResource( + { readArchivedToolResultResource: reader }, + targetSessionId, + input, + ), + ); + const expectedBody = serializeToolResultProjectionV1(response.modelProjection!); + assert.ok(placeholder.resourceRef); + assert.ok(placeholder.page); + if (view === 'result') assert.ok(placeholder.page.next, 'Exercise snapshot pagination'); + // Follow both the public address and the embedded page's continuation; + // an exact internal ledger identity would hide an address self-loop. + for (const firstPage of [ + await read({ path: placeholder.resourceRef }), + placeholder.page, + ]) { + let page = firstPage; + let body = page.content; + const seen = new Set(); + while (page.next) { + assert.ok(!seen.has(page.next.path), 'Read continuation must make progress'); + seen.add(page.next.path); + page = await read(page.next); + body += page.content; + } + assert.equal(body, expectedBody); + } + expectedProjection = transition.replacement; + } + if (archiveDepth > 0) { + assert.deepEqual( + copied.copiedMessages[0]?.type === 'tool_result' + ? copied.copiedMessages[0].content + : undefined, + { + kind: 'json', + value: expectedProjection.kind === 'json' ? expectedProjection.value : undefined, + }, + ); + const readCall = copiedEvents.find( + (event) => event.content?.kind === 'function_call' && event.content.name === 'Read', + ); + assert.ok(readCall?.content?.kind === 'function_call'); + assert.deepEqual(readCall.content.args, { + path: `maka://runtime/tool-results/${responseEvent.id}`, + }); + assert.equal( + (await artifacts.getInSession(targetSessionId, 'legacy-archive-0'))?.record ?? null, + null, + ); + } else { + assert.deepEqual( + copied.copiedMessages[0]?.type === 'tool_result' + ? copied.copiedMessages[0].content + : undefined, + snapshot, + ); + } + previousArtifactIds = [...snapshot.value.artifactIds]; + messages = copied.copiedMessages; + sourceSessionId = targetSessionId; + } + } finally { + artifacts.close(); + await owner.close(); + await rm(root, { recursive: true, force: true }); + } + }); +} diff --git a/packages/runtime/src/conversation-copy-agent-output.ts b/packages/runtime/src/conversation-copy-agent-output.ts new file mode 100644 index 0000000000..d0d87aedaa --- /dev/null +++ b/packages/runtime/src/conversation-copy-agent-output.ts @@ -0,0 +1,294 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { isRecord, isStringArray } from '@maka/core/record-schema'; +import { decodeAgentRunEvent } from '@maka/core/agent-run'; +import { + decodeRuntimeEvent, + decodeRuntimeInvocationOpened, + type RuntimeEvent, +} from '@maka/core/runtime-event'; +import { runtimeInvocationOutcome } from '@maka/core/runtime-invocation'; +import { stableJsonStringify } from '@maka/core/tool-args-identity'; +import { runtimeInvocationFailureClass } from './runtime-event-read-model.js'; +import type { ConversationCopyLinkedChildReference } from './conversation-copy.js'; + +/** Child references in every agent_output view, including historical JSON envelopes. */ +export interface ConversationCopyAgentOutput { + readonly reference: ConversationCopyLinkedChildReference & { + readonly runId: string; + readonly turnId: string; + }; + readonly snapshot: ConversationCopyAgentOutputSnapshot; +} + +/** Session-owned result data that remains recognizable on subsequent copies. */ +export interface ConversationCopyAgentOutputSnapshot { + readonly kind: 'maka.agent_output_snapshot'; + readonly schemaVersion: 1; + readonly status: ConversationCopyLinkedChildReference['status']; + readonly text?: string; + readonly textTruncated: boolean; + readonly artifactIds: readonly string[]; + readonly omittedArtifactIds: number; + readonly failureClass?: string; +} + +export function isConversationCopyAgentOutputSnapshot( + value: unknown, +): value is ConversationCopyAgentOutputSnapshot { + return ( + isRecord(value) && + value.kind === 'maka.agent_output_snapshot' && + isResultPayload(value) && + Object.keys(value).every((key) => + [ + 'kind', + 'schemaVersion', + 'status', + 'text', + 'textTruncated', + 'artifactIds', + 'omittedArtifactIds', + 'failureClass', + ].includes(key), + ) + ); +} + +function isResultPayload( + value: Record, +): value is Record & Omit { + return ( + value.schemaVersion === 1 && + resultStatus(value.status) && + isStringArray(value.artifactIds) && + value.artifactIds.every(nonempty) && + typeof value.textTruncated === 'boolean' && + typeof value.omittedArtifactIds === 'number' && + Number.isSafeInteger(value.omittedArtifactIds) && + value.omittedArtifactIds >= 0 && + (value.text === undefined || typeof value.text === 'string') && + (value.failureClass === undefined || typeof value.failureClass === 'string') + ); +} + +/** + * Recognize the defined result envelope, never arbitrary nested JSON ids. + * These are reference claims, not authority: the Host still checks them against + * the retained child's metadata, invocation ledger, and Artifact ownership. + */ +export function conversationCopyAgentOutput( + value: unknown, +): ConversationCopyAgentOutput | undefined { + if (!isRecord(value)) return undefined; + const { execution, invocation, budget } = value; + if ( + !isRecord(execution) || + execution.kind !== 'child_session' || + !nonempty(execution.sessionId) || + !isRecord(invocation) || + invocation.sessionId !== execution.sessionId || + !nonempty(invocation.runId) || + !nonempty(invocation.turnId) || + !nonempty(invocation.invocationId) || + (execution.currentRunId !== undefined && execution.currentRunId !== invocation.runId) || + !isRecord(budget) || + typeof budget.view !== 'string' || + !['result', 'events', 'runtime_events', 'all'].includes(budget.view) || + !Array.isArray(value.events) || + !Array.isArray(value.runtimeEvents) || + !Array.isArray(value.diagnostics) || + !Array.isArray(value.artifacts) + ) + return undefined; + + try { + const opening = decodeRuntimeInvocationOpened(invocation.opening); + const terminal = + invocation.terminalEvent === undefined + ? undefined + : decodeRuntimeEvent(invocation.terminalEvent); + if ( + terminal && + (terminal.sessionId !== execution.sessionId || + terminal.runId !== invocation.runId || + terminal.turnId !== invocation.turnId || + terminal.invocationId !== invocation.invocationId || + terminal.partial) + ) + return undefined; + let result = value.result; + if (budget.view === 'result') { + if ( + value.events.length || + value.runtimeEvents.length || + value.diagnostics.length || + value.artifacts.length + ) + return undefined; + } else { + // Diagnostic views have no committed-result field. Their invocation + // still identifies the observation, and their Artifact list can contain + // outputs omitted by a separately bounded view=result read. + if ( + result !== undefined || + (budget.view === 'events' && value.runtimeEvents.length > 0) || + (budget.view === 'runtime_events' && value.events.length > 0) + ) + return undefined; + const sameRun = (event: { sessionId: string; runId: string; turnId: string }) => + event.sessionId === execution.sessionId && + event.runId === invocation.runId && + event.turnId === invocation.turnId; + const events = value.events.map(decodeAgentRunEvent); + const runtimeEvents = value.runtimeEvents.map(decodeRuntimeEvent); + if ( + !events.every(sameRun) || + !runtimeEvents.every( + (event) => sameRun(event) && event.invocationId === invocation.invocationId, + ) || + !value.diagnostics.every( + (item) => + isRecord(item) && + item.runId === invocation.runId && + item.turnId === invocation.turnId && + nonempty(item.code) && + typeof item.message === 'string', + ) || + !value.artifacts.every( + (item) => + isRecord(item) && + nonempty(item.id) && + item.sessionId === execution.sessionId && + item.turnId === invocation.turnId, + ) + ) + return undefined; + const text = [ + ...runtimeEvents.flatMap(diagnosticRuntimeEventText), + ...events.flatMap((event) => (event.message ? [event.message] : [])), + ...value.diagnostics.map((item) => item.message as string), + ].join('\n'); + const failureClass = runtimeInvocationFailureClass({ terminalEvent: terminal }); + result = { + schemaVersion: 1, + status: runtimeInvocationOutcome({ terminalEvent: terminal }) ?? 'running', + ...(text ? { text } : {}), + // The bounded diagnostic view cannot establish a complete final text + // or count omitted Artifacts. Retain its readable excerpt and known ids. + textTruncated: true, + artifactIds: value.artifacts.map((item) => item.id as string), + omittedArtifactIds: 0, + ...(failureClass ? { failureClass } : {}), + }; + } + if (!isRecord(result) || !isResultPayload(result)) return undefined; + if ( + terminal && + (runtimeInvocationOutcome({ terminalEvent: terminal }) !== result.status || + (result.terminalRuntimeEventId !== undefined && + result.terminalRuntimeEventId !== terminal.id)) + ) + return undefined; + if (!terminal && result.status !== 'running' && result.status !== 'waiting_for_user') + return undefined; + let graph: ConversationCopyLinkedChildReference['graph']; + if (result.graph !== undefined) { + if ( + !isRecord(result.graph) || + !nonempty(result.graph.graphId) || + !nonempty(result.graph.workId) || + !nonempty(result.graph.operatorId) + ) + return undefined; + graph = { + graphId: result.graph.graphId, + workId: result.graph.workId, + operatorId: result.graph.operatorId, + }; + } + return { + reference: { + childSessionId: execution.sessionId, + runId: invocation.runId, + turnId: invocation.turnId, + status: result.status, + artifactIds: result.artifactIds, + ...(result.failureClass ? { failureClass: result.failureClass } : {}), + ...(opening.lineage?.resumedFromRunId + ? { resumedFromRunId: opening.lineage.resumedFromRunId } + : {}), + ...(terminal ? { terminalEventId: terminal.id } : {}), + ...(graph ? { graph } : {}), + }, + // An independent Side Conversation keeps a result snapshot, without the + // original execution, invocation, Graph, or event identities. + snapshot: { + kind: 'maka.agent_output_snapshot', + schemaVersion: 1, + status: result.status, + ...(typeof result.text === 'string' ? { text: result.text } : {}), + textTruncated: result.textTruncated, + artifactIds: result.artifactIds, + omittedArtifactIds: result.omittedArtifactIds, + ...(result.failureClass ? { failureClass: result.failureClass } : {}), + }, + }; + } catch { + return undefined; + } +} + +/** Keep tool evidence as static text without copying event or tool-call identities. */ +function diagnosticRuntimeEventText(event: RuntimeEvent): string[] { + if (event.partial) return []; + const content = event.content; + switch (content?.kind) { + case 'text': + return [content.text]; + case 'function_call': + return [stableJsonStringify({ kind: content.kind, name: content.name, args: content.args })]; + case 'function_response': + return [ + stableJsonStringify({ + kind: content.kind, + name: content.name, + result: content.result, + ...(content.isError !== undefined ? { isError: content.isError } : {}), + }), + ]; + default: + return []; + } +} + +function nonempty(value: unknown): value is string { + return typeof value === 'string' && value.length > 0; +} + +function resultStatus(value: unknown): value is ConversationCopyLinkedChildReference['status'] { + return ( + value === 'completed' || + value === 'failed' || + value === 'cancelled' || + value === 'running' || + value === 'waiting_for_user' + ); +} diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 1648b808e9..fa0a92ef78 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -80,6 +80,12 @@ import { import { archivedToolResultProjection } from './tool-result-archive-transition.js'; import { serializeToolResultProjectionV1 } from './tool-result-archive-encoding.js'; import { createHash } from 'node:crypto'; +import { encodeCanonicalRuntimeEvent } from '@maka/core/canonical-runtime-event'; +import { + conversationCopyAgentOutput, + isConversationCopyAgentOutputSnapshot, + type ConversationCopyAgentOutputSnapshot, +} from './conversation-copy-agent-output.js'; import { readToolResultPage, readableToolResult, @@ -111,6 +117,12 @@ export interface ConversationCopyLinkedChildReference { readonly artifactIds: readonly string[]; readonly status: 'completed' | 'failed' | 'cancelled' | 'running' | 'waiting_for_user'; readonly failureClass?: string; + readonly terminalEventId?: string; + readonly graph?: { + readonly graphId: string; + readonly workId: string; + readonly operatorId: string; + }; } export type ConversationCopyArtifactReferenceMap = @@ -192,7 +204,7 @@ function collectConversationCopyStorageRefs( const addSerialized = (value: unknown): void => { if (isArchivedToolResultPlaceholder(value)) return; try { - addContent(decodePersistedToolResultContent(markPersisted(value))); + addContent(decodeConversationCopyToolResult(value)); } catch { // Opaque tool results carry no typed StorageRef. } @@ -323,7 +335,8 @@ function rewriteAttachmentResourceRefs( text: string, artifactIds: ReadonlyMap, ): string { - return text.replace(/maka:\/\/runtime\/attachments\/[^\s)\]}>`'",;:!]+/g, (candidate) => { + // Snapshot tool evidence can contain JSON-escaped quotes and newlines. + return text.replace(/maka:\/\/runtime\/attachments\/[^\\\s)\]}>`'",;:!]+/g, (candidate) => { const parsed = parseAttachmentResourceRef(candidate); const artifactId = parsed ? artifactIds.get(parsed.artifactId) : undefined; return artifactId ? `maka://runtime/attachments/${artifactId}` : candidate; @@ -845,6 +858,15 @@ function rewriteReadInput( references, ); if (!input.path.startsWith('maka://read/')) return { ...input, path }; + if ( + path !== original && + references.ledgerArchives?.has(original) && + parseToolResultArchiveResourceRef(original)?.storage !== 'ledger' + ) { + // A retired v1 body has become a rewritten ledger snapshot. Its previous + // byte offset and digest cannot address the new resource. + return { path }; + } const url = new URL(input.path); if (eventId && path !== original) { const copied = clonedEvents.get(eventId); @@ -962,6 +984,14 @@ async function loadConversationCopyRunEvents( ); } +// Legacy archive bodies and provider projections can store the raw result; +// RuntimeEvents and transcript messages normally carry the JSON wrapper. +function decodeConversationCopyToolResult(value: unknown): ToolResultContent { + if (conversationCopyAgentOutput(value) || isConversationCopyAgentOutputSnapshot(value)) + return { kind: 'json', value }; + return decodePersistedToolResultContent(markPersisted(value)); +} + export function archivedToolResultContainsConversationOwnedReferences( serializedResult: string, sourceSessionId: string, @@ -972,12 +1002,21 @@ export function archivedToolResultContainsConversationOwnedReferences( let content: ToolResultContent; try { - content = decodePersistedToolResultContent(markPersisted(value)); + content = decodeConversationCopyToolResult(value); } catch { return false; } if (content.kind === 'archived_tool_result') return true; + if (content.kind === 'json') { + if (isConversationCopyAgentOutputSnapshot(content.value)) + return content.value.artifactIds.length > 0; + const output = conversationCopyAgentOutput(content.value); + return ( + output !== undefined && + !linkedChildReferencesAreExternal(output.reference, externalChildReferences) + ); + } if (content.kind === 'image') { return ( (content.ref.kind === 'session_file' || content.ref.kind === 'session_context') && @@ -1015,9 +1054,7 @@ export function archivedToolResultContainsLinkedChildReferences(serializedResult if (isArchivedToolResultPlaceholder(value)) return false; try { return ( - conversationCopyLinkedChildReferences( - decodePersistedToolResultContent(markPersisted(value)), - ).length > 0 + conversationCopyLinkedChildReferences(decodeConversationCopyToolResult(value)).length > 0 ); } catch { return false; @@ -1027,6 +1064,10 @@ export function archivedToolResultContainsLinkedChildReferences(serializedResult export function conversationCopyLinkedChildReferences( content: ToolResultContent, ): readonly ConversationCopyLinkedChildReference[] { + if (content.kind === 'json') { + const output = conversationCopyAgentOutput(content.value); + return output ? [output.reference] : []; + } if (content.kind === 'subagent') { if (!content.childSessionId) return []; return [ @@ -1063,22 +1104,24 @@ export function collectConversationCopyLinkedChildReferences(input: { readonly runtimeEvents: readonly RuntimeEvent[]; readonly archivedResults: readonly string[]; }): readonly ConversationCopyLinkedChildReference[] { - const references: ConversationCopyLinkedChildReference[] = []; + return conversationCopyToolResults(input).flatMap(conversationCopyLinkedChildReferences); +} + +function conversationCopyToolResults( + input: ConversationCopyStorageReferenceInput, +): readonly ToolResultContent[] { + const results: ToolResultContent[] = []; const add = (value: unknown): void => { if (isArchivedToolResultPlaceholder(value)) return; try { - references.push( - ...conversationCopyLinkedChildReferences( - decodePersistedToolResultContent(markPersisted(value)), - ), - ); + results.push(decodeConversationCopyToolResult(value)); } catch { - // Opaque tool results have no typed linked-child references. + // Opaque tool results have no typed conversation-copy references. } }; for (const message of input.messages) { if (message.type === 'tool_result') { - references.push(...conversationCopyLinkedChildReferences(message.content)); + results.push(message.content); } } for (const event of input.runtimeEvents) { @@ -1087,7 +1130,7 @@ export function collectConversationCopyLinkedChildReferences(input: { for (const serializedResult of input.archivedResults) { add(deserializeToolResultArchive(serializedResult)); } - return references; + return results; } /** @@ -1099,6 +1142,8 @@ export function collectConversationCopyLinkedChildReferences(input: { * their refs resolve in `rewriteStorageRef`. Walks exactly the ref sites reached * by `rewriteStorageRef`: user-message attachments, tool_result image refs, text * runtime-event attachments, and function_response / archived tool-result images. + * Static agent_output snapshots also own their Artifacts; those can retain a + * child Turn id outside the copied parent's turn closure and need explicit inclusion. */ export function collectConversationCopySessionFileRefs(input: { readonly sourceSessionId: string; @@ -1112,6 +1157,11 @@ export function collectConversationCopySessionFileRefs(input: { refs.add(ref.relativePath); } } + for (const content of conversationCopyToolResults(input)) { + if (content.kind === 'json' && isConversationCopyAgentOutputSnapshot(content.value)) { + for (const artifactId of content.value.artifactIds) refs.add(artifactId); + } + } return refs; } @@ -1227,10 +1277,9 @@ function cloneAgentRunEvent( }; } else if (event.type === MODEL_PROJECTION_TRANSITION_EVENT_TYPE) { 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 - // cannot bring replaced content back. + // A transition is omitted only when its target was excluded or its archive + // wrapper was collapsed into an already copied archive. Neither case can + // bring replaced content back into the model-visible history. if (!cloned) return null; data = { ...event.data, transition: cloned, runtimeEventId: cloned.target.runtimeEventId }; } @@ -1275,29 +1324,57 @@ function cloneModelProjectionTransition( } const placeholder = rawPlaceholder as ArchivedToolResultPlaceholder; const existing = transitionState.get(clonedTarget.id); - const sourceProjection = existing?.projection ?? baseToolResultProjection(clonedTarget); + // RuntimeEvent storage canonicalizes JSON key order. Hash the bytes a later + // archive read will reconstruct, including newly built snapshot projections. + const sourceProjection = + existing?.projection ?? + baseToolResultProjection(encodeCanonicalRuntimeEvent(clonedTarget).event); if (!sourceProjection) { throw new Error(`Cannot copy model projection transition ${event.id} onto its target`); } + const previousTransitionId = source.previousTransitionId + ? requiredMappedId(transitionIds, source.previousTransitionId, 'model projection transition') + : undefined; let rewritten: ArchivedToolResultPlaceholder; - if (placeholder.rewriteVersion === 2) { - const { previousTransitionId: _previous, ...rest } = placeholder; + // Side Conversations retire legacy archives whose bodies contain owned or + // linked-child references. Archive the rewritten projection in the copied + // ledger instead, so the transition no longer needs the excluded Artifact. + if ( + placeholder.rewriteVersion === 2 || + archivedSnapshotResult(placeholder.artifactId, references) !== undefined + ) { + const sourceRef = placeholder.resourceRef ?? buildToolResultArchiveResourceRef(placeholder); + const previousPlaceholder = + existing?.projection.kind === 'json' ? existing.projection.value : undefined; + if ( + existing && + isArchivedToolResultPlaceholder(previousPlaceholder) && + previousPlaceholder.rewriteVersion === 2 + ) { + if (previousTransitionId !== existing.transitionId) + throw new Error(`Cannot collapse disconnected archive transition ${event.id}`); + // Public Read addresses identify an event, not a transition. Archiving + // this placeholder again would make that address return itself. Keep the + // first archive of the copied body and alias every later wrapper to it. + references.ledgerArchives?.set(sourceRef, previousPlaceholder); + transitionIds.set(source.transitionId, existing.transitionId); + return null; + } + if (clonedTarget.content?.kind === 'function_response' && !clonedTarget.content.modelProjection) + clonedTarget.content.modelProjection = sourceProjection; const serialized = serializeToolResultProjectionV1(sourceProjection); rewritten = buildLedgerArchivedToolResultPlaceholder({ - ...rest, + storage: 'ledger', runtimeEventId: clonedTarget.id, + toolCallId: placeholder.toolCallId, + toolName: placeholder.toolName, + originalEstimatedTokens: placeholder.originalEstimatedTokens, + reason: placeholder.reason, + ...(placeholder.page ? { page: placeholder.page } : {}), sourceProjectionDigest: durableToolResultProjectionDigest(sourceProjection), bodySha256: createHash('sha256').update(serialized).digest('hex'), originalBytes: Buffer.byteLength(serialized), - ...(source.previousTransitionId - ? { - previousTransitionId: requiredMappedId( - transitionIds, - source.previousTransitionId, - 'model projection transition', - ), - } - : {}), + ...(previousTransitionId ? { previousTransitionId } : {}), }); if (rewritten.page) { delete rewritten.page; @@ -1307,10 +1384,7 @@ function cloneModelProjectionTransition( READ_PAGE_MAX_CHARS - JSON.stringify(rewritten).length - 32, ); } - references.ledgerArchives?.set( - placeholder.resourceRef ?? buildToolResultArchiveResourceRef(placeholder), - rewritten, - ); + references.ledgerArchives?.set(sourceRef, rewritten); } else { rewritten = rewriteArchivedToolResult(placeholder, references); } @@ -1327,15 +1401,7 @@ function cloneModelProjectionTransition( // The applied chain is copied in fold order, so a predecessor is always // rebuilt before its successor. An unmapped one means the chain broke, and // rooting the successor instead would change what the fold decides. - ...(source.previousTransitionId - ? { - previousTransitionId: requiredMappedId( - transitionIds, - source.previousTransitionId, - 'model projection transition', - ), - } - : {}), + ...(previousTransitionId ? { previousTransitionId } : {}), now: source.createdAt, }); transitionIds.set(source.transitionId, transition.transitionId); @@ -1643,9 +1709,9 @@ function rewriteRuntimeEventReferences( result: rewriteRuntimeToolResult(event.content.result, references), ...(event.content.modelProjection ? { - modelProjection: rewriteDurableToolResultProjectionArtifactRefs( + modelProjection: rewriteConversationCopyModelProjection( event.content.modelProjection, - (ref) => rewriteProjectionArtifactRef(ref, references), + references, ), } : {}), @@ -1785,6 +1851,31 @@ function rewriteToolResultContent( content: ToolResultContent, references: ConversationCopyMessageReferenceMap, ): ToolResultContent { + if (content.kind === 'json') { + if (isConversationCopyAgentOutputSnapshot(content.value)) { + return rewriteAgentOutputSnapshot(content.value, references); + } + const output = conversationCopyAgentOutput(content.value); + if (output) { + if (linkedChildrenAreSnapshots(references)) { + return rewriteAgentOutputSnapshot(output.snapshot, references); + } + // Revision children remain owned by their original physical Session. + // Validate those external references without rewriting their identities. + rewriteLinkedRunId( + output.reference.runId, + output.reference.childSessionId, + references, + 'AgentRun', + ); + rewriteLinkedArtifactIds( + output.reference.artifactIds, + output.reference.childSessionId, + references, + ); + return content; + } + } if (content.kind === 'image') { return { ...content, ref: rewriteStorageRef(content.ref, references) }; } @@ -1891,6 +1982,40 @@ function rewriteToolResultContent( return content; } +function rewriteAgentOutputSnapshot( + snapshot: ConversationCopyAgentOutputSnapshot, + references: ConversationCopyMessageReferenceMap, +): ToolResultContent { + return { + kind: 'json', + value: { + ...snapshot, + ...(snapshot.text !== undefined && references.mode === 'exact' + ? { text: rewriteAttachmentResourceRefs(snapshot.text, references.artifactIds) } + : {}), + artifactIds: rewriteArtifactIds(snapshot.artifactIds, references), + }, + }; +} + +function rewriteConversationCopyModelProjection( + projection: DurableToolResultProjection, + references: ConversationCopyMessageReferenceMap, +): DurableToolResultProjection { + if ( + projection.kind === 'json' && + (conversationCopyAgentOutput(projection.value) || + isConversationCopyAgentOutputSnapshot(projection.value)) + ) { + const content = rewriteToolResultContent({ kind: 'json', value: projection.value }, references); + if (content.kind === 'json') + return { ...projection, value: content.value as typeof projection.value }; + } + return rewriteDurableToolResultProjectionArtifactRefs(projection, (ref) => + rewriteProjectionArtifactRef(ref, references), + ); +} + function rewriteRuntimeToolResult( value: unknown, references: ConversationCopyMessageReferenceMap, @@ -1902,11 +2027,15 @@ function rewriteRuntimeToolResult( } let content: ToolResultContent; try { - content = decodePersistedToolResultContent(markPersisted(value)); + content = decodeConversationCopyToolResult(value); } catch { return value; } - return rewriteToolResultContent(content, references); + const rewritten = rewriteToolResultContent(content, references); + return (conversationCopyAgentOutput(value) || isConversationCopyAgentOutputSnapshot(value)) && + rewritten.kind === 'json' + ? rewritten.value + : rewritten; } function rewriteArtifactIds( @@ -1958,6 +2087,17 @@ function rewriteArchivedSnapshot( | Extract, references: ConversationCopyMessageReferenceMap, ): ToolResultContent | undefined { + const resourceRef = + value.resourceRef ?? + (value.artifactId && value.bodySha256 + ? buildToolResultArchiveResourceRef({ + artifactId: value.artifactId, + bodySha256: value.bodySha256, + originalBytes: value.originalBytes, + }) + : undefined); + const migrated = resourceRef ? references.ledgerArchives?.get(resourceRef) : undefined; + if (migrated) return { kind: 'json', value: migrated }; const serializedResult = archivedSnapshotResult(value.artifactId, references); if (serializedResult === undefined) return undefined; const archived = deserializeToolResultArchive(serializedResult); @@ -1965,7 +2105,7 @@ function rewriteArchivedSnapshot( return unavailableArchivedToolResult(value, references); } try { - const decoded = decodePersistedToolResultContent(markPersisted(archived)); + const decoded = decodeConversationCopyToolResult(archived); return decoded.kind === 'archived_tool_result' ? unavailableArchivedToolResult(value, references) : rewriteToolResultContent(decoded, references);