Google GitGoogle Git
Sign in
apache/maka/refs/heads/codex/cli-graph/./packages/runtime/src/runtime-kernel.ts
blob: f7c2240d8922168cf604d79a73c671e81c0b3aa5 [file]
import type {
AgentRunHeader,
AgentRunStore,
RuntimeEvent,
RuntimeEventStore,
ToolBoundaryProtocol,
} from '@maka/core';
import { isPermissionModeWithinCeiling, isSessionInlineRun } from '@maka/core';
import type {
CompleteEvent,
QueueEnqueueOutcome,
QueueUpdateEvent,
SessionEvent,
TokenUsageEvent,
} from '@maka/core/events';
import type {
SessionBlockedReason,
SessionHeader,
SessionStatus,
StoredMessage,
SystemNoteMessage,
TurnRecord,
TurnStateMessage,
} from '@maka/core/session';
import { isDeepStrictEqual } from 'node:util';
import type { ChildAgentTurnInput, UserMessageInput } from '@maka/core/runtime-inputs';
import type { PermissionResponse } from '@maka/core/permission';
import {
resolveEffectiveOrchestration,
type EffectiveOrchestration,
} from '@maka/core/orchestration';
import type { UserQuestionResponse } from '@maka/core/user-question';
import {
AgentRun,
type AgentRunActiveSession,
type AgentRunBeginResult,
type AgentRunDurability,
type AgentRunLineage,
type RuntimeContinuationFailpoint,
} from './agent-run.js';
import { AiSdkFlow, mapSessionEventToRuntimeEvent } from './ai-sdk-flow.js';
import type { AgentBackend, SteeringLease } from '@maka/core/backend-types';
import type { AgentTeamExecutionContext, MakaTool } from './tool-runtime.js';
import type {
InvocationContext,
InvocationResult,
InvocationSource,
} from './invocation-context.js';
import { RuntimeRunner } from './runtime-runner.js';
import type {
BackendRegistry,
CompactSessionInput,
SessionStore,
StopSessionInput,
} from './session-manager.js';
import type { ShellRunProcessManager } from './shell-run-manager.js';
import {
buildStatusPatch,
buildTurnStateMessage,
normalizeStopSessionSource,
turnHasRetainedOutput as messagesHaveRetainedOutput,
} from './session-projection-helpers.js';
import { assertAgentDefinitionRunnable, buildToolsForAgentDefinition } from './agent-catalog.js';
import { parseExpertAgentId, requireResolvedAgentDefinition } from './expert-catalog.js';
import { loadLatestHistoryCompactCheckpointFromRunLedger } from './history-compact-ledger.js';
import {
canReplaceHistoryCompactCheckpoint,
type HistoryCompactCheckpoint,
} from './history-compact-checkpoint.js';
import { shouldAppendContextCompactionFailedOpenNote } from './context-budget.js';
import {
buildResumePlanFromRuntimeEvents,
RuntimeContinuationRevalidationError,
type RuntimeContinuation,
type RuntimeContinuationSafetyObservation,
} from './runtime-resume.js';
import {
matchingTerminalRuntimeEvents,
terminalRunStatusFromRuntimeEvent,
} from './terminal-run-commit.js';
import {
RuntimeMessageAuthorityInvariantError,
type RuntimeMessageAuthority,
type RuntimeMessageRunOwner,
} from './message-authority.js';
import {
RuntimeInteractionFailStopError,
RuntimeInteractionInvariantError,
bindRuntimeInteractionRun,
type RuntimeInteractionAuthority,
type RuntimeInteractionRunBinding,
type RuntimeInteractionRunClosureReason,
} from './interaction-authority.js';
export interface RuntimeKernelLike {
claimExecution(sessionId: string): RuntimeExecutionClaim;
startTurn(
sessionId: string,
input: UserMessageInput,
options?: TurnStartOptions,
): AsyncIterable<SessionEvent>;
resumeContinuation?(continuation: RuntimeContinuation): AsyncIterable<SessionEvent>;
compactSession(sessionId: string, input?: CompactSessionInput): AsyncIterable<SessionEvent>;
startChildTurn(
sessionId: string,
input: ChildAgentTurnInput,
execution?: RuntimeExecutionClaim,
): AsyncIterable<SessionEvent>;
startChildRetry?(
sessionId: string,
input: ChildAgentRetryInput,
execution?: RuntimeExecutionClaim,
): AsyncIterable<SessionEvent>;
stopSession(sessionId: string, input?: StopSessionInput): Promise<void>;
respondToPermission(sessionId: string, response: PermissionResponse): Promise<void>;
respondToUserQuestion?(sessionId: string, response: UserQuestionResponse): Promise<void>;
/** Queue a user message for mid-turn injection at the next step boundary. */
steer(sessionId: string, text: string): QueueEnqueueOutcome;
/** Queue a user message to open the turn after the current one finishes. */
queueMessage(sessionId: string, text: string): QueueEnqueueOutcome;
/** Drain the followup queue into one `\n\n`-joined prompt, or null if empty. */
drainFollowup(sessionId: string): string | null;
/** Take back every queued message (both queues) as one `\n\n`-joined string. */
retractQueue(sessionId: string): string;
hasActiveRuns(sessionId: string): boolean;
hasActiveRun?(sessionId: string, runId: string, turnId?: string): boolean;
updateCachedHeader(sessionId: string, header: SessionHeader): void;
invalidateBackend(sessionId: string): Promise<void>;
invalidateCachedBackends(): Promise<void>;
disposeBackend(sessionId: string): Promise<void>;
}
export interface TurnStartOptions {
runId?: string;
userMessageId?: string;
durability?: AgentRunDurability;
/**
* Resolve turn admission after this Session has registered a pending start
* and immediately before AgentRun begins durable/Backend activation.
*/
admitTurn?: () => Promise<'admitted' | 'cancelled'>;
onRunStarted?: (runId: string, initialHeader: SessionHeader) => void | Promise<void>;
execution?: RuntimeExecutionClaim;
}
export interface RuntimeExecutionClaim {
readonly sessionId: string;
readonly stopSignal: AbortSignal;
isStopRequested(): boolean;
release(): void;
}
export class RuntimeOwnerCleanupError extends Error {
readonly name = 'RuntimeOwnerCleanupError';
constructor(message: string, cause: unknown) {
super(message, { cause });
}
}
export interface ChildAgentRetryInput {
parentRunId: string;
spec: ChildAgentTurnInput['spec'];
continuation: RuntimeContinuation;
/** Retry an ordinary session-inline AgentRun inside a linked child Session. */
linkedSession?: boolean;
onRunStarted?: () => void | Promise<void>;
}
/**
* An embedded session's authoritative pending-message queues plus its event
* sink. Hosted composition never creates this state; its Host owns admission,
* snapshots, leases, and follow-up drain.
*/
interface PendingSteeringMessage extends SteeringLease {}
/**
* A pulled lease is bound to the turn that pulled it: only the issuing turn's
* backend can settle it (ack/nack stay valid even after ownership moved to an
* overlapping turn — invalidating a delivered lease would leave it in-flight
* and redeliver an already-executed message), and no other turn's retract/
* clear/release may reclaim it while its delivery is still undetermined.
*/
interface LeasedSteeringMessage extends PendingSteeringMessage {
issuingTurnId: string;
}
interface SessionSteeringState {
/** Messages waiting to be injected into the running turn at a step boundary. */
steering: PendingSteeringMessage[];
/**
* Leased to the running turn's backend but not yet settled. pull() is the
* single atomic commit point: an in-flight lease is committed to that
* turn's delivery — retract/clear reclaim only QUEUED messages — and it
* settles exactly once, decided solely by the persistence fact: ack when
* the steering event is durably consumed (even under abort), nack when it
* provably never persisted. Snapshots count in-flight as still pending so
* the UI keeps showing the message until it lands in the transcript.
*/
inFlight: LeasedSteeringMessage[];
/** Messages waiting to open the next turn. */
followup: string[];
/** Pushes a `queue_update` into the active turn's stream; unset when idle. */
sink?: (event: QueueUpdateEvent) => void;
activeTurnId?: string;
}
export type BackendActivationBoundary = <T>(operation: () => Promise<T> | T) => Promise<T>;
export interface RuntimeKernelDeps {
store: SessionStore;
runStore?: AgentRunStore;
runtimeEventStore?: RuntimeEventStore;
/** Host capability; each run still gates it by the selected backend. */
toolBoundaryProtocol?: ToolBoundaryProtocol;
backends: BackendRegistry;
newId: () => string;
now: () => number;
childTools?: readonly MakaTool[];
runtimeSource?: InvocationSource;
runtimeInvocationObserver?: (result: InvocationResult) => void | Promise<void>;
repairRunRuntimeLedger?: (sessionId: string, runId: string) => Promise<boolean>;
shellRuns?: ShellRunProcessManager;
cleanupHistoryCompactArtifacts?: (input: HistoryCompactCleanupRequest) => Promise<void>;
inspectContinuationSafety?: (sessionId: string) => Promise<RuntimeContinuationSafetyObservation>;
safeBoundaryResumeEnabled?: boolean;
continuationFailpoint?: (point: RuntimeContinuationFailpoint) => Promise<void>;
runBackendActivation?: BackendActivationBoundary;
/** Hosted composition capability. When present, the Host owns all message queues. */
messageAuthority?: RuntimeMessageAuthority;
/** Hosted composition capability. Omit for embedded interaction ownership. */
interactionAuthority?: RuntimeInteractionAuthority;
}
export interface HistoryCompactCleanupRequest {
sessionId: string;
checkpoint: HistoryCompactCheckpoint;
runtimeEvents: readonly RuntimeEvent[];
}
interface BackendGeneration extends AgentRunActiveSession {
sessionId: string;
generation: number;
route: { kind: 'parent' } | { kind: 'child'; activeKey: string };
phase: 'active' | 'stopping' | 'disposing' | 'failed' | 'terminated';
backend: AgentBackend;
stopBackend: AgentBackend['stop'];
stopState:
| { kind: 'idle' }
| { kind: 'pending'; task: Promise<void> }
| { kind: 'failed'; error: unknown };
disposal?: Promise<BackendDisposalOutcome>;
disposalFailure?: Error;
cachedHeader: SessionHeader;
activeRuns: Map<string, AgentRun>;
turnToRunId: Map<string, string>;
}
interface StopTarget {
active?: BackendGeneration;
readonly generation: number;
readonly runs: Map<string, StopRunTarget>;
delivery: { kind: 'pending' } | { kind: 'delivered' } | { kind: 'failed'; error: unknown };
}
interface StopRunTarget {
run?: AgentRun;
readonly runId: string;
readonly turnId: string;
readonly lineage: AgentRunLineage;
readonly sessionInline: boolean;
stopCompleted: boolean;
}
interface StopOperation {
abortSource: string | undefined;
ts: number;
statusProjected: boolean;
turnProjections: Map<
string,
{
id: string;
turnId: string;
lineage: AgentRunLineage;
message?: TurnStateMessage;
projected: boolean;
}
>;
abortNote: SystemNoteMessage;
abortNoteProjected: boolean;
targets: Map<number, StopTarget>;
queue: Promise<void>;
}
interface SessionStopIntent {
input: StopSessionInput;
readonly claims: Set<PendingExecutionClaim>;
}
type ExecutionClaimOutcome = { ok: true } | { ok: false; error: unknown };
interface PendingExecutionClaim {
readonly handle: RuntimeExecutionClaim;
readonly sessionId: string;
readonly abortController: AbortController;
readonly cancellation: RuntimeExecutionCancellation;
readonly settled: Promise<void>;
resolveSettled(): void;
rejectSettled(error: unknown): void;
phase: 'pending' | 'attached' | 'reserved' | 'released' | 'failed';
run?: AgentRun;
stopIntent?: SessionStopIntent;
finalization?: ExecutionClaimOutcome;
}
type BackendDisposalOutcome = { ok: true } | { ok: false; error: unknown };
interface BackendInvalidationState {
readonly outcome: Promise<BackendDisposalOutcome>;
resolve(outcome: BackendDisposalOutcome): void;
disposal?: Promise<void>;
failure?: Error;
}
export class RuntimeKernel implements RuntimeKernelLike {
private readonly active = new Map<string, BackendGeneration>();
private readonly childActive = new Map<string, BackendGeneration>();
private readonly backendGenerations = new Map<number, BackendGeneration>();
private readonly backendActivationBuilds = new Map<string, Promise<BackendGeneration>>();
private readonly stopOperations = new Map<string, StopOperation>();
private readonly stopAttempts = new Map<string, Promise<void>>();
private readonly executionClaims = new Map<string, Set<PendingExecutionClaim>>();
private readonly executionClaimStates = new WeakMap<
RuntimeExecutionClaim,
PendingExecutionClaim
>();
private readonly stopIntents = new Map<string, SessionStopIntent>();
private readonly historyCompactCheckpoints = new Map<
string,
HistoryCompactCheckpoint | undefined
>();
private readonly historyCompactCheckpointLoads = new Map<
string,
Promise<HistoryCompactCheckpoint | undefined>
>();
private readonly historyCompactCheckpointWrites = new Map<string, Promise<void>>();
private readonly historyCompactCleanupWrites = new Map<string, Promise<void>>();
private readonly pendingContinuationClaims = new Set<string>();
private readonly pendingContinuationSessions = new Set<string>();
private readonly steeringBySession = new Map<string, SessionSteeringState>();
private readonly backendInvalidations = new Map<string, BackendInvalidationState>();
private nextBackendGeneration = 0;
private readonly interactionRuns = new Map<AgentRun, RuntimeInteractionRunBinding>();
constructor(private readonly deps: RuntimeKernelDeps) {
if (deps.runStore && !deps.runtimeEventStore) {
throw new Error('RuntimeEventStore is required when AgentRunStore is configured');
}
}
private async runBackendActivation<T>(operation: () => Promise<T> | T): Promise<T> {
return await (this.deps.runBackendActivation?.(operation) ?? operation());
}
claimExecution(sessionId: string): RuntimeExecutionClaim {
if (this.stopIntents.has(sessionId)) {
throw new Error(`Session ${sessionId} is stopping and cannot admit a new execution`);
}
let resolveSettled!: () => void;
let rejectSettled!: (error: unknown) => void;
const settled = new Promise<void>((resolve, reject) => {
resolveSettled = resolve;
rejectSettled = reject;
});
// A failed claim may have no concurrent stop subscriber; stop still observes this same promise.
void settled.catch(() => undefined);
const abortController = new AbortController();
const cancellation = new RuntimeExecutionCancellation(sessionId);
const handle: RuntimeExecutionClaim = {
sessionId,
stopSignal: abortController.signal,
isStopRequested: () => state.stopIntent !== undefined,
release: () => this.releaseExecutionClaim(state),
};
const state: PendingExecutionClaim = {
handle,
sessionId,
abortController,
cancellation,
settled,
resolveSettled,
rejectSettled,
phase: 'pending',
};
let claims = this.executionClaims.get(sessionId);
if (!claims) {
claims = new Set();
this.executionClaims.set(sessionId, claims);
}
claims.add(state);
this.executionClaimStates.set(handle, state);
return handle;
}
private takeExecutionClaim(
sessionId: string,
supplied?: RuntimeExecutionClaim,
): PendingExecutionClaim {
const handle = supplied ?? this.claimExecution(sessionId);
const state = this.executionClaimStates.get(handle);
if (!state || state.sessionId !== sessionId || state.phase !== 'pending') {
throw new Error(`Execution claim does not own pending admission for session ${sessionId}`);
}
return state;
}
private attachExecutionClaim(execution: PendingExecutionClaim, run: AgentRun): void {
if (execution.phase !== 'pending') {
throw new Error(
`Execution claim cannot attach Run ${run.runId} from phase ${execution.phase}`,
);
}
execution.run = run;
execution.phase = 'attached';
if (execution.stopIntent) run.stop(execution.stopIntent.input.source);
}
private reserveExecutionClaim(
execution: PendingExecutionClaim,
active: BackendGeneration,
run: AgentRun,
): void {
if (execution.phase !== 'attached' || execution.run !== run) {
throw new Error(`Execution claim does not own attached Run ${run.runId}`);
}
try {
if (execution.stopIntent) {
this.claimRunForStop(execution.sessionId, execution.stopIntent.input, active, run);
}
} catch (error) {
this.unregisterRun(active, run);
execution.phase = 'failed';
this.settleExecutionClaim(execution, { ok: false, error });
throw error;
}
execution.phase = 'reserved';
}
private settleReservedExecutionClaim(
execution: PendingExecutionClaim,
run: AgentRun,
outcome: ExecutionClaimOutcome,
): void {
if (
(execution.phase === 'attached' || execution.phase === 'failed') &&
execution.run === run &&
!outcome.ok
) {
return;
}
if (execution.phase !== 'reserved' || execution.run !== run) {
throw new Error(`Execution claim cannot settle reserved Run ${run.runId}`);
}
execution.phase = outcome.ok ? 'released' : 'failed';
this.settleExecutionClaim(execution, outcome);
}
private releaseExecutionClaim(execution: PendingExecutionClaim): void {
if (execution.phase !== 'pending' && execution.phase !== 'attached') return;
if (execution.phase === 'attached' && execution.stopIntent) {
this.settleStoppedAttachedExecution(execution);
return;
}
execution.phase = 'released';
this.settleExecutionClaim(execution, { ok: true });
}
private settleExecutionClaim(
execution: PendingExecutionClaim,
outcome: ExecutionClaimOutcome,
): void {
const claims = this.executionClaims.get(execution.sessionId);
claims?.delete(execution);
if (claims?.size === 0) this.executionClaims.delete(execution.sessionId);
if (outcome.ok) execution.resolveSettled();
else execution.rejectSettled(outcome.error);
}
private async finalizeExecutionClaimRun(
execution: PendingExecutionClaim,
run: AgentRun,
finalize: () => Promise<void>,
): Promise<void> {
let outcome: ExecutionClaimOutcome;
try {
await finalize();
outcome = { ok: true };
} catch (error) {
outcome = { ok: false, error };
}
if (execution.phase === 'attached' && execution.run === run) {
execution.finalization = outcome;
this.settleStoppedAttachedExecution(execution);
}
if (!outcome.ok) throw outcome.error;
}
private settleStoppedAttachedExecution(execution: PendingExecutionClaim): void {
if (execution.phase !== 'attached' || !execution.stopIntent || !execution.finalization) {
return;
}
const outcome = execution.finalization;
execution.phase = outcome.ok ? 'released' : 'failed';
this.settleExecutionClaim(execution, outcome);
}
async *startTurn(
sessionId: string,
input: UserMessageInput,
options: TurnStartOptions = {},
): AsyncIterable<SessionEvent> {
if (this.pendingContinuationSessions.has(sessionId)) {
throw new Error('Cannot start a turn while a runtime continuation is being claimed');
}
const execution = this.takeExecutionClaim(sessionId, options.execution);
try {
const header = await this.deps.store.readHeader(sessionId);
let workspaceIdentity: string | undefined;
if (this.deps.safeBoundaryResumeEnabled === true && this.deps.inspectContinuationSafety) {
try {
workspaceIdentity = (await this.deps.inspectContinuationSafety(sessionId))
.workspaceIdentity;
} catch {
// A new turn remains usable without continuation metadata. Actual
// continuation claims inspect the same facts strictly below.
}
}
const run = new AgentRun({
sessionId,
header,
userInput: input,
runId: options.runId,
userMessageId: options.userMessageId,
durability: options.durability,
store: this.deps.store,
runStore: this.deps.runStore,
runtimeEventStore: this.deps.runtimeEventStore,
...(runtimeToolBoundaryProtocol(this.deps, header)
? { toolBoundaryProtocol: runtimeToolBoundaryProtocol(this.deps, header) }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
...(workspaceIdentity ? { workspaceIdentity } : {}),
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
const active = await this.reserveParentRun(
targetSessionId,
nextHeader,
activeRun,
execution,
);
this.reserveExecutionClaim(execution, active, activeRun);
return active;
},
unregisterRun: (active, activeRun) => this.unregisterParentRun(active, activeRun),
updateHeader: (targetSessionId, patch) => this.updateHeader(targetSessionId, patch),
updateStatus: (targetSessionId, status, blockedReason, ts) =>
this.updateStatus(targetSessionId, status, blockedReason, ts),
appendTurnState: (targetSessionId, turnId, status, lineage, options) =>
this.appendTurnState(targetSessionId, turnId, status, lineage, options),
},
});
if (options.admitTurn && (await options.admitTurn()) === 'cancelled') {
throw new Error('Turn start was cancelled before runtime admission');
}
this.attachExecutionClaim(execution, run);
yield* this.runAgentTurn(
sessionId,
input,
run,
execution,
true,
options.onRunStarted,
header,
);
} finally {
this.releaseExecutionClaim(execution);
}
}
async *resumeContinuation(continuation: RuntimeContinuation): AsyncIterable<SessionEvent> {
const claimKey = [
continuation.sessionId,
continuation.sourceRunId,
continuation.sourceRuntimeEventHighWater,
].join(':');
if (this.pendingContinuationClaims.has(claimKey)) {
throw new Error('Runtime continuation source claim is already in progress');
}
if (this.pendingContinuationSessions.has(continuation.sessionId)) {
throw new Error('Runtime continuation session claim is already in progress');
}
const execution = this.takeExecutionClaim(continuation.sessionId);
this.pendingContinuationClaims.add(claimKey);
this.pendingContinuationSessions.add(continuation.sessionId);
try {
yield* this.resumeContinuationClaimed(continuation, execution);
} finally {
this.pendingContinuationClaims.delete(claimKey);
this.pendingContinuationSessions.delete(continuation.sessionId);
this.releaseExecutionClaim(execution);
}
}
private async *resumeContinuationClaimed(
continuation: RuntimeContinuation,
execution: PendingExecutionClaim,
): AsyncIterable<SessionEvent> {
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new Error('Runtime continuation requires AgentRunStore and RuntimeEventStore');
}
if (
this.hasActiveRuns(continuation.sessionId) ||
(this.executionClaims.get(continuation.sessionId)?.size ?? 0) > 1
) {
throw new Error('Cannot continue while another run is active');
}
const header = await this.deps.store.readHeader(continuation.sessionId);
const sourceRun = await this.deps.runStore.readRun(
continuation.sessionId,
continuation.sourceRunId,
);
const sourceEvents = await this.deps.runtimeEventStore.readRuntimeEvents(
continuation.sessionId,
continuation.sourceRunId,
);
assertContinuationSourceUnchanged(continuation, sourceRun, sourceEvents);
if (!this.deps.inspectContinuationSafety) {
throw new Error('Runtime continuation requires an authoritative safety inspector');
}
const observation = await this.deps.inspectContinuationSafety(continuation.sessionId);
assertContinuationSafetyUnchanged(continuation, observation);
const sessionRuns = await this.deps.runStore.listSessionRuns(continuation.sessionId);
const existingClaim = sessionRuns.find(
(runHeader) =>
runHeader.continuationSource?.sourceRunId === continuation.sourceRunId &&
runHeader.continuationSource.sourceRuntimeEventHighWater ===
continuation.sourceRuntimeEventHighWater,
);
if (existingClaim) {
throw new RuntimeContinuationRevalidationError(
'continuation_claim_conflict',
`Runtime continuation source already has a continuation child: ${existingClaim.runId}`,
);
}
const existingTarget = sessionRuns.find((runHeader) => runHeader.runId === continuation.runId);
if (existingTarget) {
throw new RuntimeContinuationRevalidationError(
'target_run_conflict',
'Runtime continuation target run already exists',
);
}
const userInput: UserMessageInput = {
turnId: continuation.turnId,
text: '',
parentRunId: continuation.sourceRunId,
parentTurnId: continuation.sourceTurnId,
};
const run = new AgentRun({
sessionId: continuation.sessionId,
header,
userInput,
runId: continuation.runId,
invocationId: continuation.invocationId,
store: this.deps.store,
runStore: this.deps.runStore,
runtimeEventStore: this.deps.runtimeEventStore,
...(runtimeToolBoundaryProtocol(this.deps, header)
? { toolBoundaryProtocol: runtimeToolBoundaryProtocol(this.deps, header) }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration: effectiveOrchestrationForRun(sourceRun, header),
continuationFailpoint: this.deps.continuationFailpoint,
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
const active = await this.reserveParentRun(
targetSessionId,
nextHeader,
activeRun,
execution,
);
this.reserveExecutionClaim(execution, active, activeRun);
return active;
},
unregisterRun: (active, activeRun) => this.unregisterParentRun(active, activeRun),
updateHeader: (targetSessionId, patch) => this.updateHeader(targetSessionId, patch),
updateStatus: (targetSessionId, status, blockedReason, ts) =>
this.updateStatus(targetSessionId, status, blockedReason, ts),
appendTurnState: (targetSessionId, turnId, status, lineage, options) =>
this.appendTurnState(targetSessionId, turnId, status, lineage, options),
},
});
this.attachExecutionClaim(execution, run);
yield* this.runAgentContinuation(continuation, run, execution);
}
async *compactSession(
sessionId: string,
input: CompactSessionInput = {},
): AsyncIterable<SessionEvent> {
const execution = this.takeExecutionClaim(sessionId);
try {
yield* this.compactSessionClaimed(sessionId, input, execution);
} finally {
this.releaseExecutionClaim(execution);
}
}
private async *compactSessionClaimed(
sessionId: string,
input: CompactSessionInput,
execution: PendingExecutionClaim,
): AsyncIterable<SessionEvent> {
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new Error('Runtime compaction requires AgentRunStore and RuntimeEventStore');
}
if (this.hasActiveRuns(sessionId)) {
throw new Error('Cannot compact while a turn is running; wait for the turn to finish.');
}
const header = await this.deps.store.readHeader(sessionId);
const turnId = input.turnId ?? this.deps.newId();
const run = new AgentRun({
sessionId,
header,
userInput: { turnId, text: '' },
store: this.deps.store,
runStore: this.deps.runStore,
runtimeEventStore: this.deps.runtimeEventStore,
...(runtimeToolBoundaryProtocol(this.deps, header)
? { toolBoundaryProtocol: runtimeToolBoundaryProtocol(this.deps, header) }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
effectiveOrchestration: resolveEffectiveOrchestration('default', undefined),
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
const active = await this.reserveParentRun(
targetSessionId,
nextHeader,
activeRun,
execution,
);
this.reserveExecutionClaim(execution, active, activeRun);
return active;
},
unregisterRun: (active, activeRun) => this.unregisterParentRun(active, activeRun),
updateHeader: (targetSessionId, patch) => this.updateHeader(targetSessionId, patch),
updateStatus: (targetSessionId, status, blockedReason, ts) =>
this.updateStatus(targetSessionId, status, blockedReason, ts),
appendTurnState: (targetSessionId, nextTurnId, status, lineage, options) =>
this.appendTurnState(targetSessionId, nextTurnId, status, lineage, options),
},
});
this.attachExecutionClaim(execution, run);
let begin: Awaited<ReturnType<typeof run.beginOperation>>;
try {
begin = await this.runBackendActivation(() => run.beginOperation());
this.settleReservedExecutionClaim(execution, run, { ok: true });
} catch (error) {
this.settleReservedExecutionClaim(execution, run, { ok: false, error });
await run.recordFailure(error);
await this.finalizeExecutionClaimRun(execution, run, () => run.finalize());
if (run.isStopped() && isExecutionCancellation(error, execution.cancellation)) return;
throw error;
}
try {
if (run.isStopped()) return;
if (!begin.backend.compactHistory)
throw new Error(`Backend ${header.backend} does not support runtime compaction`);
this.assertRunCanDispatch(run, begin.backend);
const result = await begin.backend.compactHistory({
turnId: run.turnId,
runtimeContext: begin.runtimeContext,
});
if (run.isStopped()) return;
const tokenUsageEvent: TokenUsageEvent = {
type: 'token_usage',
id: this.deps.newId(),
turnId: run.turnId,
ts: this.deps.now(),
input: 0,
output: 0,
...(result.contextBudget ? { contextBudget: result.contextBudget } : {}),
};
const completeEvent: CompleteEvent = {
type: 'complete',
id: this.deps.newId(),
turnId: run.turnId,
ts: this.deps.now(),
stopReason: 'end_turn',
};
const invocation = this.compactInvocationContext({
sessionId,
runId: run.runId,
turnId: run.turnId,
startedAt: begin.startedAt,
});
await run.acceptMappedEvent(
tokenUsageEvent,
mapSessionEventToRuntimeEvent(tokenUsageEvent, invocation),
{ requireTerminalWrite: true },
);
if (run.isStopped()) return;
await run.recordStoredSessionEvent(tokenUsageEvent);
if (run.isStopped()) return;
if (shouldAppendContextCompactionFailedOpenNote(result.contextBudget)) {
const note: SystemNoteMessage = {
type: 'system_note',
id: this.deps.newId(),
turnId: run.turnId,
ts: this.deps.now(),
kind: 'context_compaction_failed_open',
};
await this.deps.store.appendMessage(sessionId, note).catch(() => {});
}
yield tokenUsageEvent;
if (run.isStopped()) return;
await run.acceptMappedEvent(
completeEvent,
mapSessionEventToRuntimeEvent(completeEvent, invocation),
{ requireTerminalWrite: true },
);
if (run.isStopped()) return;
yield completeEvent;
} catch (error) {
await run.recordFailure(error);
throw error;
} finally {
await this.finalizeExecutionClaimRun(execution, run, () => run.finalize());
}
}
async *startChildTurn(
sessionId: string,
input: ChildAgentTurnInput,
suppliedExecution?: RuntimeExecutionClaim,
): AsyncIterable<SessionEvent> {
const execution = this.takeExecutionClaim(sessionId, suppliedExecution);
try {
yield* this.startChildTurnClaimed(sessionId, input, execution);
} finally {
this.releaseExecutionClaim(execution);
}
}
private async *startChildTurnClaimed(
sessionId: string,
input: ChildAgentTurnInput,
execution: PendingExecutionClaim,
): AsyncIterable<SessionEvent> {
const parentHeader = await this.deps.store.readHeader(sessionId);
const definition = requireResolvedAgentDefinition(input.spec.id);
const availableChildTools = this.deps.childTools ?? [];
assertAgentDefinitionRunnable({
parentPermissionMode: parentHeader.permissionMode,
definition,
tools: availableChildTools,
});
const childTools = buildToolsForAgentDefinition(availableChildTools, definition);
const expertIdentity = parseExpertAgentId(definition.id);
const agentTeam: AgentTeamExecutionContext | undefined = expertIdentity
? {
role: 'member',
teamId: expertIdentity.teamId,
agentId: definition.id,
parentRunId: input.parentRunId,
}
: undefined;
const childHeader: SessionHeader = {
...parentHeader,
permissionMode: definition.permissionMode,
connectionLocked: true,
};
const userInput: UserMessageInput = {
turnId: input.turnId,
text: input.prompt,
parentRunId: input.parentRunId,
...(input.resumedFromRunId ? { resumedFromRunId: input.resumedFromRunId } : {}),
agentId: definition.id,
agentName: definition.name,
};
const activeKey = childActiveKey(sessionId, input.turnId);
const run = new AgentRun({
sessionId,
header: childHeader,
userInput,
store: this.deps.store,
runStore: this.deps.runStore,
runtimeEventStore: this.deps.runtimeEventStore,
...(runtimeToolBoundaryProtocol(this.deps, childHeader)
? { toolBoundaryProtocol: runtimeToolBoundaryProtocol(this.deps, childHeader) }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
effectiveOrchestration: resolveEffectiveOrchestration('default', undefined),
recordSessionMessages: false,
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
const active = await this.reserveChildRun(
activeKey,
targetSessionId,
nextHeader,
definition.systemPrompt,
childTools,
agentTeam,
activeRun,
execution,
);
this.reserveExecutionClaim(execution, active, activeRun);
return active;
},
unregisterRun: (active, activeRun) => this.unregisterChildRun(active, activeRun),
updateHeader: async (_targetSessionId, patch) => ({ ...childHeader, ...patch }),
updateStatus: async () => {},
appendTurnState: async () => {},
},
});
this.attachExecutionClaim(execution, run);
yield* this.runAgentTurn(sessionId, userInput, run, execution);
}
async *startChildRetry(
sessionId: string,
input: ChildAgentRetryInput,
suppliedExecution?: RuntimeExecutionClaim,
): AsyncIterable<SessionEvent> {
const execution = this.takeExecutionClaim(sessionId, suppliedExecution);
try {
yield* this.startChildRetryClaimed(sessionId, input, execution);
} finally {
this.releaseExecutionClaim(execution);
}
}
private async *startChildRetryClaimed(
sessionId: string,
input: ChildAgentRetryInput,
execution: PendingExecutionClaim,
): AsyncIterable<SessionEvent> {
const { continuation } = input;
if (continuation.sessionId !== sessionId) {
throw new Error('Child retry continuation belongs to a different session');
}
const parentHeader = await this.deps.store.readHeader(sessionId);
const linkedSnapshot = input.linkedSession ? parentHeader.subagentRuntime : undefined;
if (
input.linkedSession &&
(parentHeader.subagentParent?.kind !== 'subagent' ||
!linkedSnapshot ||
linkedSnapshot.agentId !== input.spec.id)
) {
throw new Error('Linked child retry is missing its durable runtime snapshot');
}
const definition = linkedSnapshot
? {
id: linkedSnapshot.agentId,
name: linkedSnapshot.agentName,
systemPrompt: linkedSnapshot.systemPrompt,
permissionMode: parentHeader.permissionMode,
tools: linkedSnapshot.toolNames,
categoryPolicy: linkedSnapshot.categoryPolicy,
}
: requireResolvedAgentDefinition(input.spec.id);
const availableChildTools = this.deps.childTools ?? [];
if (!linkedSnapshot) {
assertAgentDefinitionRunnable({
parentPermissionMode: parentHeader.permissionMode,
definition: requireResolvedAgentDefinition(input.spec.id),
tools: availableChildTools,
});
}
const childTools = buildToolsForAgentDefinition(availableChildTools, definition);
if (linkedSnapshot && childTools.length !== linkedSnapshot.toolNames.length) {
throw new Error('Linked child retry durable runtime tool snapshot is unavailable');
}
const expertIdentity = linkedSnapshot ? undefined : parseExpertAgentId(definition.id);
const agentTeam: AgentTeamExecutionContext | undefined = expertIdentity
? {
role: 'member',
teamId: expertIdentity.teamId,
agentId: definition.id,
parentRunId: input.parentRunId,
}
: undefined;
const childHeader: SessionHeader = linkedSnapshot
? parentHeader
: {
...parentHeader,
permissionMode: definition.permissionMode,
connectionLocked: true,
};
const userInput: UserMessageInput = {
turnId: continuation.turnId,
text: '',
...(!linkedSnapshot ? { parentRunId: input.parentRunId } : {}),
retriedFromRunId: continuation.sourceRunId,
agentId: definition.id,
agentName: definition.name,
};
const activeKey = childActiveKey(sessionId, continuation.turnId);
const run = new AgentRun({
sessionId,
header: childHeader,
userInput,
runId: continuation.runId,
invocationId: continuation.invocationId,
store: this.deps.store,
runStore: this.deps.runStore,
runtimeEventStore: this.deps.runtimeEventStore,
...(runtimeToolBoundaryProtocol(this.deps, childHeader)
? { toolBoundaryProtocol: runtimeToolBoundaryProtocol(this.deps, childHeader) }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration: resolveEffectiveOrchestration('default', undefined),
recordSessionMessages: false,
hooks: {
reserveRun: async (targetSessionId, nextHeader, activeRun) => {
const active = linkedSnapshot
? await this.reserveParentRun(targetSessionId, nextHeader, activeRun, execution)
: await this.reserveChildRun(
activeKey,
targetSessionId,
nextHeader,
definition.systemPrompt,
childTools,
agentTeam,
activeRun,
execution,
);
this.reserveExecutionClaim(execution, active, activeRun);
return active;
},
unregisterRun: (active, activeRun) =>
linkedSnapshot
? this.unregisterParentRun(active, activeRun)
: this.unregisterChildRun(active, activeRun),
updateHeader: (targetSessionId, patch) =>
linkedSnapshot
? this.updateHeader(targetSessionId, patch)
: Promise.resolve({ ...childHeader, ...patch }),
updateStatus: (targetSessionId, status, blockedReason, ts) =>
linkedSnapshot
? this.updateStatus(targetSessionId, status, blockedReason, ts)
: Promise.resolve(),
appendTurnState: (targetSessionId, turnId, status, lineage, options) =>
linkedSnapshot
? this.appendTurnState(targetSessionId, turnId, status, lineage, options)
: Promise.resolve(),
},
});
this.attachExecutionClaim(execution, run);
// A provider retry replays the source ledger without recording a second
// user prompt and without turning the child into a session continuation.
yield* this.runAgentContinuation(
continuation,
run,
execution,
false,
input.linkedSession === true,
input.onRunStarted,
);
}
private async *runAgentTurn(
sessionId: string,
input: UserMessageInput,
run: AgentRun,
execution: PendingExecutionClaim,
steering = false,
onRunStarted?: (runId: string, initialHeader: SessionHeader) => void | Promise<void>,
initialHeader?: SessionHeader,
): AsyncIterable<SessionEvent> {
const sessionEvents = new AsyncEventQueue<SessionEvent>();
const { abortController, release: releaseExecutionAbort } =
this.inheritExecutionAbort(execution);
let flowDone = false;
const owners = this.createRunOwnerScope(run, execution);
let begin: AgentRunBeginResult;
try {
if (steering) {
owners.bindMessage(this.deps.messageAuthority, {
sessionId,
turnId: run.turnId,
runId: run.runId,
});
}
begin = await this.runBackendActivation(async () => {
const started = await run.begin();
await owners.bindInteraction(this.deps.interactionAuthority, {
sessionId,
turnId: run.turnId,
runId: run.runId,
});
return started;
});
if (onRunStarted && initialHeader) await onRunStarted(run.runId, initialHeader);
} catch (error) {
releaseExecutionAbort();
await this.finalizeFailedRunStart(owners, run, execution, error);
return;
}
const interactionRun = owners.interactionRun;
const messageOwner = owners.messageOwner;
// Steering is a top-level-turn affordance only; child agent turns run
// without a queue. Hosted ownership is bound before begin so a pre-start
// cancellation can release the exact admitted owner. The pull hook still
// re-checks this run's turnId so stale or overlapping runs cannot drain
// messages queued for the current owner.
let pullSteering: (() => readonly SteeringLease[]) | undefined;
let ackSteering: ((leaseIds: readonly string[]) => void) | undefined;
let nackSteering: ((leaseIds: readonly string[]) => void) | undefined;
if (messageOwner) {
pullSteering = () => messageOwner?.pull() ?? [];
ackSteering = (leaseIds) => messageOwner?.ack(leaseIds);
nackSteering = (leaseIds) => messageOwner?.nack(leaseIds);
} else if (steering) {
const state = this.ensureSteering(sessionId);
state.sink = (event) => {
void sessionEvents.push(event).catch(() => {});
};
state.activeTurnId = run.turnId;
// Lease, don't consume: pulled messages move to in-flight and only an
// ack (durable + injected) removes them; a nack or a retract/clear/
// release reclaims them, so an abort window can never drop text.
pullSteering = () => {
const current = this.steeringBySession.get(sessionId);
if (!current || current.activeTurnId !== run.turnId) return [];
if (current.steering.length === 0) return [];
const leased = current.steering.splice(0);
current.inFlight.push(
...leased.map((message) => ({ ...message, issuingTurnId: run.turnId })),
);
return leased.map((message) => ({ ...message }));
};
// Settlement is keyed by lease id + issuing turn, NOT by current
// ownership: an overlapping turn that takes the owner slot must not
// invalidate the issuer's ack (the message was delivered to ITS
// provider) or intercept its nack. A late settle for a reclaimed lease
// finds no match and is a no-op.
ackSteering = (leaseIds) => {
const current = this.steeringBySession.get(sessionId);
if (!current) return;
const ids = new Set(leaseIds);
const before = current.inFlight.length;
current.inFlight = current.inFlight.filter(
(message) => !(ids.has(message.id) && message.issuingTurnId === run.turnId),
);
if (current.inFlight.length !== before) this.emitQueueUpdate(sessionId, current);
};
nackSteering = (leaseIds) => {
const current = this.steeringBySession.get(sessionId);
if (!current) return;
const ids = new Set(leaseIds);
const returned = current.inFlight.filter(
(message) => ids.has(message.id) && message.issuingTurnId === run.turnId,
);
if (returned.length === 0) return;
current.inFlight = current.inFlight.filter(
(message) => !(ids.has(message.id) && message.issuingTurnId === run.turnId),
);
if (current.activeTurnId === run.turnId) {
// Back to the FRONT of the queue: a re-pull at the next step
// boundary preserves the user's original ordering.
current.steering = [
...returned.map(({ id, messageId, content }) => ({ id, messageId, content })),
...current.steering,
];
} else {
// The issuer no longer owns the queue (an overlapping turn took
// over and possibly released): it will never pull again, so the
// steering queue would strand the text ownerless. The followup
// queue is its only safe home — the same direction a release-time
// fold takes.
current.followup = [
...returned.map((message) => message.content.text),
...current.followup,
];
}
this.emitQueueUpdate(sessionId, current);
};
}
const aiSdkFlow = new AiSdkFlow({
backend: begin.backend,
stopBackend: this.stopBackendFor(begin.backend),
beforeDispatch: () => this.assertRunCanDispatch(run, begin.backend),
...(interactionRun ? { hostedInteraction: interactionRun } : {}),
drainAfterTerminal: true,
onSessionEvent: async (sessionEvent, runtimeEvent) => {
this.assertInteractionPublication(interactionRun, sessionEvent);
await run.acceptMappedEvent(sessionEvent, runtimeEvent, {
requireTerminalWrite: Boolean(this.deps.runtimeEventStore),
allowInteractionResume: await interactionResumeAllowed(interactionRun, sessionEvent),
});
await sessionEvents.push(sessionEvent);
},
onError: async (error) => {
if (!isAsyncEventQueueClosed(error)) {
await run.recordFailure(error);
sessionEvents.fail(error);
}
},
onFinally: async () => {
flowDone = true;
try {
await owners.finalize();
// Release Runtime access BEFORE the event stream closes. Embedded
// queues still emit their final steering → followup projection here;
// a hosted owner is only sealed, then the Host performs that handoff
// under its Session admission gate. The outer finally remains an
// idempotent backstop for paths that never reach this hook.
if (messageOwner) owners.releaseMessage();
else if (steering) this.releaseSteeringTurn(sessionId, run.turnId);
sessionEvents.close();
} catch (error) {
sessionEvents.fail(error);
throw error;
}
},
});
const runner = new RuntimeRunner({
flow: aiSdkFlow,
providers: { newId: this.deps.newId, now: this.deps.now },
stopOnTerminal: false,
...(run.toolBoundaryProtocol ? { toolBoundaryProtocol: run.toolBoundaryProtocol } : {}),
});
if (run.isStopped()) abortController.abort();
const runnerResult = runner
.run({
sessionId,
invocationId: begin.initialRuntimeEvent.invocationId,
runId: run.runId,
turnId: run.turnId,
...(begin.backendInput.orchestration
? { orchestration: begin.backendInput.orchestration }
: {}),
text: input.text,
...(begin.backendInput.attachments ? { attachments: begin.backendInput.attachments } : {}),
...(begin.backendInput.quotes ? { quotes: begin.backendInput.quotes } : {}),
context: begin.backendInput.context,
...(begin.backendInput.runtimeContext !== undefined
? { runtimeContext: begin.backendInput.runtimeContext }
: {}),
initialRuntimeEvent: begin.initialRuntimeEvent,
source: this.deps.runtimeSource ?? 'desktop',
lineage: run.lineage,
...(pullSteering ? { pullSteering } : {}),
...(ackSteering ? { ackSteering } : {}),
...(nackSteering ? { nackSteering } : {}),
abortSignal: abortController.signal,
})
.then(
async (result) => {
if (!flowDone) {
try {
flowDone = true;
await owners.finalize();
owners.releaseMessage();
sessionEvents.close();
} catch (error) {
sessionEvents.fail(error);
throw error;
}
}
await this.deps.runtimeInvocationObserver?.(result);
return result;
},
(error) => {
sessionEvents.fail(error);
throw error;
},
);
try {
for await (const event of sessionEvents) {
yield event;
}
await runnerResult;
} finally {
try {
await this.cleanupRunExecution({
run,
flow: aiSdkFlow,
flowDone,
abortController,
sessionEvents,
runnerResult,
interactionRun,
finalizeRun: () => owners.finalize(),
releaseOwner: () => {
if (messageOwner) owners.releaseMessage();
else if (steering) this.releaseSteeringTurn(sessionId, run.turnId);
},
});
} finally {
releaseExecutionAbort();
}
}
}
private async *runAgentContinuation(
continuation: RuntimeContinuation,
run: AgentRun,
execution: PendingExecutionClaim,
persistContinuationSource = true,
bindHostedRoot = false,
onRunStarted?: () => void | Promise<void>,
): AsyncIterable<SessionEvent> {
const sessionEvents = new AsyncEventQueue<SessionEvent>();
const { abortController, release: releaseExecutionAbort } =
this.inheritExecutionAbort(execution);
let flowDone = false;
const owners = this.createRunOwnerScope(run, execution);
let begin: Awaited<ReturnType<AgentRun['beginContinuation']>>;
try {
if (bindHostedRoot) {
owners.bindMessage(this.deps.messageAuthority, {
sessionId: continuation.sessionId,
turnId: continuation.turnId,
runId: continuation.runId,
});
}
begin = await this.runBackendActivation(async () => {
const started = persistContinuationSource
? await run.beginContinuation(continuation)
: await run.beginOperation();
await owners.bindInteraction(this.deps.interactionAuthority, {
sessionId: continuation.sessionId,
turnId: run.turnId,
runId: run.runId,
});
return started;
});
await onRunStarted?.();
} catch (error) {
releaseExecutionAbort();
await this.finalizeFailedRunStart(owners, run, execution, error);
return;
}
const interactionRun = owners.interactionRun;
const aiSdkFlow = new AiSdkFlow({
backend: begin.backend,
stopBackend: this.stopBackendFor(begin.backend),
beforeDispatch: () => this.assertRunCanDispatch(run, begin.backend),
...(interactionRun ? { hostedInteraction: interactionRun } : {}),
drainAfterTerminal: true,
onSessionEvent: async (sessionEvent, runtimeEvent) => {
this.assertInteractionPublication(interactionRun, sessionEvent);
await run.acceptMappedEvent(sessionEvent, runtimeEvent, {
requireTerminalWrite: true,
allowInteractionResume: await interactionResumeAllowed(interactionRun, sessionEvent),
});
await sessionEvents.push(sessionEvent);
},
onError: async (error) => {
if (!isAsyncEventQueueClosed(error)) {
await run.recordFailure(error);
sessionEvents.fail(error);
}
},
onFinally: async () => {
flowDone = true;
try {
await owners.finalize();
owners.releaseMessage();
sessionEvents.close();
} catch (error) {
sessionEvents.fail(error);
throw error;
}
},
});
const runner = new RuntimeRunner({
flow: aiSdkFlow,
providers: { newId: this.deps.newId, now: this.deps.now },
stopOnTerminal: false,
...(run.toolBoundaryProtocol ? { toolBoundaryProtocol: run.toolBoundaryProtocol } : {}),
commitContinuationStart: async (event) => {
await run.recordRuntimeEvents([event], { requireTerminalWrite: true });
if (persistContinuationSource) {
await this.deps.continuationFailpoint?.('after_continuation_start_committed');
}
},
});
if (run.isStopped()) abortController.abort();
let runnerFailure: unknown;
const runnerResult = runner
.resume(continuation, {
source: this.deps.runtimeSource ?? 'desktop',
orchestration: run.effectiveOrchestration,
abortSignal: abortController.signal,
})
.then(
async (result) => {
if (!flowDone) {
try {
flowDone = true;
await owners.finalize();
owners.releaseMessage();
sessionEvents.close();
} catch (error) {
runnerFailure = error;
sessionEvents.fail(error);
throw error;
}
}
await this.deps.runtimeInvocationObserver?.(result);
return result;
},
(error) => {
runnerFailure = error;
sessionEvents.fail(error);
throw error;
},
);
try {
for await (const event of sessionEvents) {
yield event;
}
await runnerResult;
} finally {
try {
await this.cleanupRunExecution({
run,
flow: aiSdkFlow,
flowDone,
abortController,
sessionEvents,
runnerResult,
interactionRun,
...(runnerFailure !== undefined ? { runnerFailure } : {}),
finalizeRun: () => owners.finalize(),
releaseOwner: () => owners.releaseMessage(),
});
} finally {
releaseExecutionAbort();
}
}
}
private inheritExecutionAbort(execution: PendingExecutionClaim): {
abortController: AbortController;
release(): void;
} {
const abortController = new AbortController();
const onAbort = (): void => abortController.abort(execution.abortController.signal.reason);
execution.abortController.signal.addEventListener('abort', onAbort, { once: true });
if (execution.abortController.signal.aborted) onAbort();
return {
abortController,
release: () => execution.abortController.signal.removeEventListener('abort', onAbort),
};
}
private async finalizeFailedRunStart(
owners: RuntimeRunOwnerScope,
run: AgentRun,
execution: PendingExecutionClaim,
error: unknown,
): Promise<void> {
try {
await owners.failStart(error);
} catch (failure) {
if (run.isStopped() && isExecutionCancellation(failure, execution.cancellation)) return;
throw failure;
}
}
private createRunOwnerScope(
run: AgentRun,
execution: PendingExecutionClaim,
): RuntimeRunOwnerScope {
return new RuntimeRunOwnerScope(run, {
registerInteraction: (binding) => this.registerInteractionRun(run, binding),
releaseInteraction: (binding) => this.releaseInteractionRun(run, binding),
settleReservedExecution: (outcome) =>
this.settleReservedExecutionClaim(execution, run, outcome),
finalizeExecution: (operation) => this.finalizeExecutionClaimRun(execution, run, operation),
});
}
private async cleanupRunExecution(input: {
run: AgentRun;
flow: AiSdkFlow;
flowDone: boolean;
abortController: AbortController;
sessionEvents: AsyncEventQueue<SessionEvent>;
runnerResult: Promise<InvocationResult>;
interactionRun: RuntimeInteractionRunBinding | undefined;
runnerFailure?: unknown;
finalizeRun: () => Promise<void>;
releaseOwner: () => void;
}): Promise<void> {
const failures = new FailureCollector();
if (!input.flowDone) {
input.run.stop('stop_button');
let interactionClose: Promise<void> | undefined;
try {
interactionClose = input.interactionRun?.close(interactionClosureReason(input.run));
} catch (error) {
failures.add(error);
}
const backendStop = input.flow.stop('user_stop');
input.abortController.abort();
input.sessionEvents.close();
await Promise.all([
failures.capture(() => interactionClose),
failures.capture(() => backendStop),
]);
if (input.runnerFailure !== undefined) {
await failures.capture(() => input.run.recordFailure(input.runnerFailure));
}
}
await input.runnerResult.catch(() => undefined);
await failures.capture(input.finalizeRun);
await failures.capture(input.releaseOwner);
const message = `Run cleanup failed for ${input.run.runId}`;
try {
failures.throwIfAny(message);
} catch (error) {
if (containsRuntimeOwnerCleanupFailure(error)) {
throw runtimeOwnerCleanupFailure(message, error);
}
throw error;
}
}
private registerInteractionRun(run: AgentRun, binding: RuntimeInteractionRunBinding): void {
if (
binding.sessionId !== run.sessionId ||
binding.turnId !== run.turnId ||
binding.runId !== run.runId ||
this.interactionRuns.has(run)
) {
throw new RuntimeInteractionFailStopError(
`RuntimeKernel could not register exact Interaction Run ${run.runId}`,
new Error('Interaction Run identity or ownership mismatch'),
);
}
this.interactionRuns.set(run, binding);
}
private releaseInteractionRun(run: AgentRun, binding: RuntimeInteractionRunBinding): void {
const current = this.interactionRuns.get(run);
if (current && current !== binding) {
throw new RuntimeInteractionFailStopError(
`RuntimeKernel could not release exact Interaction Run ${run.runId}`,
new Error('Interaction Run owner changed before release'),
);
}
binding.release();
if (current === binding) this.interactionRuns.delete(run);
}
private assertInteractionPublication(
binding: RuntimeInteractionRunBinding | undefined,
event: SessionEvent,
): void {
if (
binding &&
(event.type === 'permission_request' || event.type === 'user_question_request')
) {
binding.assertPendingAdmission(event);
}
}
private compactInvocationContext(input: {
sessionId: string;
runId: string;
turnId: string;
startedAt: number;
}): InvocationContext {
const request = {
sessionId: input.sessionId,
invocationId: input.runId,
runId: input.runId,
turnId: input.turnId,
text: '',
context: [],
source: this.deps.runtimeSource ?? 'desktop',
} satisfies InvocationContext['request'];
return {
sessionId: input.sessionId,
invocationId: input.runId,
runId: input.runId,
turnId: input.turnId,
source: this.deps.runtimeSource ?? 'desktop',
startedAt: input.startedAt,
request,
newId: this.deps.newId,
now: this.deps.now,
};
}
stopSession(sessionId: string, input: StopSessionInput = {}): Promise<void> {
const existing = this.stopAttempts.get(sessionId);
if (existing) return existing;
const intent: SessionStopIntent = { input, claims: new Set() };
this.stopIntents.set(sessionId, intent);
const executions = [...(this.executionClaims.get(sessionId) ?? [])];
for (const execution of executions) {
execution.stopIntent = intent;
intent.claims.add(execution);
}
for (const execution of executions) execution.run?.stop(input.source);
for (const execution of executions) {
execution.abortController.abort(execution.cancellation);
}
const attempt = this.stopSessionAttempt(sessionId, intent).finally(() => {
if (this.stopAttempts.get(sessionId) === attempt) {
this.stopAttempts.delete(sessionId);
}
if (this.stopIntents.get(sessionId) === intent) {
this.stopIntents.delete(sessionId);
}
});
this.stopAttempts.set(sessionId, attempt);
return attempt;
}
private async stopSessionAttempt(sessionId: string, intent: SessionStopIntent): Promise<void> {
// Interrupt clears both queues before the abort lands; the emitted empty
// snapshot lets the UI collapse its pending bar, and callers refill their
// editor from the mirror captured before the clear.
this.clearSteering(sessionId);
const failures: unknown[] = [];
let operation = this.stopOperations.get(sessionId);
try {
for (const active of this.backendGenerationsFor(sessionId)) {
for (const run of active.activeRuns.values()) {
operation = this.claimRunForStop(sessionId, intent.input, active, run) ?? operation;
}
}
} catch (error) {
failures.push(error);
}
const claimResults = await Promise.allSettled(
[...intent.claims].map((execution) => execution.settled),
);
for (const result of claimResults) {
if (result.status === 'rejected') failures.push(result.reason);
}
if (failures.length > 0) {
const message = `Session ${sessionId} stop ownership failed`;
const error = failures.length === 1 ? failures[0] : new AggregateError(failures, message);
if (containsRuntimeOwnerCleanupFailure(error)) {
throw runtimeOwnerCleanupFailure(message, error);
}
throw error;
}
operation = this.stopOperations.get(sessionId) ?? operation;
if (operation) {
await this.enqueueStopOperation(sessionId, operation, intent.input, true);
}
}
private claimRunForStop(
sessionId: string,
input: StopSessionInput,
active: BackendGeneration,
run: AgentRun,
): StopOperation | undefined {
run.stop(input.source);
if (!run.hasPendingStop()) return this.stopOperations.get(sessionId);
const existingOperation = this.stopOperations.get(sessionId);
const operation = existingOperation ?? this.buildStopOperation(input);
const existingTarget = operation.targets.get(active.generation);
const target =
existingTarget ??
({
active,
generation: active.generation,
runs: new Map(),
delivery: { kind: 'pending' },
} satisfies StopTarget);
const needsRun = !target.runs.has(run.runId);
const projection =
needsRun && run.isSessionInline() && !operation.turnProjections.has(run.runId)
? {
id: this.deps.newId(),
turnId: run.turnId,
lineage: run.lineage,
projected: false,
}
: undefined;
if (!existingOperation) this.stopOperations.set(sessionId, operation);
if (!existingTarget) {
operation.targets.set(active.generation, target);
}
if (needsRun) {
target.runs.set(run.runId, {
run,
runId: run.runId,
turnId: run.turnId,
lineage: run.lineage,
sessionInline: run.isSessionInline(),
stopCompleted: false,
});
if (projection) operation.turnProjections.set(run.runId, projection);
}
return operation;
}
private buildStopOperation(input: StopSessionInput): StopOperation {
const abortSource = normalizeStopSessionSource(input.source);
const ts = this.deps.now();
return {
abortSource,
ts,
statusProjected: false,
turnProjections: new Map(),
abortNote: {
type: 'system_note',
id: this.deps.newId(),
ts,
kind: 'abort',
...(abortSource ? { data: { source: abortSource } } : {}),
},
abortNoteProjected: false,
targets: new Map(),
queue: Promise.resolve(),
};
}
private enqueueStopOperation(
sessionId: string,
operation: StopOperation,
input: StopSessionInput,
deliverPending: boolean,
): Promise<void> {
const attempt = operation.queue
.catch(() => undefined)
.then(() => this.advanceStopOperation(sessionId, operation, input, deliverPending));
operation.queue = attempt.catch(() => undefined);
return attempt;
}
private async advanceStopOperation(
sessionId: string,
operation: StopOperation,
input: StopSessionInput,
deliverPending: boolean,
): Promise<void> {
const stoppedRuns = new Map(
[...operation.targets.values()].flatMap((target) => [...target.runs.entries()]),
);
const failures = new FailureCollector();
let newlyFailed = false;
const interactionClosures = [...stoppedRuns.values()].map(async (target) => {
const run = target.run;
if (!run) return;
try {
await this.interactionRuns.get(run)?.close('turn_stopped');
} catch (error) {
newlyFailed = true;
failures.add(
interactionFailStop(
`Could not durably close stopped Runs for session ${sessionId}`,
error,
),
);
}
});
const undelivered = deliverPending
? [...operation.targets.values()].filter((target) => target.delivery.kind === 'pending')
: [];
const backendStops = undelivered.map(async (target) => {
try {
const active = target.active;
if (!active) {
throw new Error(`Backend generation ${target.generation} lost its pending stop owner`);
}
if (active.phase === 'active') active.phase = 'stopping';
await active.stopBackend('user_stop', input.mode);
target.delivery = { kind: 'delivered' };
} catch (error) {
newlyFailed = true;
target.delivery = { kind: 'failed', error };
failures.add(error);
}
});
await Promise.all([...interactionClosures, ...backendStops]);
if (newlyFailed) {
const message = `Stop cleanup failed for session ${sessionId}`;
try {
failures.throwIfAny(message);
} catch (error) {
if (containsRuntimeOwnerCleanupFailure(error)) {
throw runtimeOwnerCleanupFailure(message, error);
}
throw error;
}
}
if (!operation.statusProjected) {
await this.updateStatus(sessionId, 'aborted', undefined, operation.ts);
operation.statusProjected = true;
}
for (const projection of operation.turnProjections.values()) {
if (projection.projected) continue;
projection.message ??= buildTurnStateMessage({
id: projection.id,
turnId: projection.turnId,
ts: operation.ts,
status: 'aborted',
lineage: projection.lineage,
...(operation.abortSource ? { abortSource: operation.abortSource } : {}),
partialOutputRetained: await this.turnHasRetainedOutput(sessionId, projection.turnId),
});
await this.appendStopProjection(sessionId, projection.message);
projection.projected = true;
}
if (!operation.abortNoteProjected) {
await this.appendStopProjection(sessionId, operation.abortNote);
operation.abortNoteProjected = true;
}
for (const target of stoppedRuns.values()) {
target.run?.completeStop();
target.stopCompleted = true;
}
const completed =
operation.statusProjected &&
operation.abortNoteProjected &&
[...operation.turnProjections.values()].every((projection) => projection.projected) &&
[...operation.targets.values()].every(
(target) =>
target.delivery.kind !== 'pending' &&
[...target.runs.values()].every((run) => run.stopCompleted),
);
if (completed && this.stopOperations.get(sessionId) === operation) {
this.stopOperations.delete(sessionId);
}
await Promise.all(
[...operation.targets.values()].map((target) =>
target.active ? this.settleBackendGenerationAfterRunExit(target.active) : Promise.resolve(),
),
);
for (const target of operation.targets.values()) {
if (target.delivery.kind === 'failed') failures.add(target.delivery.error);
}
failures.throwIfAny(`Stop cleanup failed for session ${sessionId}`);
}
private async appendStopProjection(sessionId: string, message: StoredMessage): Promise<void> {
const existing = (await this.deps.store.readMessages(sessionId)).find(
(candidate) => candidate.id === message.id,
);
if (existing) {
if (!isDeepStrictEqual(existing, message)) {
throw new Error(`stop projection ${message.id} conflicts with an existing message`);
}
return;
}
await this.deps.store.appendMessage(sessionId, message);
}
async respondToPermission(sessionId: string, response: PermissionResponse): Promise<void> {
const generations = this.backendGenerationsFor(sessionId);
await Promise.all(generations.map((active) => active.backend.respondToPermission(response)));
}
async respondToUserQuestion(sessionId: string, response: UserQuestionResponse): Promise<void> {
if (this.deps.interactionAuthority) {
throw new RuntimeInteractionInvariantError(
'Hosted user questions must settle through their captured continuation',
);
}
const generations = this.backendGenerationsFor(sessionId);
await Promise.all(
generations.map((active) => active.backend.respondToUserQuestion?.(response)),
);
}
// --------------------------------------------------------------------------
// Steering / followup queues (authoritative source of truth)
// --------------------------------------------------------------------------
steer(sessionId: string, text: string): QueueEnqueueOutcome {
this.assertEmbeddedMessageQueue('steer');
// Steering's delivery contract is anchored to the runtime event ledger
// (fail-closed persist + durable-consume ack). Without a RuntimeEventStore
// that anchor does not exist — same condition as requireTerminalWrite —
// so fall back to a fresh turn, whose user message the SessionStore
// persists with the ordinary turn-open guarantee.
if (!this.deps.runtimeEventStore) return { kind: 'fallback' };
// Double responsibility (codex): with no live steering owner to inject
// into — the turn just ended, begin() failed, or only child/compact runs
// are active (they never consume this queue) — tell the caller to open a
// fresh turn instead so the message is never dropped.
const state = this.liveSteeringState(sessionId);
if (!state) return { kind: 'fallback' };
const messageId = this.deps.newId();
state.steering.push({ id: messageId, messageId, content: { text } });
this.emitQueueUpdate(sessionId, state);
return { kind: 'queued' };
}
queueMessage(sessionId: string, text: string): QueueEnqueueOutcome {
this.assertEmbeddedMessageQueue('queueMessage');
const state = this.liveSteeringState(sessionId);
if (!state) return { kind: 'fallback' };
state.followup.push(text);
this.emitQueueUpdate(sessionId, state);
return { kind: 'queued' };
}
drainFollowup(sessionId: string): string | null {
this.assertEmbeddedMessageQueue('drainFollowup');
const state = this.steeringBySession.get(sessionId);
if (!state || state.followup.length === 0) return null;
const drained = state.followup.splice(0);
this.emitQueueUpdate(sessionId, state);
return drained.join('\n\n');
}
retractQueue(sessionId: string): string {
this.assertEmbeddedMessageQueue('retractQueue');
const state = this.steeringBySession.get(sessionId);
if (!state) return '';
// Retract reclaims QUEUED messages only. pull() is the single atomic
// commit point of delivery: an in-flight lease is already committed to
// the running turn — its durable append may land at any moment, so
// handing its text back to the user here would refill AND execute the
// same directive. An in-flight lease settles only by the persistence
// fact (ack when the ledger owns it, nack back to a queue otherwise).
const all = [...state.steering.map((message) => message.content.text), ...state.followup];
state.steering = [];
state.followup = [];
this.emitQueueUpdate(sessionId, state);
return all.join('\n\n');
}
private ensureSteering(sessionId: string): SessionSteeringState {
const existing = this.steeringBySession.get(sessionId);
if (existing) return existing;
const created: SessionSteeringState = { steering: [], inFlight: [], followup: [] };
this.steeringBySession.set(sessionId, created);
return created;
}
private assertEmbeddedMessageQueue(operation: string): void {
if (this.deps.messageAuthority) {
throw new RuntimeMessageAuthorityInvariantError(
`Hosted Runtime cannot ${operation}; the Runtime Host owns message admission and queues`,
);
}
}
/**
* The session's steering state only while a steering-capable top-level run
* owns it (sink registered after begin() succeeded and not yet released).
* Child agent and compact runs never establish ownership, so their activity
* alone yields undefined — enqueue must fall back rather than strand text.
*/
private liveSteeringState(sessionId: string): SessionSteeringState | undefined {
const state = this.steeringBySession.get(sessionId);
return state?.sink ? state : undefined;
}
private emitQueueUpdate(sessionId: string, state: SessionSteeringState): void {
state.sink?.({
type: 'queue_update',
id: this.deps.newId(),
turnId: state.activeTurnId ?? '',
ts: this.deps.now(),
steering: [
...state.inFlight.map((message) => message.content.text),
...state.steering.map((message) => message.content.text),
],
followup: [...state.followup],
});
}
private clearSteering(sessionId: string): void {
const state = this.steeringBySession.get(sessionId);
if (!state) return;
// Same commit-point rule as retractQueue: only QUEUED messages are
// clearable. An in-flight lease is already committed to the running
// turn's delivery and settles only by the persistence fact.
if (state.steering.length === 0 && state.followup.length === 0) return;
state.steering = [];
state.followup = [];
this.emitQueueUpdate(sessionId, state);
}
private releaseSteeringTurn(sessionId: string, turnId: string): void {
const state = this.steeringBySession.get(sessionId);
if (!state) return;
// A release folds only the leases THIS turn issued; an overlapping turn's
// in-flight lease stays for its issuer to settle (acked = delivered, so
// folding it into followup would redeliver an already-executed message).
const own = state.inFlight.filter((message) => message.issuingTurnId === turnId);
if (state.activeTurnId !== turnId) {
// Not (or no longer) the owner. The issuer's backend settles every
// lease before its turn ends, so `own` is normally empty; this is a
// backstop that keeps a never-settled lease from stranding invisibly.
if (own.length === 0) return;
state.inFlight = state.inFlight.filter((message) => message.issuingTurnId !== turnId);
state.followup = [...own.map((message) => message.content.text), ...state.followup];
this.emitQueueUpdate(sessionId, state);
return;
}
// Stranded steering (arrived after the final step boundary, so no step is
// left to consume it) becomes the head of the followup queue instead of
// vanishing — the next turn opens with it first (grok-build safety). The
// migration is a queue change, so emit the final snapshot BEFORE the sink
// is cleared; otherwise observers stay on the stale pre-fold snapshot.
if (state.steering.length > 0 || own.length > 0) {
state.followup = [
...own.map((message) => message.content.text),
...state.steering.map((message) => message.content.text),
...state.followup,
];
state.inFlight = state.inFlight.filter((message) => message.issuingTurnId !== turnId);
state.steering = [];
this.emitQueueUpdate(sessionId, state);
}
state.sink = undefined;
state.activeTurnId = undefined;
}
hasActiveRuns(sessionId: string): boolean {
return this.backendGenerationsFor(sessionId).some((active) => active.activeRuns.size > 0);
}
hasActiveRun(sessionId: string, runId: string, turnId?: string): boolean {
return this.backendGenerationsFor(sessionId).some((active) => {
const run = active.activeRuns.get(runId);
return run !== undefined && (turnId === undefined || run.turnId === turnId);
});
}
updateCachedHeader(sessionId: string, header: SessionHeader): void {
const active = this.active.get(sessionId);
if (active) active.cachedHeader = header;
}
async invalidateBackend(sessionId: string): Promise<void> {
this.ensureBackendInvalidation(sessionId);
await this.flushBackendInvalidation(sessionId);
}
async invalidateCachedBackends(): Promise<void> {
const sessionIds = new Set(
[...this.backendGenerations.values()].map((generation) => generation.sessionId),
);
for (const sessionId of this.backendInvalidations.keys()) sessionIds.add(sessionId);
await Promise.all(
[...sessionIds].map(async (sessionId) => {
const failedGeneration = this.backendGenerationsFor(sessionId).find(
(generation) => generation.phase === 'failed',
);
if (failedGeneration) {
const retained = await failedGeneration.disposal;
if (!retained?.ok) throw retained?.error ?? failedGeneration.disposalFailure;
}
const invalidation = this.ensureBackendInvalidation(sessionId);
await this.flushBackendInvalidation(sessionId);
const outcome = await invalidation.outcome;
if (!outcome.ok) throw outcome.error;
}),
);
}
async disposeBackend(sessionId: string): Promise<void> {
const invalidation = this.ensureBackendInvalidation(sessionId);
await this.startBackendDisposal(sessionId, invalidation);
const outcome = await invalidation.outcome;
if (!outcome.ok) throw outcome.error;
}
private async disposeBackendNow(sessionId: string): Promise<BackendDisposalOutcome> {
const generations = this.backendGenerationsFor(sessionId);
this.steeringBySession.delete(sessionId);
this.historyCompactCheckpoints.delete(sessionId);
this.historyCompactCheckpointLoads.delete(sessionId);
let disposalError: unknown;
for (const active of generations) {
const outcome = await this.quarantineBackendGeneration(active);
if (!outcome.ok) disposalError ??= outcome.error;
}
return disposalError === undefined ? { ok: true } : { ok: false, error: disposalError };
}
private backendGenerationsFor(sessionId: string): BackendGeneration[] {
return [...this.backendGenerations.values()].filter(
(generation) => generation.sessionId === sessionId && generation.phase !== 'terminated',
);
}
private stopBackendFor(backend: AgentBackend): AgentBackend['stop'] {
for (const active of this.backendGenerations.values()) {
if (active.backend === backend) return active.stopBackend;
}
throw new Error(`Backend stop owner is unavailable for session ${backend.sessionId}`);
}
private createBackendStopOwner(active: BackendGeneration): AgentBackend['stop'] {
return (reason, mode) => {
if (active.stopState.kind === 'failed') {
return Promise.reject(active.stopState.error);
}
if (active.stopState.kind === 'pending') return active.stopState.task;
if (active.phase === 'active') active.phase = 'stopping';
const attempt = Promise.resolve()
.then(() => active.backend.stop(reason, mode))
.catch(async (stopError: unknown) => {
const disposal = await this.quarantineBackendGeneration(active);
const failure = disposal.ok
? stopError
: new AggregateError(
[stopError, disposal.error],
`Backend generation ${active.generation} stop and disposal failed`,
);
active.stopState = { kind: 'failed', error: failure };
throw failure;
});
active.stopState = { kind: 'pending', task: attempt };
const clear = (): void => {
if (active.stopState.kind === 'pending' && active.stopState.task === attempt) {
active.stopState = { kind: 'idle' };
}
};
void attempt.then(clear, clear);
return attempt;
};
}
private quarantineBackendGeneration(active: BackendGeneration): Promise<BackendDisposalOutcome> {
if (active.phase === 'terminated') return Promise.resolve({ ok: true });
if (active.phase === 'failed') {
return active.disposal ?? Promise.resolve({ ok: false, error: active.disposalFailure });
}
active.phase = 'disposing';
active.disposal ??= this.disposeBackendGeneration(active);
return active.disposal;
}
private disposeBackendGeneration(active: BackendGeneration): Promise<BackendDisposalOutcome> {
return (async () => {
let result: BackendDisposalOutcome;
try {
await active.backend.dispose();
result = { ok: true };
} catch (error) {
result = { ok: false, error };
}
if (result.ok) {
if (active.activeRuns.size === 0 && !this.stopOperationReferences(active)) {
this.terminateBackendGeneration(active);
}
} else {
active.disposalFailure = new Error(
`Backend generation ${active.generation} is permanently quarantined after disposal failed`,
{ cause: result.error },
);
active.phase = 'failed';
}
return result;
})();
}
private terminateBackendGeneration(active: BackendGeneration): void {
if (
active.phase === 'failed' ||
active.activeRuns.size > 0 ||
this.stopOperationReferences(active)
) {
return;
}
active.phase = 'terminated';
this.detachBackendGeneration(active);
this.backendGenerations.delete(active.generation);
}
private detachBackendGeneration(active: BackendGeneration): void {
if (this.active.get(active.sessionId) === active) this.active.delete(active.sessionId);
for (const [key, child] of this.childActive.entries()) {
if (child === active) this.childActive.delete(key);
}
}
private stopOperationReferences(active: BackendGeneration): boolean {
return [...this.stopOperations.values()].some((operation) =>
[...operation.targets.values()].some((target) => target.active === active),
);
}
private loadHistoryCompactCheckpoint(
sessionId: string,
): Promise<HistoryCompactCheckpoint | undefined> {
if (this.historyCompactCheckpoints.has(sessionId)) {
return Promise.resolve(this.historyCompactCheckpoints.get(sessionId));
}
const existing = this.historyCompactCheckpointLoads.get(sessionId);
if (existing) return existing;
if (!this.deps.runStore) return Promise.resolve(undefined);
let guardedLoad: Promise<HistoryCompactCheckpoint | undefined>;
guardedLoad = loadLatestHistoryCompactCheckpointFromRunLedger(this.deps.runStore, sessionId)
.then((checkpoint) => {
if (checkpoint) this.scheduleHistoryCompactCleanup(sessionId, checkpoint);
if (
this.historyCompactCheckpointLoads.get(sessionId) === guardedLoad &&
!this.historyCompactCheckpoints.has(sessionId)
) {
this.historyCompactCheckpoints.set(sessionId, checkpoint);
}
return this.historyCompactCheckpoints.has(sessionId)
? this.historyCompactCheckpoints.get(sessionId)
: checkpoint;
})
.finally(() => {
if (this.historyCompactCheckpointLoads.get(sessionId) === guardedLoad) {
this.historyCompactCheckpointLoads.delete(sessionId);
}
});
this.historyCompactCheckpointLoads.set(sessionId, guardedLoad);
return guardedLoad;
}
private recordHistoryCompactCheckpoint(
sessionId: string,
checkpoint: HistoryCompactCheckpoint,
run: AgentRun | undefined,
): Promise<void> {
if (!run) return Promise.reject(new Error('No active AgentRun for history compact checkpoint'));
const previous = this.historyCompactCheckpointWrites.get(sessionId) ?? Promise.resolve();
let tracked: Promise<void>;
tracked = previous
.catch(() => {})
.then(async () => {
const durableCheckpoint = await this.loadHistoryCompactCheckpoint(sessionId);
if (!canReplaceHistoryCompactCheckpoint(durableCheckpoint, checkpoint)) {
throw new Error('History compact checkpoint was superseded before persistence');
}
await run.recordHistoryCompactCheckpoint(checkpoint);
this.historyCompactCheckpoints.set(sessionId, checkpoint);
this.scheduleHistoryCompactCleanup(sessionId, checkpoint);
})
.finally(() => {
if (this.historyCompactCheckpointWrites.get(sessionId) === tracked) {
this.historyCompactCheckpointWrites.delete(sessionId);
}
});
this.historyCompactCheckpointWrites.set(sessionId, tracked);
return tracked;
}
private scheduleHistoryCompactCleanup(
sessionId: string,
checkpoint: HistoryCompactCheckpoint,
): void {
if (
!this.deps.cleanupHistoryCompactArtifacts ||
!this.deps.runStore ||
!this.deps.runtimeEventStore
)
return;
const previous = this.historyCompactCleanupWrites.get(sessionId) ?? Promise.resolve();
let tracked: Promise<void>;
tracked = previous
.catch(() => {})
.then(async () => {
const runs = (await this.deps.runStore!.listSessionRuns(sessionId)).filter(
isSessionInlineRun,
);
const runtimeEvents: RuntimeEvent[] = [];
for (const run of runs) {
runtimeEvents.push(
...(await this.deps.runtimeEventStore!.readRuntimeEvents(sessionId, run.runId)),
);
}
await this.deps.cleanupHistoryCompactArtifacts!({
sessionId,
checkpoint,
runtimeEvents,
});
})
.catch(() => {
// Legacy cleanup is reclaim-only. Runtime replay must remain available on failure.
})
.finally(() => {
if (this.historyCompactCleanupWrites.get(sessionId) === tracked) {
this.historyCompactCleanupWrites.delete(sessionId);
}
});
this.historyCompactCleanupWrites.set(sessionId, tracked);
}
private async ensureActive(
sessionId: string,
header: SessionHeader,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
await this.clearBackendQuarantineForActivation(sessionId, execution);
let existing = this.active.get(sessionId);
if (existing) {
existing.cachedHeader = header;
return existing;
}
await this.waitForBackendDisposal(sessionId);
existing = this.active.get(sessionId);
if (existing) {
existing.cachedHeader = header;
return existing;
}
const entry = await this.shareBackendActivation(`parent:${sessionId}`, async () => {
const current = this.active.get(sessionId);
if (current) return current;
const subagent = this.resolveSubagentActivation(header);
const backend = await this.deps.backends.build(header.backend, {
sessionId,
workspaceRoot: header.workspaceRoot,
header,
store: this.deps.store,
abortSignal: execution.abortController.signal,
...(subagent
? {
systemPrompt: subagent.systemPrompt,
tools: subagent.tools,
}
: {}),
recordRunTrace: (event) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(event.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordRunTrace(event);
},
...(this.deps.runStore
? {
recordProviderRequestCapture: (capture) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(capture.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
if (!run)
return Promise.reject(
new Error('No active AgentRun for provider request capture'),
);
return run.recordProviderRequestCapture(capture);
},
recordProviderRequestAttempt: (attempt) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(attempt.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordProviderRequestAttempt(attempt);
},
loadHistoryCompactCheckpoint: () => this.loadHistoryCompactCheckpoint(sessionId),
recordHistoryCompactCheckpoint: (
checkpoint: HistoryCompactCheckpoint,
turnId: string,
) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
return this.recordHistoryCompactCheckpoint(sessionId, checkpoint, run);
},
}
: {}),
...(this.deps.runtimeEventStore
? {
loadTurnRuntimeEvents: (turnId: string) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
if (!run)
return Promise.reject(new Error('No active AgentRun for turn runtime events'));
return run.loadTurnRuntimeEvents();
},
}
: {}),
allowMidTurnHistoryCompaction: Boolean(this.deps.runtimeEventStore),
recordActiveFullCompactBlock: (block) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(block.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordActiveFullCompactBlock(block);
},
recordSemanticCompactBlock: (block) => {
const active = this.active.get(sessionId);
const runId = active?.turnToRunId.get(block.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordSemanticCompactBlock(block);
},
shellRunContextSummary: () =>
this.deps.shellRuns?.buildContextSummary(sessionId) ?? Promise.resolve(undefined),
});
await this.rejectCancelledBackendActivation(backend, header, { kind: 'parent' }, execution);
const generation = this.createBackendGeneration(sessionId, backend, header, {
kind: 'parent',
});
this.active.set(sessionId, generation);
return generation;
});
entry.cachedHeader = header;
return entry;
}
private async shareBackendActivation(
activationKey: string,
activate: () => Promise<BackendGeneration>,
): Promise<BackendGeneration> {
let activation = this.backendActivationBuilds.get(activationKey);
if (!activation) {
activation = activate();
this.backendActivationBuilds.set(activationKey, activation);
}
try {
return await activation;
} finally {
if (this.backendActivationBuilds.get(activationKey) === activation) {
this.backendActivationBuilds.delete(activationKey);
}
}
}
private resolveSubagentActivation(
header: SessionHeader,
): { systemPrompt: string; tools: MakaTool[] } | undefined {
const snapshot = header.subagentRuntime;
if (!snapshot) {
if (header.subagentParent) {
throw new Error('Linked child session is missing its durable runtime snapshot');
}
return undefined;
}
if (!header.subagentParent) {
throw new Error('Subagent runtime snapshot requires a linked child session');
}
if (!isPermissionModeWithinCeiling(header.permissionMode, snapshot.permissionCeiling)) {
throw new Error('Subagent runtime permission mode exceeds its durable ceiling');
}
const snapshotDefinition = {
id: snapshot.agentId,
permissionMode: header.permissionMode,
tools: snapshot.toolNames,
categoryPolicy: snapshot.categoryPolicy,
};
const availableTools = this.deps.childTools ?? [];
const tools = buildToolsForAgentDefinition(availableTools, snapshotDefinition);
if (tools.length !== snapshot.toolNames.length) {
throw new Error('Subagent runtime tool snapshot is unavailable');
}
return { systemPrompt: snapshot.systemPrompt, tools };
}
private async ensureChildActive(
activeKey: string,
sessionId: string,
header: SessionHeader,
systemPrompt: string,
tools: readonly MakaTool[],
execution: PendingExecutionClaim,
agentTeam?: AgentTeamExecutionContext,
): Promise<BackendGeneration> {
await this.clearBackendQuarantineForActivation(sessionId, execution);
let existing = this.childActive.get(activeKey);
if (existing) {
existing.cachedHeader = header;
return existing;
}
await this.waitForBackendDisposal(sessionId);
existing = this.childActive.get(activeKey);
if (existing) {
existing.cachedHeader = header;
return existing;
}
const entry = await this.shareBackendActivation(`child:${activeKey}`, async () => {
const current = this.childActive.get(activeKey);
if (current) return current;
const backend = await this.deps.backends.build(header.backend, {
sessionId,
workspaceRoot: header.workspaceRoot,
header,
store: this.deps.store,
abortSignal: execution.abortController.signal,
appendMessage: async () => {},
systemPrompt,
tools,
...(agentTeam ? { agentTeam } : {}),
recordRunTrace: (event) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(event.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordRunTrace(event);
},
...(this.deps.runStore
? {
recordProviderRequestCapture: (capture) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(capture.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
if (!run)
return Promise.reject(
new Error('No active AgentRun for provider request capture'),
);
return run.recordProviderRequestCapture(capture);
},
recordProviderRequestAttempt: (attempt) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(attempt.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordProviderRequestAttempt(attempt);
},
loadHistoryCompactCheckpoint: () => this.loadHistoryCompactCheckpoint(sessionId),
recordHistoryCompactCheckpoint: (
checkpoint: HistoryCompactCheckpoint,
turnId: string,
) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
return this.recordHistoryCompactCheckpoint(sessionId, checkpoint, run);
},
}
: {}),
...(this.deps.runtimeEventStore
? {
loadTurnRuntimeEvents: (turnId: string) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
if (!run)
return Promise.reject(new Error('No active AgentRun for turn runtime events'));
return run.loadTurnRuntimeEvents();
},
}
: {}),
// A child-only ledger cannot claim coverage of the parent session prefix.
allowMidTurnHistoryCompaction: false,
recordActiveFullCompactBlock: (block) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(block.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordActiveFullCompactBlock(block);
},
recordSemanticCompactBlock: (block) => {
const active = this.childActive.get(activeKey);
const runId = active?.turnToRunId.get(block.turnId);
const run = runId ? active?.activeRuns.get(runId) : undefined;
run?.recordSemanticCompactBlock(block);
},
});
await this.rejectCancelledBackendActivation(
backend,
header,
{ kind: 'child', activeKey },
execution,
);
const generation = this.createBackendGeneration(sessionId, backend, header, {
kind: 'child',
activeKey,
});
this.childActive.set(activeKey, generation);
return generation;
});
entry.cachedHeader = header;
return entry;
}
private async reserveParentRun(
sessionId: string,
header: SessionHeader,
run: AgentRun,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
const active = await this.ensureActive(sessionId, header, execution);
this.reserveGenerationRun(active, run);
return active;
}
private async reserveChildRun(
activeKey: string,
sessionId: string,
header: SessionHeader,
systemPrompt: string,
tools: readonly MakaTool[],
agentTeam: AgentTeamExecutionContext | undefined,
run: AgentRun,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
const active = await this.ensureChildActive(
activeKey,
sessionId,
header,
systemPrompt,
tools,
execution,
agentTeam,
);
this.reserveGenerationRun(active, run);
return active;
}
private createBackendGeneration(
sessionId: string,
backend: AgentBackend,
header: SessionHeader,
route: BackendGeneration['route'],
): BackendGeneration {
const active: BackendGeneration = {
sessionId,
generation: ++this.nextBackendGeneration,
route,
phase: 'active',
backend,
stopBackend: undefined as never,
stopState: { kind: 'idle' },
cachedHeader: header,
activeRuns: new Map(),
turnToRunId: new Map(),
};
active.stopBackend = this.createBackendStopOwner(active);
this.backendGenerations.set(active.generation, active);
return active;
}
private async rejectCancelledBackendActivation(
backend: AgentBackend,
header: SessionHeader,
route: BackendGeneration['route'],
execution: PendingExecutionClaim,
): Promise<void> {
if (!execution.abortController.signal.aborted) return;
const generation = this.createBackendGeneration(execution.sessionId, backend, header, route);
const disposal = await this.quarantineBackendGeneration(generation);
if (!disposal.ok) {
throw new AggregateError(
[execution.cancellation, disposal.error],
`Cancelled backend activation disposal failed for session ${execution.sessionId}`,
);
}
throw execution.cancellation;
}
private reserveGenerationRun(active: BackendGeneration, run: AgentRun): void {
if (
active.phase !== 'active' ||
this.backendGenerations.get(active.generation) !== active ||
!this.isCurrentGeneration(active)
) {
throw new Error(
`Backend generation ${active.generation} no longer owns activation for session ${active.sessionId}`,
);
}
if (active.activeRuns.has(run.runId) || active.turnToRunId.has(run.turnId)) {
throw new Error(`Backend generation ${active.generation} already reserved this Run identity`);
}
active.activeRuns.set(run.runId, run);
active.turnToRunId.set(run.turnId, run.runId);
}
private assertRunCanDispatch(run: AgentRun, backend: AgentBackend): void {
const active = [...this.backendGenerations.values()].find(
(candidate) => candidate.backend === backend,
);
if (
run.isStopped() ||
!active ||
active.phase !== 'active' ||
!this.isCurrentGeneration(active) ||
active.activeRuns.get(run.runId) !== run ||
active.turnToRunId.get(run.turnId) !== run.runId
) {
throw new Error(`Run ${run.runId} no longer owns an active backend generation`);
}
}
private isCurrentGeneration(active: BackendGeneration): boolean {
return active.route.kind === 'parent'
? this.active.get(active.sessionId) === active
: this.childActive.get(active.route.activeKey) === active;
}
private unregisterRun(active: AgentRunActiveSession, run: AgentRun): void {
active.activeRuns.delete(run.runId);
if (active.turnToRunId.get(run.turnId) === run.runId) {
active.turnToRunId.delete(run.turnId);
}
}
private async unregisterParentRun(active: AgentRunActiveSession, run: AgentRun): Promise<void> {
this.unregisterRun(active, run);
await this.settleRunStopOperation(active.sessionId, run);
await this.settleBackendGenerationAfterRunExit(active as BackendGeneration);
await this.flushBackendInvalidation(active.sessionId);
}
private async unregisterChildRun(active: AgentRunActiveSession, run: AgentRun): Promise<void> {
this.unregisterRun(active, run);
if (active.activeRuns.size > 0) return;
await this.settleRunStopOperation(active.sessionId, run);
const generation = active as BackendGeneration;
await this.settleBackendGenerationAfterRunExit(generation);
if (generation.phase === 'active') {
await this.quarantineBackendGeneration(generation);
}
await this.settleBackendGenerationAfterRunExit(generation);
await this.flushBackendInvalidation(active.sessionId);
}
private async settleRunStopOperation(sessionId: string, run: AgentRun): Promise<void> {
const operation = this.stopOperations.get(sessionId);
if (
!operation ||
![...operation.targets.values()].some((target) => target.runs.get(run.runId)?.run === run)
) {
return;
}
try {
await this.enqueueStopOperation(sessionId, operation, {}, false);
} catch {
// A later public retry continues the retained canonical projection.
} finally {
this.releaseStoppedRunReferences(operation, run);
}
}
private releaseStoppedRunReferences(operation: StopOperation, run: AgentRun): void {
for (const target of operation.targets.values()) {
const stoppedRun = target.runs.get(run.runId);
if (stoppedRun?.run === run) stoppedRun.run = undefined;
if (
target.active &&
target.delivery.kind !== 'pending' &&
![...target.runs.values()].some((candidate) => candidate.run)
) {
target.active = undefined;
}
}
}
private async settleBackendGenerationAfterRunExit(active: BackendGeneration): Promise<void> {
if (active.activeRuns.size > 0 || this.stopOperationReferences(active)) return;
if (active.phase === 'stopping') {
active.phase = 'active';
return;
}
if (active.phase !== 'disposing') return;
const outcome = await active.disposal;
if (outcome?.ok) this.terminateBackendGeneration(active);
}
private async flushBackendInvalidation(sessionId: string): Promise<void> {
const invalidation = this.backendInvalidations.get(sessionId);
if (!invalidation || this.hasActiveRuns(sessionId)) return;
await this.startBackendDisposal(sessionId, invalidation);
}
private async waitForBackendDisposal(sessionId: string): Promise<void> {
const invalidation = this.backendInvalidations.get(sessionId);
if (!invalidation?.disposal) return;
const outcome = await invalidation.outcome;
if (!outcome.ok) throw invalidation.failure ?? outcome.error;
}
private async clearBackendQuarantineForActivation(
sessionId: string,
execution: PendingExecutionClaim,
): Promise<void> {
const ownsCurrentStop =
execution.phase === 'attached' &&
execution.stopIntent !== undefined &&
this.stopIntents.get(sessionId) === execution.stopIntent;
if (this.stopOperations.has(sessionId) && !ownsCurrentStop) {
throw new Error(`Session ${sessionId} is quarantined by a retained stop operation`);
}
for (const generation of this.backendGenerationsFor(sessionId)) {
if (generation.phase === 'failed') {
throw generation.disposalFailure ?? new Error('Backend generation disposal failed');
}
if (generation.phase === 'stopping') {
throw new Error(
`Backend generation ${generation.generation} is stopping for session ${sessionId}`,
);
}
if (generation.phase === 'disposing') {
const outcome = await generation.disposal;
if (!outcome?.ok) {
throw generation.disposalFailure ?? outcome?.error;
}
if (generation.activeRuns.size > 0 || this.stopOperationReferences(generation)) {
throw new Error(
`Backend generation ${generation.generation} is quarantined for session ${sessionId}`,
);
}
this.terminateBackendGeneration(generation);
}
}
const invalidation = this.backendInvalidations.get(sessionId);
if (!invalidation) return;
await this.flushBackendInvalidation(sessionId);
if (this.hasActiveRuns(sessionId)) {
throw new Error(`Backend generation is quarantined for session ${sessionId}`);
}
await this.startBackendDisposal(sessionId, invalidation);
const outcome = await invalidation.outcome;
if (!outcome.ok) throw invalidation.failure ?? outcome.error;
}
private async startBackendDisposal(
sessionId: string,
invalidation: BackendInvalidationState,
): Promise<void> {
if (!invalidation.disposal) {
invalidation.disposal = (async () => {
let outcome: BackendDisposalOutcome;
try {
outcome = await this.disposeBackendNow(sessionId);
} catch (error) {
outcome = { ok: false, error };
}
if (!outcome.ok) {
invalidation.failure = new Error(
`Backend invalidation is permanently quarantined for session ${sessionId}`,
{ cause: outcome.error },
);
}
invalidation.resolve(outcome);
if (outcome.ok && this.backendInvalidations.get(sessionId) === invalidation) {
this.backendInvalidations.delete(sessionId);
}
})();
}
await invalidation.disposal;
}
private ensureBackendInvalidation(sessionId: string): BackendInvalidationState {
const existing = this.backendInvalidations.get(sessionId);
if (existing) return existing;
let resolve!: (outcome: BackendDisposalOutcome) => void;
const outcome = new Promise<BackendDisposalOutcome>((resolvePromise) => {
resolve = resolvePromise;
});
const invalidation = { outcome, resolve };
this.backendInvalidations.set(sessionId, invalidation);
return invalidation;
}
private async updateStatus(
sessionId: string,
status: SessionStatus,
blockedReason?: SessionBlockedReason,
ts = this.deps.now(),
): Promise<void> {
await this.updateHeader(sessionId, buildStatusPatch(status, ts, blockedReason));
}
private async updateHeader(
sessionId: string,
patch: Partial<SessionHeader>,
): Promise<SessionHeader> {
const next = await this.deps.store.updateHeader(sessionId, patch);
this.updateCachedHeader(sessionId, next);
return next;
}
private async appendTurnState(
sessionId: string,
turnId: string,
status: TurnRecord['status'],
lineage: AgentRunLineage = {},
options: { id?: string; ts?: number; errorClass?: string; abortSource?: string } = {},
): Promise<void> {
const ts = options.ts ?? this.deps.now();
await this.deps.store.appendMessage(
sessionId,
buildTurnStateMessage({
id: options.id ?? this.deps.newId(),
turnId,
ts,
status,
lineage,
...(options.abortSource ? { abortSource: options.abortSource } : {}),
...(options.errorClass !== undefined ? { errorClass: options.errorClass } : {}),
partialOutputRetained: await this.turnHasRetainedOutput(sessionId, turnId),
}),
);
}
private async turnHasRetainedOutput(sessionId: string, turnId: string): Promise<boolean> {
const messages = await this.deps.store.readMessages(sessionId).catch(() => []);
return messagesHaveRetainedOutput(messages, turnId);
}
}
function assertContinuationSourceUnchanged(
continuation: RuntimeContinuation,
sourceRun: AgentRunHeader,
sourceEvents: readonly RuntimeEvent[],
): void {
if (
sourceRun.runId !== continuation.sourceRunId ||
sourceRun.turnId !== continuation.sourceTurnId ||
sourceRun.sessionId !== continuation.sessionId
) {
throw new RuntimeContinuationRevalidationError(
'source_identity_changed',
'Runtime continuation source run identity changed after planning',
);
}
const terminalEvents = matchingTerminalRuntimeEvents(sourceRun, sourceEvents);
const terminalStatus =
terminalEvents.length === 1 ? terminalRunStatusFromRuntimeEvent(terminalEvents[0]!) : undefined;
if (terminalStatus === undefined || terminalStatus !== sourceRun.status) {
throw new RuntimeContinuationRevalidationError(
'source_terminal_changed',
'Runtime continuation source is no longer terminal',
);
}
if (sourceEvents.length !== continuation.sourceRuntimeEventHighWater) {
throw new RuntimeContinuationRevalidationError(
'source_high_water_changed',
'Runtime continuation source high-water changed after planning',
);
}
const mismatchedEvent = sourceEvents.find(
(event) =>
event.sessionId !== continuation.sessionId ||
event.invocationId !== continuation.sourceInvocationId ||
event.runId !== continuation.sourceRunId ||
event.turnId !== continuation.sourceTurnId,
);
if (mismatchedEvent) {
throw new RuntimeContinuationRevalidationError(
'source_ledger_identity_changed',
'Runtime continuation source ledger identity changed after planning',
);
}
const replayPlan = buildResumePlanFromRuntimeEvents(sourceEvents, {
expectedRuntimeEventHighWater: continuation.sourceRuntimeEventHighWater,
});
const sourceRuntimeContext = continuation.sourceRuntimeContext ?? continuation.runtimeContext;
if (
replayPlan.disposition !== 'safe_replay' ||
!isDeepStrictEqual(replayPlan.replayRuntimeEvents, sourceRuntimeContext)
) {
throw new RuntimeContinuationRevalidationError(
'source_replay_changed',
'Runtime continuation replay context changed after planning',
);
}
}
function assertContinuationSafetyUnchanged(
continuation: RuntimeContinuation,
observation: RuntimeContinuationSafetyObservation,
): void {
const snapshot = continuation.safetySnapshot;
if (observation.workspaceIdentity !== snapshot.workspaceIdentity) {
throw new RuntimeContinuationRevalidationError(
'workspace_identity_changed',
'Runtime continuation workspace identity changed after planning',
);
}
if (!observation.backgroundOperationsSettled) {
throw new RuntimeContinuationRevalidationError(
'background_operation_started',
'Runtime continuation background operation started after planning',
);
}
const availableToolNames = new Set(observation.availableToolNames);
const missingToolNames = snapshot.availableToolNames.filter(
(name) => !availableToolNames.has(name),
);
if (missingToolNames.length > 0) {
throw new RuntimeContinuationRevalidationError(
'tool_catalog_changed',
`Runtime continuation tool catalog changed after planning: ${missingToolNames.join(', ')}`,
);
}
if (snapshot.workspaceCheckpoint) {
const current = observation.workspaceCheckpoint;
if (
!current?.restored ||
current.ref !== snapshot.workspaceCheckpoint.ref ||
current.runtimeEventHighWater !== snapshot.workspaceCheckpoint.runtimeEventHighWater
) {
throw new RuntimeContinuationRevalidationError(
'workspace_checkpoint_changed',
'Runtime continuation workspace checkpoint changed after planning',
);
}
}
}
interface RuntimeRunOwnerScopeCallbacks {
registerInteraction(binding: RuntimeInteractionRunBinding): void;
releaseInteraction(binding: RuntimeInteractionRunBinding): void;
settleReservedExecution(outcome: ExecutionClaimOutcome): void;
finalizeExecution(operation: () => Promise<void>): Promise<void>;
}
class RuntimeRunOwnerScope {
interactionRun: RuntimeInteractionRunBinding | undefined;
messageOwner: RuntimeMessageRunOwner | undefined;
private messageReleased = false;
private reservedExecutionSettled = false;
private finalizePromise: Promise<void> | undefined;
constructor(
private readonly run: AgentRun,
private readonly callbacks: RuntimeRunOwnerScopeCallbacks,
) {}
async bindInteraction(
authority: RuntimeInteractionAuthority | undefined,
identity: { sessionId: string; turnId: string; runId: string },
): Promise<void> {
try {
if (authority) {
this.interactionRun = await bindRuntimeInteractionRun(authority, identity);
this.callbacks.registerInteraction(this.interactionRun);
}
} catch (error) {
this.settleReservedExecution({ ok: false, error });
throw error;
}
this.settleReservedExecution({ ok: true });
}
bindMessage(
authority: RuntimeMessageAuthority | undefined,
identity: { sessionId: string; turnId: string; runId: string },
): void {
if (!authority) return;
this.messageOwner = authority.bindRun(identity);
}
async failStart(error: unknown): Promise<never> {
this.settleReservedExecution({ ok: false, error });
let failure = error;
if (this.interactionRun) {
try {
await this.interactionRun.close(interactionClosureReason(this.run));
await this.interactionRun.settleLocalClosures();
this.callbacks.releaseInteraction(this.interactionRun);
} catch (closeError) {
failure = new AggregateError(
[failure, closeError],
'Interaction owner bind cleanup failed',
);
}
}
try {
this.releaseMessage();
} catch (releaseError) {
failure = new AggregateError([failure, releaseError], 'Message owner bind cleanup failed');
}
await this.run.recordFailure(failure);
await this.callbacks.finalizeExecution(() => this.run.finalize());
throw failure;
}
finalize(): Promise<void> {
if (!this.finalizePromise) {
this.interactionRun?.sealPublications();
this.finalizePromise = this.callbacks.finalizeExecution(() => this.finalizeOwnedRun());
}
return this.finalizePromise;
}
releaseMessage(): void {
if (!this.messageOwner || this.messageReleased) return;
this.messageReleased = true;
try {
this.messageOwner.release();
} catch (error) {
throw runtimeOwnerCleanupFailure(`Message owner release failed for ${this.run.runId}`, error);
}
}
private settleReservedExecution(outcome: ExecutionClaimOutcome): void {
if (this.reservedExecutionSettled) return;
this.reservedExecutionSettled = true;
this.callbacks.settleReservedExecution(outcome);
}
private async finalizeOwnedRun(): Promise<void> {
const failures = new FailureCollector();
const interactionRun = this.interactionRun;
if (interactionRun) {
await failures.capture(async () => {
await interactionRun.close(interactionClosureReason(this.run));
await interactionRun.settleLocalClosures();
});
}
await failures.capture(() => this.run.finalize());
if (!failures.hasFailures && interactionRun) {
await failures.capture(() => this.callbacks.releaseInteraction(interactionRun));
}
const message = `Interaction and Run finalization failed for ${this.run.runId}`;
try {
failures.throwIfAny(message);
} catch (error) {
throw runtimeOwnerCleanupFailure(message, error);
}
}
}
function childActiveKey(sessionId: string, turnId: string): string {
return `${sessionId}:${turnId}`;
}
function runtimeToolBoundaryProtocol(
deps: Pick<RuntimeKernelDeps, 'toolBoundaryProtocol'>,
header: Pick<SessionHeader, 'backend'>,
): ToolBoundaryProtocol | undefined {
return header.backend === 'ai-sdk' ? deps.toolBoundaryProtocol : undefined;
}
function effectiveOrchestrationForRun(
run: AgentRunHeader,
session: SessionHeader,
): EffectiveOrchestration {
if (
run.orchestrationMode !== undefined &&
run.orchestrationSource !== undefined &&
run.agentSwarmAuthorization !== undefined
) {
return {
mode: run.orchestrationMode,
source: run.orchestrationSource,
agentSwarmAuthorization: run.agentSwarmAuthorization,
};
}
return resolveEffectiveOrchestration(session.orchestrationMode, undefined);
}
class AsyncEventQueueClosed extends Error {
constructor() {
super('Async event queue closed');
this.name = 'AsyncEventQueueClosed';
}
}
function isAsyncEventQueueClosed(error: unknown): boolean {
return error instanceof AsyncEventQueueClosed;
}
async function interactionResumeAllowed(
interactionRun: RuntimeInteractionRunBinding | undefined,
event: SessionEvent,
): Promise<boolean> {
if (
!interactionRun ||
(event.type !== 'permission_answer_ack' &&
event.type !== 'permission_closure_ack' &&
event.type !== 'permission_decision_ack' &&
event.type !== 'user_question_answer_ack')
) {
return true;
}
return await interactionRun.canResumeAfterSettlementAck(event);
}
function interactionClosureReason(run: AgentRun): RuntimeInteractionRunClosureReason {
return run.isStopped() ? 'turn_stopped' : 'turn_terminal';
}
function interactionFailStop(message: string, error: unknown): Error {
return error instanceof RuntimeInteractionFailStopError
? error
: new RuntimeInteractionFailStopError(message, error);
}
function runtimeOwnerCleanupFailure(message: string, error: unknown): Error {
return error instanceof RuntimeOwnerCleanupError ||
error instanceof RuntimeMessageAuthorityInvariantError ||
error instanceof RuntimeInteractionInvariantError ||
error instanceof RuntimeInteractionFailStopError
? error
: new RuntimeOwnerCleanupError(message, error);
}
function containsRuntimeOwnerCleanupFailure(error: unknown): boolean {
if (
error instanceof RuntimeOwnerCleanupError ||
error instanceof RuntimeMessageAuthorityInvariantError ||
error instanceof RuntimeInteractionInvariantError ||
error instanceof RuntimeInteractionFailStopError
) {
return true;
}
return (
error instanceof AggregateError &&
error.errors.some((nested) => containsRuntimeOwnerCleanupFailure(nested))
);
}
class RuntimeExecutionCancellation extends Error {
constructor(sessionId: string) {
super(`Execution for session ${sessionId} was cancelled before dispatch`);
this.name = 'RuntimeExecutionCancellation';
}
}
function isExecutionCancellation(
error: unknown,
cancellation: RuntimeExecutionCancellation,
): boolean {
const seen = new Set<unknown>();
let current = error;
while (current !== null && (typeof current === 'object' || typeof current === 'function')) {
if (current === cancellation) return true;
if (seen.has(current)) return false;
seen.add(current);
current = 'cause' in current ? current.cause : undefined;
}
return false;
}
class FailureCollector {
private readonly failures: unknown[] = [];
private readonly seen = new Set<unknown>();
get hasFailures(): boolean {
return this.failures.length > 0;
}
add(error: unknown): void {
if (error instanceof AggregateError) {
for (const nested of error.errors) this.add(nested);
return;
}
if (this.seen.has(error)) return;
this.seen.add(error);
this.failures.push(error);
}
async capture(operation: () => Promise<unknown> | unknown): Promise<void> {
try {
await operation();
} catch (error) {
this.add(error);
}
}
throwIfAny(message: string): void {
if (this.failures.length === 1) throw this.failures[0];
if (this.failures.length > 1) throw new AggregateError(this.failures, message);
}
}
interface AsyncEventQueueEntry<T> {
value: T;
delivered: () => void;
rejected: (error: unknown) => void;
}
class AsyncEventQueue<T> implements AsyncIterable<T> {
private readonly values: Array<AsyncEventQueueEntry<T>> = [];
private readonly waiters: Array<{
resolve: (entry: AsyncEventQueueEntry<T> | undefined) => void;
reject: (error: unknown) => void;
}> = [];
private closed = false;
private failure: unknown;
[Symbol.asyncIterator](): AsyncIterator<T> {
return this.consume()[Symbol.asyncIterator]();
}
push(value: T): Promise<void> {
if (this.failure) return Promise.reject(this.failure);
if (this.closed) return Promise.reject(new AsyncEventQueueClosed());
return new Promise<void>((resolve, reject) => {
const entry = { value, delivered: resolve, rejected: reject };
const waiter = this.waiters.shift();
if (waiter) {
waiter.resolve(entry);
return;
}
this.values.push(entry);
});
}
fail(error: unknown): void {
if (this.failure) return;
this.failure = error;
for (const value of this.values.splice(0)) value.rejected(error);
for (const waiter of this.waiters.splice(0)) waiter.reject(error);
}
close(): void {
if (this.closed) return;
this.closed = true;
const closed = new AsyncEventQueueClosed();
for (const value of this.values.splice(0)) value.rejected(closed);
for (const waiter of this.waiters.splice(0)) waiter.resolve(undefined);
}
private async *consume(): AsyncIterable<T> {
while (true) {
const entry = await this.nextEntry();
if (!entry) return;
try {
yield entry.value;
} finally {
entry.delivered();
}
}
}
private nextEntry(): Promise<AsyncEventQueueEntry<T> | undefined> {
if (this.values.length > 0) {
const next = this.values.shift()!;
return Promise.resolve(next);
}
if (this.failure) return Promise.reject(this.failure);
if (this.closed) return Promise.resolve(undefined);
return new Promise<AsyncEventQueueEntry<T> | undefined>((resolve, reject) => {
this.waiters.push({ resolve, reject });
});
}
}
export type { AgentRunLineage };