diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 76346cf745..4d761d3f27 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -27,6 +27,7 @@ import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { RuntimeEventStore } from '@maka/core/runtime-event-store'; import type { StoredMessage } from '@maka/core/session'; import { decodeCanonicalToolResultContent } from '@maka/core/tool-result-record-schema'; +import { decodeModelCallAttempt } from '@maka/core/model-call-attempt'; import { isSessionInlineRun } from '@maka/core/agent-run'; import { canonicalToolArgsHash } from '@maka/core/tool-args-identity'; import { createSqliteAgentRunStore } from '@maka/storage/agent-run-store'; @@ -1266,6 +1267,215 @@ test('conversation copy validates operational events before persisting target le } }); +test('conversation copy rewrites the nested identity of a model call attempt', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-conversation-model-call-copy-')); + try { + const runStore = createSqliteAgentRunStore(root); + const runtimeEventStore = createWorkspaceRuntimeStore(root); + await runStore.createRun( + agentRunHeader({ + runId: 'run-source', + invocationId: 'invocation-source', + turnId: 'turn-1', + cwd: root, + }), + ); + for (const event of [ + runtimeEvent({ + id: 'event-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'copy this turn' }, + }), + runtimeEvent({ id: 'event-terminal', ts: 2, status: 'completed' }), + ]) { + await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); + } + // The envelope identity (session/run/id) and the nested ModelCallAttempt + // identity start out equal, exactly as the writer emits them. + await runStore.appendEvent('session-source', 'run-source', { + type: 'model_call_attempt_recorded', + id: 'attempt-source', + runId: 'run-source', + sessionId: 'session-source', + turnId: 'turn-1', + ts: 2, + data: { + schemaVersion: 1, + logicalCallId: 'logical-source', + attemptId: 'attempt-source', + traceId: 'trace-source', + sessionId: 'session-source', + runId: 'run-source', + turnId: 'turn-1', + step: 0, + attempt: 0, + callKind: 'main', + providerId: 'provider', + modelId: 'model', + captureArtifactId: 'artifact-source', + startedAt: 1, + completedAt: 2, + latencyMs: 1, + status: 'completed', + usageBasis: 'reported', + inputTokens: 10, + outputTokens: 5, + costBasis: 'priced', + costUsd: 0.01, + }, + }); + const source = await new RuntimeReadModel({ + runStore, + 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([['artifact-source', 'artifact-target']]), + relativePaths: new Map(), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const [targetRun] = await runStore.listSessionRuns('session-target'); + assert.ok(targetRun); + const targetEvents = await runStore.readEvents('session-target', targetRun.runId); + const attempt = targetEvents.find((event) => event.type === 'model_call_attempt_recorded'); + assert.ok(attempt); + // The envelope moved to the target session/run. + assert.equal(attempt.sessionId, 'session-target'); + assert.equal(attempt.runId, targetRun.runId); + // The nested payload identity now agrees with the rewritten envelope instead + // of retaining the source identity — the model-call projection guard rejects + // any attempt whose payload disagrees with its envelope as unreadable. + assert.equal(attempt.data?.sessionId, 'session-target'); + assert.equal(attempt.data?.runId, targetRun.runId); + assert.equal(attempt.data?.attemptId, attempt.id); + assert.equal(attempt.data?.turnId, 'turn-1'); + // Owned trace/logical-call/artifact identity is remapped, not carried over. + assert.notEqual(attempt.data?.logicalCallId, 'logical-source'); + assert.notEqual(attempt.data?.traceId, 'trace-source'); + assert.equal(attempt.data?.captureArtifactId, 'artifact-target'); + // The rewritten record is still a valid accounting authority whose identity + // matches the envelope the ledger projects it under. + const decoded = decodeModelCallAttempt(attempt.data); + assert.equal(decoded.sessionId, attempt.sessionId); + assert.equal(decoded.runId, attempt.runId); + assert.equal(decoded.attemptId, attempt.id); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('conversation copy repairs a model call attempt stranded by a pre-fix copy', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-conversation-model-call-legacy-')); + try { + const runStore = createSqliteAgentRunStore(root); + const runtimeEventStore = createWorkspaceRuntimeStore(root); + await runStore.createRun( + agentRunHeader({ + runId: 'run-source', + invocationId: 'invocation-source', + turnId: 'turn-1', + cwd: root, + }), + ); + for (const event of [ + runtimeEvent({ + id: 'event-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'copy this turn again' }, + }), + runtimeEvent({ id: 'event-terminal', ts: 2, status: 'completed' }), + ]) { + await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); + } + // Simulate a session that was itself copied before this fix existed: the + // pre-fix copy path rewrote the envelope id but left the nested payload at + // the *grandparent* identity, so the envelope id and the nested attemptId / + // session / run disagree. Such a session must still be copyable. + await runStore.appendEvent('session-source', 'run-source', { + type: 'model_call_attempt_recorded', + id: 'attempt-envelope', + runId: 'run-source', + sessionId: 'session-source', + turnId: 'turn-1', + ts: 2, + data: { + schemaVersion: 1, + logicalCallId: 'logical-grandparent', + attemptId: 'attempt-grandparent', + traceId: 'trace-grandparent', + sessionId: 'session-grandparent', + runId: 'run-grandparent', + turnId: 'turn-1', + step: 0, + attempt: 0, + callKind: 'main', + providerId: 'provider', + modelId: 'model', + startedAt: 1, + completedAt: 2, + latencyMs: 1, + status: 'completed', + usageBasis: 'reported', + inputTokens: 10, + outputTokens: 5, + costBasis: 'priced', + costUsd: 0.01, + }, + }); + const source = await new RuntimeReadModel({ + runStore, + runtimeEventStore, + }).getSessionView('session-source'); + // The whole copy must not throw `Cannot copy invalid model call attempt`. + 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(), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const [targetRun] = await runStore.listSessionRuns('session-target'); + assert.ok(targetRun); + const targetEvents = await runStore.readEvents('session-target', targetRun.runId); + const attempt = targetEvents.find((event) => event.type === 'model_call_attempt_recorded'); + assert.ok(attempt); + // The stranded nested identity is repaired to the target, not carried over. + assert.equal(attempt.data?.sessionId, 'session-target'); + assert.equal(attempt.data?.runId, targetRun.runId); + assert.equal(attempt.data?.attemptId, attempt.id); + assert.notEqual(attempt.data?.attemptId, 'attempt-grandparent'); + assert.notEqual(attempt.data?.logicalCallId, 'logical-grandparent'); + assert.notEqual(attempt.data?.traceId, 'trace-grandparent'); + // The repaired record decodes and its identity matches the envelope the + // ledger projects it under. + const legacyDecoded = decodeModelCallAttempt(attempt.data); + assert.equal(legacyDecoded.sessionId, attempt.sessionId); + assert.equal(legacyDecoded.runId, attempt.runId); + assert.equal(legacyDecoded.attemptId, attempt.id); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('conversation copy clones one terminal Runtime ledger with new owned identities', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-conversation-runtime-copy-')); try { diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index d013f0b7d7..79877010b0 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -30,6 +30,10 @@ import { markPersisted } from '@maka/core/persisted-value'; import type { StoredMessage } from '@maka/core/session'; import { decodePersistedToolResultContent } from '@maka/core/tool-result-record-schema'; import { isEmittedAgentRunEventType, isSessionInlineRun } from '@maka/core/agent-run'; +import { + decodeModelCallAttempt, + MODEL_CALL_ATTEMPT_EVENT_TYPE, +} from '@maka/core/model-call-attempt'; import { TOOL_RECOVERY_DECISION_FACT_KIND } from '@maka/core/tool-recovery-fact'; import { buildHistoryCompactCheckpoint, @@ -320,6 +324,7 @@ export async function cloneConversationRuntimeLedger( ), ); const providerTraceIds = providerTraceIdMap(flattenedPlans, input.newId); + const logicalCallIds = logicalModelCallIdMap(flattenedPlans, input.newId); const operationIds = toolOperationIdMap(flattenedPlans, targetInvocationIds); const references: ConversationCopyReferenceMap = { ...input.referenceMap, @@ -369,6 +374,7 @@ export async function cloneConversationRuntimeLedger( checkpointIds, operationalEventIds, providerTraceIds, + logicalCallIds, ); return clonedEvent ? [clonedEvent] : []; }); @@ -629,6 +635,7 @@ function cloneAgentRunEvent( checkpointIds: Map, operationalEventIds: ReadonlyMap, providerTraceIds: ReadonlyMap, + logicalCallIds: ReadonlyMap, ): EmittedAgentRunEvent | null { if (event.type === 'event_corrupt') { throw new Error(`Cannot copy corrupt AgentRun event ${event.id}`); @@ -649,6 +656,14 @@ function cloneAgentRunEvent( operationalEventIds, providerTraceIds, ); + } else if (event.type === MODEL_CALL_ATTEMPT_EVENT_TYPE) { + data = rewriteModelCallAttempt( + event, + { sessionId: ids.sessionId, runId: ids.runId, attemptId: ids.eventId }, + references, + providerTraceIds, + logicalCallIds, + ); } else if (event.type === 'history_compact_checkpoint_recorded') { const sourceCheckpoint = event.data?.checkpoint; if (!validateHistoryCompactCheckpointShape(sourceCheckpoint, event.sessionId)) { @@ -760,6 +775,49 @@ function rewriteProviderRequestAttempt( }; } +function rewriteModelCallAttempt( + event: AgentRunEvent, + ids: { + readonly sessionId: string; + readonly runId: string; + readonly attemptId: string; + }, + references: ConversationCopyReferenceMap, + providerTraceIds: ReadonlyMap, + logicalCallIds: ReadonlyMap, +): Record { + // A ModelCallAttempt is an accounting authority whose payload identity is its + // portable source of truth and must agree with the rewritten envelope. Leaving + // the source `sessionId`/`runId` in place makes the model-call projection + // reject the attempt as unreadable (its envelope now disagrees), and reusing + // the source `attemptId` — the ledger's global primary key — lets the copy + // overwrite the source session's own row. Rewrite the owned identity the same + // way the sibling provider-request rewriters do. + // + // Unlike those siblings, do NOT require `attempt.attemptId === event.id`. A + // well-formed writer emits them equal, but the pre-fix copy path had no + // rewriter for this event, so it rewrote the envelope id while leaving the + // nested payload at the source identity. Sessions copied before that fix carry + // attempts whose nested `attemptId` disagrees with their envelope; asserting + // the writer contract on the *source* would strand them — they could never be + // copied again. The rewrite below reassigns the identity wholesale, so a stale + // nested identity is repaired rather than trusted and the *output* still + // satisfies the `event.id === attemptId` contract. `decodeModelCallAttempt` + // still rejects a schema-invalid payload. + const attempt = decodeModelCallAttempt(event.data); + return { + ...attempt, + sessionId: ids.sessionId, + runId: ids.runId, + attemptId: ids.attemptId, + logicalCallId: requiredMappedId(logicalCallIds, attempt.logicalCallId, 'logical model call'), + traceId: requiredMappedId(providerTraceIds, attempt.traceId, 'provider trace'), + ...(attempt.captureArtifactId !== undefined + ? { captureArtifactId: rewriteOwnedArtifactId(attempt.captureArtifactId, references) } + : {}), + }; +} + function providerRequestCapture(event: AgentRunEvent): Record & { readonly traceId: string; readonly captureId: string; @@ -838,7 +896,8 @@ function providerTraceIdMap( for (const event of operationalEvents) { if ( event.type !== 'provider_request_captured' && - event.type !== 'provider_request_attempt_recorded' + event.type !== 'provider_request_attempt_recorded' && + event.type !== MODEL_CALL_ATTEMPT_EVENT_TYPE ) { continue; } @@ -849,6 +908,23 @@ function providerTraceIdMap( return result; } +function logicalModelCallIdMap( + plans: readonly { readonly operationalEvents: readonly AgentRunEvent[] }[], + newId: () => string, +): Map { + const result = new Map(); + for (const { operationalEvents } of plans) { + for (const event of operationalEvents) { + if (event.type !== MODEL_CALL_ATTEMPT_EVENT_TYPE) continue; + const logicalCallId = event.data?.logicalCallId; + if (typeof logicalCallId === 'string' && !result.has(logicalCallId)) { + result.set(logicalCallId, newId()); + } + } + } + return result; +} + function toolOperationIdMap( plans: readonly { readonly run: AgentRunHeader;