Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
5 changes: 5 additions & 0 deletions packages/runtime-host/src/server/canonical-turn-snapshot.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,11 @@ export async function readCanonicalTurnSnapshot(
const terminal = classifyTerminalRuntimeLedger(logical.tip, logical.events);
if (terminal.kind === 'fact') {
const fact = terminal.fact;
if (fact.runStatus === 'handed_off') {
// The pause ends this run, not the turn: the logical execution continues
// in the handoff successor.
return { sessionId, turnId, runId, status: 'running', ...rootExecutionKind };
}
if (fact.runStatus === 'completed') {
const contextCompactionOutcome = readContextCompactionOutcome(
fact.terminalEvent.actions?.stateDelta?.contextCompactionOutcome,
Expand Down
36 changes: 35 additions & 1 deletion packages/runtime-host/src/server/session-revision-coordinator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import { createHash, randomUUID } from 'node:crypto';
import { SIDE_CONVERSATION_SESSION_LABEL } from '@maka/core/side-conversation';
import type { CreateSessionInput } from '@maka/core/runtime-inputs';
import { redactSecrets } from '@maka/core/redaction';
import {
isWorkHubCoordinationSessionId,
isWorkHubCoordinationSessionTarget,
Expand All @@ -39,6 +40,7 @@ import {
type ConversationRuntimeLedgerCopyPlan,
} from '@maka/runtime/conversation-copy';
import { type SessionManager } from '@maka/runtime/session-manager';
import { RuntimeReadModelError } from '@maka/runtime/runtime-read-model';
import {
authenticateInteractiveArtifactStoreWriter,
type InteractiveArtifactStoreWriter,
Expand Down Expand Up @@ -345,7 +347,10 @@ export class HostSessionRevisionCoordinator {
input.sourceTurnId === undefined
? { messages: [], events: [] }
: await this.options.manager.readConversationCopySnapshot(input.sourceSessionId);
} catch {
} catch (error) {
console.error(
`[runtime-host] session conversation copy could not read the source ledger (${input.sourceSessionId}): ${redactSecrets(describeConversationCopyFailure(error))}`,
);
return copyFailure('persistence_failed', 'Source conversation ledger is unavailable');
}
// An empty copy carries no source transcript: skip the slice so a side
Expand Down Expand Up @@ -384,6 +389,9 @@ export class HostSessionRevisionCoordinator {
'Session conversation copy does not yet support continuation authority facts',
);
}
console.error(
`[runtime-host] session conversation copy could not prepare the source lineage (${input.sourceSessionId}): ${redactSecrets(describeConversationCopyFailure(error))}`,
);
return copyFailure('persistence_failed', 'Source conversation lineage is unavailable');
}
if (
Expand Down Expand Up @@ -837,6 +845,32 @@ function persistedConversationCopyKind(kind: ConversationCopySemanticKind): Conv
return kind === 'revision' ? 'revision' : 'branch';
}

const LOGGED_COPY_FAILURE_DIAGNOSTICS = 5;

/**
* The identity a RuntimeReadModelError carries in its diagnostics, not in its
* message: a bounded, redacted list of codes and event/run/turn ids.
*/
function describeConversationCopyFailure(error: unknown): string {
const message = error instanceof Error ? error.message : String(error);
if (!(error instanceof RuntimeReadModelError) || error.diagnostics.length === 0) {
return message;
}
const shown = error.diagnostics.slice(0, LOGGED_COPY_FAILURE_DIAGNOSTICS);
const summary = shown
.map((diagnostic) =>
[
diagnostic.code,
...(diagnostic.eventId ? [`event=${diagnostic.eventId}`] : []),
...(diagnostic.runId ? [`run=${diagnostic.runId}`] : []),
...(diagnostic.turnId ? [`turn=${diagnostic.turnId}`] : []),
].join(' '),
)
.join('; ');
const rest = error.diagnostics.length - shown.length;
return `${message} [${summary}${rest > 0 ? `; +${rest} more` : ''}]`;
}

function copySuccess(result: SessionConversationCopyResult): ConversationCopyOutcome {
return { ok: true, result };
}
Expand Down
33 changes: 33 additions & 0 deletions packages/runtime/src/__tests__/runtime-event-read-model.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import type { SessionHeader, SessionSummary, StoredMessage, TurnRecord } from '@
import { deriveTurnRecords, decodeCanonicalMessage } from '@maka/core/session';
import type { CanonicalPermissionOutcomeRecord } from '../interaction-authority.js';
import {
classifyRuntimeEventTerminalFact,
createRuntimeEventStoredMessageProjector,
isHardRuntimeEventReadModelDiagnostic,
isUnclaimedRuntimeEventDiagnostic,
Expand Down Expand Up @@ -2102,6 +2103,38 @@ describe('projectRuntimeEventsToStoredMessages', () => {
assert.deepStrictEqual(out.diagnostics, []);
});

// The exact shape AgentRun.reachHandoffBoundary commits: endInvocation plus
// the pause, and deliberately no terminal status. Ledgers written before the
// classifier learned this shape carry this event, immutable.
test('handoff-pause terminal RuntimeEvent classifies as handed off', () => {
const pauseEvent = ev({
id: 'evt-handoff-pause',
author: 'host',
modelVisibility: 'hidden',
actions: {
endInvocation: true,
handoffPause: {
claimId: 'claim-1',
handoffId: 'handoff-1',
hostEpoch: 'epoch-1',
protocol: 'runtime_handoff_pause_v1',
remainingSteps: null,
rootRunId: runId,
successorInvocationId: 'successor-inv-1',
successorRunId: 'successor-run-1',
},
},
});
const result = classifyRuntimeEventTerminalFact({ sessionId, runId, turnId }, [
ev({ id: 'evt-open' }),
pauseEvent,
]);
assert.equal(result.fact?.runStatus, 'handed_off');
assert.equal(result.fact?.turnStatus, 'handed_off');
assert.equal(result.fact?.terminalEvent.id, 'evt-handoff-pause');
assert.deepEqual(result.diagnostics, []);
});

test('projects tool_call stepId from refs so the UI timeline keeps step pairing', () => {
const stepCall = (id: string, stepId?: string) =>
ev({
Expand Down
213 changes: 211 additions & 2 deletions packages/runtime/src/__tests__/runtime-handoff.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -259,8 +259,9 @@ for (const decision of [
prepareConversationRuntimeLedgerCopy({
sourceSessionId: session.id,
sourceEvents,
// The paused run carries no terminal status yet, so the read model
// refuses to project the Session; the copy only reads the Turn its
// The read model classifies the pause as handed_off, but the copy
// preflight deliberately refuses handoff lineage until typed
// identity rewriting lands; the copy only reads the Turn its
// messages name.
copiedMessages: sourceEvents
.map((event) => projectRuntimeEventUserMessage(event, event.id))
Expand Down Expand Up @@ -371,3 +372,211 @@ for (const decision of [
}
});
}

test('chained handoff parks planning at the original boundary as continuation_already_exists', {
timeout: 10_000,
}, async () => {
const root = await mkdtemp(join(tmpdir(), 'maka-runtime-handoff-chain-'));
const owner = await tryAcquireInteractiveRootOwner(
await resolveStorageRoot({ path: root, kind: 'interactive' }),
);
assert.ok(owner);
const store = createSessionStore(root);
const runStore = createSqliteAgentRunStore(root);
const planStore = await openInteractivePlanStoreForWrite(owner.lease);
const runtimeEventStore = createSqliteRuntimeStore(join(root, 'runtime.sqlite'));
const originalEntered = deferred<void>();
const originalGo = deferred<void>();
const continuationEntered = deferred<void>();
const continuationGo = deferred<void>();
const executionAbort = new AbortController();
const backends = new BackendRegistry();
backends.register('ai-sdk', (ctx) => {
return new (class extends FakeBackend {
async prepareRunComposition(input: { runId: string; turnId: string }): Promise<void> {
await ctx.recordRunComposition!(
input.runId,
createRunCompositionSnapshot({
composerId: 'test.handoff',
composerRevision: '1',
sourceRevisions: [],
baseSystemPromptHash: `sha256:${'0'.repeat(64)}`,
toolCatalogHash: `sha256:${'0'.repeat(64)}`,
toolAvailabilityHash: `sha256:${'0'.repeat(64)}`,
baseProviderOptionsHash: `sha256:${'0'.repeat(64)}`,
toolNames: [],
contextWindow: null,
}),
);
}

override async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
if (!input.continuation) {
await this.prepareRunComposition({ runId: input.runId!, turnId: input.turnId });
originalEntered.resolve();
await originalGo.promise;
const result = await input.handoffBoundary!(executionAbort.signal, 2);
if (result === 'pause' && !executionAbort.signal.aborted) return;
} else {
continuationEntered.resolve();
await continuationGo.promise;
const result = await input.handoffBoundary!(executionAbort.signal, null);
if (result === 'pause' && !executionAbort.signal.aborted) return;
}
yield {
type: 'complete',
id: `complete-${input.runId}`,
turnId: input.turnId,
ts: Date.now(),
stopReason: 'end_turn',
};
}

override async stop(): Promise<void> {
executionAbort.abort();
}
})({ sessionId: ctx.sessionId });
});
let id = 0;
const deps = {
store,
runStore,
runtimeEventStore,
planStore,
backends,
safeBoundaryResumeEnabled: true,
toolBoundaryProtocol: 't1_after_preflight_v1' as const,
inspectContinuationSafety: async () => ({
workspaceIdentity: 'workspace-1',
backgroundOperationsSettled: true,
availableToolNames: [] as string[],
}),
newId: () => `id-${++id}`,
now: Date.now,
};
const kernel = new RuntimeKernel(deps);
const manager = new SessionManager({ ...deps, runtimeKernel: kernel });
const session = await manager.createSession({
cwd: root,
llmConnectionSlug: 'fake',
model: 'fake-model',
permissionMode: 'ask',
name: 'handoff-chain',
});
try {
await planStore.submitProposal({
sessionId: session.id,
turnId: 'proposal-turn',
title: 'work',
steps: [{ id: 'step-1', title: 'finish', description: 'finish the logical task' }],
});
const proposal = (await planStore.readState(session.id)).proposals[0]!;
await planStore.approveProposal({
sessionId: session.id,
proposalId: proposal.proposalId,
expectedRevision: proposal.revision,
});
const running = (async () => {
for await (const _ of manager.sendMessage(
session.id,
{
turnId: 'logical-turn',
text: 'continue work',
origin: { kind: 'goal', goalId: 'goal-1' },
},
{ runId: 'original-run', durability: 'required' },
)) {
void _;
}
})();
void running.catch(() => {});
await originalEntered.promise;
const pause1 = {
protocol: 'runtime_handoff_pause_v1' as const,
handoffId: 'handoff-1',
hostEpoch: 'host-1',
rootRunId: 'original-run',
successorRunId: 'successor-run',
successorInvocationId: 'successor-run',
claimId: 'chained-claim-1',
};
const request1 = kernel.requestRunHandoff(
session.id,
'original-run',
pause1,
new AbortController().signal,
);
assert.ok(request1);
originalGo.resolve();
assert.equal(await request1.ready, true);
assert.equal(request1.commit(), true);
await running;
assert.deepEqual(
runtimeHandoffPause(
(await runtimeEventStore.readRunInvocation(session.id, 'original-run'))!.terminalEvent!,
),
{ ...pause1, remainingSteps: 2 },
);
await manager.recoverInterruptedSessions();

const plan1 = await manager.planAuthoritativeSafeBoundaryContinuation(session.id, {
sourceRunId: 'original-run',
purpose: 'handoff',
});
assert.ok(plan1.continuation, JSON.stringify(plan1));

const resume = (async () => {
for await (const _ of manager.resumeSafeBoundaryContinuation(plan1.continuation!)) {
void _;
}
})();
void resume.catch(() => {});
await continuationEntered.promise;
const pause2 = {
protocol: 'runtime_handoff_pause_v1' as const,
handoffId: 'handoff-2',
hostEpoch: 'host-2',
rootRunId: 'original-run',
successorRunId: 'successor-run-2',
successorInvocationId: 'successor-run-2',
claimId: 'chained-claim-2',
};
const request2 = kernel.requestRunHandoff(
session.id,
'successor-run',
pause2,
new AbortController().signal,
);
assert.ok(request2);
continuationGo.resolve();
assert.equal(await request2.ready, true);
assert.equal(request2.commit(), true);
await resume;
assert.deepEqual(
runtimeHandoffPause(
(await runtimeEventStore.readRunInvocation(session.id, 'successor-run'))!.terminalEvent!,
),
{ ...pause2, remainingSteps: null },
);

// The continuation target itself handed off: planning the same handoff at
// the original boundary finds a terminal continuation (handed_off), not
// an ambiguous one that needs repair.
const chainedPlan = await manager.planAuthoritativeSafeBoundaryContinuation(session.id, {
sourceRunId: 'original-run',
purpose: 'handoff',
});
assert.equal(chainedPlan.disposition, 'park');
assert.deepEqual(chainedPlan.rejectionReasons, ['continuation_already_exists']);
} finally {
originalGo.resolve();
continuationGo.resolve();
await kernel.disposeBackend(session.id).catch(() => {});
runtimeEventStore.close();
planStore.close();
await runStore.close?.();
await store.close?.();
await owner.close();
await rm(root, { recursive: true, force: true });
}
});
6 changes: 6 additions & 0 deletions packages/runtime/src/conversation-copy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -813,6 +813,12 @@ export async function cloneConversationRuntimeLedger(
}

if (plan.terminal.kind === 'fact' && terminalEvent) {
// The preflight rejects handoff lineage before a plan reaches cloning,
// so a handed-off run never gets here; the guard keeps the outcome type
// honest at the boundary.
if (plan.terminal.fact.runStatus === 'handed_off') {
throw new Error(`Copied AgentRun ${plan.run.runId} carries unresolved handoff lineage`);
}
await commitTerminalRunWithRuntimeFact({
runtimeEventStore: input.runtimeEventStore,
newId: input.newId,
Expand Down
Loading