Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 108 additions & 0 deletions packages/runtime/src/__tests__/conversation-copy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -1266,6 +1267,113 @@ 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 clones one terminal Runtime ledger with new owned identities', async () => {
const root = await mkdtemp(join(tmpdir(), 'maka-conversation-runtime-copy-'));
try {
Expand Down
70 changes: 69 additions & 1 deletion packages/runtime/src/conversation-copy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -369,6 +374,7 @@ export async function cloneConversationRuntimeLedger(
checkpointIds,
operationalEventIds,
providerTraceIds,
logicalCallIds,
);
return clonedEvent ? [clonedEvent] : [];
});
Expand Down Expand Up @@ -629,6 +635,7 @@ function cloneAgentRunEvent(
checkpointIds: Map<string, string>,
operationalEventIds: ReadonlyMap<string, string>,
providerTraceIds: ReadonlyMap<string, string>,
logicalCallIds: ReadonlyMap<string, string>,
): EmittedAgentRunEvent | null {
if (event.type === 'event_corrupt') {
throw new Error(`Cannot copy corrupt AgentRun event ${event.id}`);
Expand All @@ -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)) {
Expand Down Expand Up @@ -760,6 +775,41 @@ function rewriteProviderRequestAttempt(
};
}

function rewriteModelCallAttempt(
event: AgentRunEvent,
ids: {
readonly sessionId: string;
readonly runId: string;
readonly attemptId: string;
},
references: ConversationCopyReferenceMap,
providerTraceIds: ReadonlyMap<string, string>,
logicalCallIds: ReadonlyMap<string, string>,
): Record<string, unknown> {
// 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.
const attempt = decodeModelCallAttempt(event.data);
if (attempt.attemptId !== event.id) {
throw new Error(`Cannot copy invalid model call attempt ${event.id}`);
}
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<string, unknown> & {
readonly traceId: string;
readonly captureId: string;
Expand Down Expand Up @@ -838,7 +888,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;
}
Expand All @@ -849,6 +900,23 @@ function providerTraceIdMap(
return result;
}

function logicalModelCallIdMap(
plans: readonly { readonly operationalEvents: readonly AgentRunEvent[] }[],
newId: () => string,
): Map<string, string> {
const result = new Map<string, string>();
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;
Expand Down