Google GitGoogle Git
Sign in
apache/maka/refs/heads/codex/remove-duplicate-runtime-architecture/./packages/runtime/src/runtime-kernel.ts
blob: 3e60cd4c627e8aa4a9368efa5f5e19b328ba5755 [file]
import type { AgentRunHeader, AgentRunStore } from '@maka/core/agent-run';
import type { ContinuationClaimV1, ImmutableRuntimePrefixV1 } from '@maka/core/runtime-boundary';
import type { RuntimeEvent, ToolBoundaryProtocol } from '@maka/core/runtime-event';
import type {
RuntimeContinuationAuthorityStore,
RuntimeEventStore,
} from '@maka/core/runtime-event-store';
import { isSessionInlineRun } from '@maka/core/agent-run';
import type {
ActiveInteractionRequestEvent,
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 { SandboxBoundaryResponse } from '@maka/core/sandbox-boundary';
import {
resolveEffectiveOrchestration,
type EffectiveOrchestration,
} from '@maka/core/orchestration';
import type { UserQuestionResponse } from '@maka/core/user-question';
import { DEFAULT_TOOL_MODE, type ToolMode } from '@maka/core/tool-mode';
import {
AgentRun,
ContinuationStartCommitError,
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 { MakaTool } from './tool-runtime.js';
import type {
InvocationContext,
InvocationResult,
InvocationSource,
} from './invocation-context.js';
import {
issueRuntimeContinuationAdmissionReceipt,
RuntimeRunner,
runAdmittedRuntimeContinuation,
runLegacyProviderRetry,
} from './runtime-runner.js';
import type {
BackendFactoryContext,
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,
requireBuiltinAgentDefinition,
} from './agent-catalog.js';
import { loadLatestHistoryCompactCheckpointFromRunLedger } from './history-compact-ledger.js';
import {
canReplaceHistoryCompactCheckpoint,
type HistoryCompactCheckpoint,
} from './history-compact-checkpoint.js';
import {
HistoryCompactCheckpointCoordinator,
type HistoryCompactCleanupRequest,
} from './history-compact-checkpoint-coordinator.js';
import { shouldAppendContextCompactionFailedOpenNote } from './context-budget.js';
import {
buildResumePlanFromRuntimeEvents,
RuntimeContinuationRevalidationError,
type RuntimeContinuation,
type RuntimeContinuationSafetyObservation,
} from './runtime-resume.js';
import { buildContinuationReplayPlan } from './continuation-replay.js';
import { PROVIDER_REPLAY_PROJECTION_VERSION } from './model-history.js';
import {
matchingTerminalRuntimeEvents,
terminalRunStatusFromRuntimeEvent,
} from './terminal-run-commit.js';
import {
RuntimeMessageAuthorityInvariantError,
type RuntimeMessageAuthority,
type RuntimeMessageRunIdentity,
type RuntimeMessageRunOwner,
} from './message-authority.js';
import {
RuntimeInteractionFailStopError,
RuntimeInteractionInvariantError,
bindRuntimeInteractionRun,
type RuntimeInteractionAuthority,
type RuntimeInteractionRunBinding,
type RuntimeInteractionRunClosureReason,
} from './interaction-authority.js';
import { DeliveryAckQueue, isDeliveryAckQueueClosed } from './delivery-ack-queue.js';
export interface RuntimeKernelLike {
claimExecution(sessionId: string): RuntimeExecutionClaim;
runSessionAdmissionMutation?<T>(
sessionIds: readonly string[],
operation: () => Promise<T> | T,
): Promise<T>;
runSessionQuiescentMutation?<T>(
sessionIds: readonly string[],
operation: () => Promise<T> | T,
): Promise<T>;
startTurn(
sessionId: string,
input: UserMessageInput,
options?: TurnStartOptions,
): AsyncIterable<SessionEvent>;
resumeContinuation?(
continuation: RuntimeContinuation,
options?: ResumeContinuationOptions,
): AsyncIterable<SessionEvent>;
compactSession(sessionId: string, input?: CompactSessionInput): AsyncIterable<SessionEvent>;
preflightContextCompaction(sessionId: string): Promise<void>;
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>;
respondToSandboxBoundary(sessionId: string, response: SandboxBoundaryResponse): Promise<void>;
listActiveInteractions?(sessionId: string): ActiveInteractionRequestEvent[];
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;
/**
* The turns of the runs in flight for this session. The same fact
* `hasActiveRuns` reports, named — which is what lets a client tell a turn
* that has not started yet from one that already ended.
*
* A set, not one turn: a session can carry concurrent runs, and a client
* asking "is anything OTHER than my own turn running" cannot answer that
* from an arbitrary one of them.
*/
runningTurnIds?(sessionId: string): string[];
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 class SessionQuiescentMutationBusyError extends Error {
readonly name = 'SessionQuiescentMutationBusyError';
constructor(readonly sessionIds: readonly string[]) {
super('Session mutation cannot start while an execution claim is active');
}
}
export class RuntimeContextCompactError extends Error {
readonly name = 'RuntimeContextCompactError';
constructor(
readonly code: 'operation_unavailable' | 'session_busy',
message: string,
) {
super(message);
}
}
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 ResumeContinuationOptions {
onRunStarted?: () => void | Promise<void>;
}
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;
/**
* Chosen by SessionManager before any claim, Run creation, or provider T1.
* There is no fallback between these modes after execution begins.
*/
admissionMode: 'durable_continuation' | 'legacy_provider_retry';
/** 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 type { HistoryCompactCleanupRequest } from './history-compact-checkpoint-coordinator.js';
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 admissionBarrier: Promise<void>;
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;
}
interface InteractionRequestOwner {
sessionId: string;
turnId: string;
generation: number;
request: ActiveInteractionRequestEvent;
}
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 sessionMutationTails = new Map<string, Promise<void>>();
private readonly executionClaimStates = new WeakMap<
RuntimeExecutionClaim,
PendingExecutionClaim
>();
private readonly stopIntents = new Map<string, SessionStopIntent>();
private readonly historyCompactCoordinator: HistoryCompactCheckpointCoordinator;
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 readonly interactionRequestOwners = new Map<string, InteractionRequestOwner>();
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');
}
this.historyCompactCoordinator = new HistoryCompactCheckpointCoordinator(deps);
}
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,
admissionBarrier: this.sessionMutationTails.get(sessionId) ?? Promise.resolve(),
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;
}
async runSessionAdmissionMutation<T>(
sessionIds: readonly string[],
operation: () => Promise<T> | T,
): Promise<T> {
const ids = this.normalizeSessionMutationIds(sessionIds);
return this.enqueueSessionMutation(ids, operation);
}
async runSessionQuiescentMutation<T>(
sessionIds: readonly string[],
operation: () => Promise<T> | T,
): Promise<T> {
const ids = this.normalizeSessionMutationIds(sessionIds);
if (ids.some((sessionId) => (this.executionClaims.get(sessionId)?.size ?? 0) > 0)) {
throw new SessionQuiescentMutationBusyError(ids);
}
return this.enqueueSessionMutation(ids, operation);
}
private normalizeSessionMutationIds(sessionIds: readonly string[]): string[] {
const ids = [...new Set(sessionIds)].sort();
if (ids.length === 0 || ids.some((sessionId) => sessionId.length === 0)) {
throw new Error('Session mutation requires at least one valid Session identity');
}
return ids;
}
private async enqueueSessionMutation<T>(
ids: readonly string[],
operation: () => Promise<T> | T,
): Promise<T> {
const precedingMutations = ids.map(
(sessionId) => this.sessionMutationTails.get(sessionId) ?? Promise.resolve(),
);
const preceding = Promise.all(precedingMutations).then(() => undefined);
let complete!: () => void;
const completion = new Promise<void>((resolve) => {
complete = resolve;
});
const tail = preceding.then(() => completion);
for (const sessionId of ids) this.sessionMutationTails.set(sessionId, tail);
void tail.then(() => {
for (const sessionId of ids) {
if (this.sessionMutationTails.get(sessionId) === tail) {
this.sessionMutationTails.delete(sessionId);
}
}
});
try {
await preceding;
return await operation();
} finally {
complete();
}
}
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 async enterExecutionClaim(execution: PendingExecutionClaim): Promise<void> {
await execution.admissionBarrier;
if (execution.phase !== 'pending') {
throw new Error(
`Execution claim cannot enter admission from phase ${execution.phase} for session ${execution.sessionId}`,
);
}
}
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 {
await this.enterExecutionClaim(execution);
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(
continuationInput: RuntimeContinuation,
options: ResumeContinuationOptions = {},
): AsyncIterable<SessionEvent> {
const continuation = snapshotRuntimeContinuation(continuationInput);
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, options);
} finally {
this.pendingContinuationClaims.delete(claimKey);
this.pendingContinuationSessions.delete(continuation.sessionId);
this.releaseExecutionClaim(execution);
}
}
private async *resumeContinuationClaimed(
continuation: RuntimeContinuation,
execution: PendingExecutionClaim,
options: ResumeContinuationOptions,
): AsyncIterable<SessionEvent> {
await this.enterExecutionClaim(execution);
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new Error('Runtime continuation requires AgentRunStore and RuntimeEventStore');
}
const continuationAuthority = requireRuntimeContinuationAuthority(this.deps.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 revalidateContinuationBoundary(continuationAuthority, continuation);
assertContinuationSourceUnchanged(continuation, sourceRun, sourceEvents);
await this.revalidateContinuationSafety(continuation);
const userInput: UserMessageInput = {
turnId: continuation.turnId,
text: '',
parentRunId: continuation.sourceRunId,
parentTurnId: continuation.sourceTurnId,
};
const effectiveOrchestration = effectiveOrchestrationForRun(sourceRun, header);
const effectiveToolMode = effectiveToolModeForRun(sourceRun);
const claimedAt = this.deps.now();
const targetRunHeader = continuationTargetRunHeaderForExecution({
continuation,
sessionHeader: header,
userInput,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration,
effectiveToolMode,
claimedAt,
});
const claim = continuationClaimForExecution(continuation, claimedAt, targetRunHeader);
const claimResult = await continuationAuthority.claimContinuation({ claim });
if (claimResult.kind !== 'acquired') {
throw new RuntimeContinuationRevalidationError(
'continuation_claim_conflict',
`Runtime continuation boundary is already claimed by ${claimResult.claim.claimId}`,
);
}
await this.deps.continuationFailpoint?.('after_continuation_claim_committed');
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 continuationToolBoundaryProtocol = runtimeToolBoundaryProtocol(this.deps, header);
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,
...(continuationToolBoundaryProtocol
? { toolBoundaryProtocol: continuationToolBoundaryProtocol }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration,
claimedRunHeader: claim.targetRunHeader,
effectiveToolMode,
continuationFailpoint: this.deps.continuationFailpoint,
commitContinuationStart: async (startedAt) => {
const source = claim.boundary.segments.at(-1)!;
const eventId = this.deps.newId();
const result = await continuationAuthority.commitContinuationStart({
claim,
event: {
id: eventId,
...claim.target,
ts: startedAt,
partial: false,
role: 'system',
author: 'system',
actions: {
...(continuationToolBoundaryProtocol
? {
runtimeProtocol: {
toolBoundary: continuationToolBoundaryProtocol,
},
}
: {}),
continuationStart: {
protocol: 'continuation_start_v2',
provenance: 'runtime_admission',
claimId: claim.claimId,
boundaryDigest: claim.boundaryDigest,
immediateSource: {
sessionId: source.identity.sessionId,
invocationId: source.identity.invocationId,
runId: source.identity.runId,
turnId: source.identity.turnId,
highWater: source.position.lastEventSeq,
prefixDigest: source.prefixDigest,
},
replayManifestDigest: claim.boundary.manifestDigest,
providerProjectionVersion: claim.providerProjectionVersion,
providerReplayDigest: claim.providerReplayDigest,
},
},
},
});
if (!result.created) {
throw new Error(
'Continuation-start already existed; refusing to reissue provider admission',
);
}
return { startEventId: eventId, created: true };
},
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,
'durable_continuation',
{
sessionId: continuation.sessionId,
turnId: continuation.turnId,
runId: continuation.runId,
},
options.onRunStarted,
() => this.revalidateContinuationSafety(continuation),
);
}
async *compactSession(
sessionId: string,
input: CompactSessionInput = {},
): AsyncIterable<SessionEvent> {
const execution = this.takeExecutionClaim(sessionId);
try {
yield* this.compactSessionClaimed(sessionId, input, execution);
} finally {
this.releaseExecutionClaim(execution);
}
}
async preflightContextCompaction(sessionId: string): Promise<void> {
const execution = this.takeExecutionClaim(sessionId);
try {
await this.enterExecutionClaim(execution);
if (this.hasActiveRuns(sessionId)) {
throw new RuntimeContextCompactError(
'session_busy',
'Cannot compact while a Turn is running',
);
}
const header = await this.deps.store.readHeader(sessionId);
await this.requireContextCompactionBackend(sessionId, header, execution);
} finally {
this.releaseExecutionClaim(execution);
}
}
private async *compactSessionClaimed(
sessionId: string,
input: CompactSessionInput,
execution: PendingExecutionClaim,
): AsyncIterable<SessionEvent> {
await this.enterExecutionClaim(execution);
if (
input.minRecentTurns !== undefined &&
(!Number.isSafeInteger(input.minRecentTurns) || input.minRecentTurns < 0)
) {
throw new Error('Runtime compaction minRecentTurns must be a non-negative safe integer');
}
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new RuntimeContextCompactError(
'operation_unavailable',
'Runtime compaction requires execution stores',
);
}
if (this.hasActiveRuns(sessionId)) {
throw new RuntimeContextCompactError(
'session_busy',
'Cannot compact while a Turn is running',
);
}
const header = await this.deps.store.readHeader(sessionId);
const turnId = input.turnId ?? this.deps.newId();
const run = new AgentRun({
sessionId,
header,
userInput: { turnId, text: '' },
rootExecutionKind: 'context_compact',
...(input.hostedRoot ? { runId: input.hostedRoot.runId } : {}),
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);
const owners = this.createRunOwnerScope(run, execution);
let begin: Awaited<ReturnType<typeof run.beginOperation>>;
try {
if (input.hostedRoot) {
owners.bindMessage(this.deps.messageAuthority, {
sessionId,
turnId,
runId: run.runId,
});
}
begin = await this.runBackendActivation(() => run.beginOperation());
await input.hostedRoot?.onRunStarted?.();
this.settleReservedExecutionClaim(execution, run, { ok: true });
} catch (error) {
await this.finalizeFailedRunStart(owners, run, execution, error);
return;
}
try {
if (run.isStopped()) return;
if (!begin.backend.compactHistory) {
throw new Error(`Backend ${header.backend} changed runtime compaction capability`);
}
this.assertRunCanDispatch(run, begin.backend);
const result = await begin.backend.compactHistory({
turnId: run.turnId,
runId: run.runId,
runtimeContext: begin.runtimeContext,
...(input.minRecentTurns !== undefined ? { minRecentTurns: input.minRecentTurns } : {}),
});
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 {
const failures = new FailureCollector();
await failures.capture(() => owners.finalize());
await failures.capture(() => owners.releaseMessage());
failures.throwIfAny(`Runtime compaction cleanup failed for ${run.runId}`);
}
}
private async requireContextCompactionBackend(
sessionId: string,
header: SessionHeader,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
const active = await this.runBackendActivation(() =>
this.ensureActive(sessionId, header, execution),
);
if (!active.backend.compactHistory) {
throw new RuntimeContextCompactError(
'operation_unavailable',
`Backend ${header.backend} does not support runtime compaction`,
);
}
return active;
}
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> {
await this.enterExecutionClaim(execution);
const parentHeader = await this.deps.store.readHeader(sessionId);
const definition = requireBuiltinAgentDefinition(input.spec.id);
const availableChildTools = this.deps.childTools ?? [];
assertAgentDefinitionRunnable({
definition,
tools: availableChildTools,
});
const childTools = buildToolsForAgentDefinition(availableChildTools, definition);
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,
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 = snapshotRuntimeContinuation(input.continuation);
await this.enterExecutionClaim(execution);
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,
}
: requireBuiltinAgentDefinition(input.spec.id);
const availableChildTools = this.deps.childTools ?? [];
if (!linkedSnapshot) {
assertAgentDefinitionRunnable({
definition: requireBuiltinAgentDefinition(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 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 effectiveOrchestration = resolveEffectiveOrchestration('default', undefined);
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new Error('Child retry continuation requires AgentRunStore and RuntimeEventStore');
}
const sourceRun = await this.deps.runStore.readRun(sessionId, continuation.sourceRunId);
const effectiveToolMode = effectiveToolModeForRun(sourceRun);
let durableAdmission:
| {
claimedRunHeader: AgentRunHeader;
commitContinuationStart: (
startedAt: number,
) => Promise<{ startEventId: string; created: true }>;
}
| undefined;
if (input.admissionMode === 'durable_continuation') {
const continuationAuthority = requireRuntimeContinuationAuthority(
this.deps.runtimeEventStore,
);
const sourceEvents = await revalidateContinuationBoundary(
continuationAuthority,
continuation,
);
assertContinuationSourceUnchanged(continuation, sourceRun, sourceEvents);
await this.revalidateContinuationSafety(
continuation,
childTools.map((tool) => tool.name),
);
const claimedAt = this.deps.now();
const targetRunHeader = continuationTargetRunHeaderForExecution({
continuation,
sessionHeader: childHeader,
userInput,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration,
effectiveToolMode,
claimedAt,
});
const claim = continuationClaimForExecution(continuation, claimedAt, targetRunHeader);
const claimResult = await continuationAuthority.claimContinuation({
claim,
});
if (claimResult.kind !== 'acquired') {
throw new RuntimeContinuationRevalidationError(
'continuation_claim_conflict',
`Child retry continuation boundary is already claimed by ${claimResult.claim.claimId}`,
);
}
await this.deps.continuationFailpoint?.('after_continuation_claim_committed');
const continuationToolBoundaryProtocol = runtimeToolBoundaryProtocol(this.deps, childHeader);
durableAdmission = {
claimedRunHeader: claim.targetRunHeader,
commitContinuationStart: async (startedAt) => {
const source = claim.boundary.segments.at(-1)!;
const eventId = this.deps.newId();
const result = await continuationAuthority.commitContinuationStart({
claim,
event: {
id: eventId,
...claim.target,
ts: startedAt,
partial: false,
role: 'system',
author: 'system',
actions: {
...(continuationToolBoundaryProtocol
? {
runtimeProtocol: {
toolBoundary: continuationToolBoundaryProtocol,
},
}
: {}),
continuationStart: {
protocol: 'continuation_start_v2',
provenance: 'runtime_admission',
claimId: claim.claimId,
boundaryDigest: claim.boundaryDigest,
immediateSource: {
sessionId: source.identity.sessionId,
invocationId: source.identity.invocationId,
runId: source.identity.runId,
turnId: source.identity.turnId,
highWater: source.position.lastEventSeq,
prefixDigest: source.prefixDigest,
},
replayManifestDigest: claim.boundary.manifestDigest,
providerProjectionVersion: claim.providerProjectionVersion,
providerReplayDigest: claim.providerReplayDigest,
},
},
},
});
if (!result.created) {
throw new Error(
'Continuation-start already existed; refusing to reissue provider admission',
);
}
return { startEventId: eventId, created: true };
},
};
} else {
const readImmutable = this.deps.runtimeEventStore.readImmutableRuntimeEvents;
if (!readImmutable) {
throw new Error('Legacy child provider retry requires immutable RuntimeEvent reads');
}
if (sourceRun.status !== 'failed' || sourceRun.failureClass !== 'RateLimit') {
throw new Error('Legacy child provider retry requires a provider rate-limit failure');
}
const sourceEvents = await readImmutable.call(
this.deps.runtimeEventStore,
sessionId,
continuation.sourceRunId,
);
assertContinuationSourceUnchanged(continuation, sourceRun, sourceEvents);
}
const activeKey = childActiveKey(sessionId, continuation.turnId);
const continuationToolBoundaryProtocol = runtimeToolBoundaryProtocol(this.deps, childHeader);
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,
...(continuationToolBoundaryProtocol
? { toolBoundaryProtocol: continuationToolBoundaryProtocol }
: {}),
repairRunRuntimeLedger: this.deps.repairRunRuntimeLedger,
newId: this.deps.newId,
now: this.deps.now,
workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
effectiveOrchestration,
effectiveToolMode,
...(durableAdmission ?? {}),
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,
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);
// Both paths replay without a second user prompt. Only the durable mode
// consumes the source boundary through claim + continuation-start T1.
yield* this.runAgentContinuation(
continuation,
run,
execution,
input.admissionMode,
input.linkedSession === true
? {
sessionId,
turnId: continuation.turnId,
runId: continuation.runId,
}
: undefined,
input.onRunStarted,
input.admissionMode === 'durable_continuation'
? () =>
this.revalidateContinuationSafety(
continuation,
childTools.map((tool) => tool.name),
)
: undefined,
);
}
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 DeliveryAckQueue<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),
});
this.observeInteractionEvent(sessionId, begin.backend, sessionEvent);
await sessionEvents.push(sessionEvent);
},
onError: async (error) => {
if (!isDeliveryAckQueueClosed(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 }
: {}),
...(begin.backendInput.toolMode ? { toolMode: begin.backendInput.toolMode } : {}),
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 {
this.clearInteractionRequestOwners(sessionId, run.turnId);
releaseExecutionAbort();
}
}
}
private async *runAgentContinuation(
continuation: RuntimeContinuation,
run: AgentRun,
execution: PendingExecutionClaim,
admissionMode: ChildAgentRetryInput['admissionMode'],
messageOwner?: RuntimeMessageRunIdentity,
onRunStarted?: () => void | Promise<void>,
revalidateSafety?: () => Promise<void>,
): AsyncIterable<SessionEvent> {
const sessionEvents = new DeliveryAckQueue<SessionEvent>();
const { abortController, release: releaseExecutionAbort } =
this.inheritExecutionAbort(execution);
let flowDone = false;
const owners = this.createRunOwnerScope(run, execution);
let begin:
| Awaited<ReturnType<AgentRun['beginContinuation']>>
| Awaited<ReturnType<AgentRun['beginOperation']>>;
try {
if (messageOwner) owners.bindMessage(this.deps.messageAuthority, messageOwner);
begin = await this.runBackendActivation(async () => {
if (admissionMode === 'durable_continuation') {
if (!revalidateSafety) {
throw new Error('Durable continuation omitted final safety revalidation');
}
await revalidateSafety();
}
const started =
admissionMode === 'durable_continuation'
? 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();
if (
admissionMode === 'durable_continuation' &&
error instanceof ContinuationStartCommitError
) {
await owners.abandonUnstartedContinuation(error);
return;
}
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),
});
this.observeInteractionEvent(continuation.sessionId, begin.backend, sessionEvent);
await sessionEvents.push(sessionEvent);
},
onError: async (error) => {
if (!isDeliveryAckQueueClosed(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 } : {}),
});
if (run.isStopped()) abortController.abort();
let runnerFailure: unknown;
const runnerResult = (
admissionMode === 'durable_continuation'
? runAdmittedRuntimeContinuation(
runner,
issueRuntimeContinuationAdmissionReceipt(
runner,
continuation,
'continuationStartAdmission' in begin
? begin.continuationStartAdmission
: (() => {
throw new Error('Durable continuation is missing its start admission');
})(),
{ orchestration: run.effectiveOrchestration, toolMode: run.toolMode },
),
{
source: this.deps.runtimeSource ?? 'desktop',
abortSignal: abortController.signal,
},
)
: runLegacyProviderRetry(runner, continuation, {
source: this.deps.runtimeSource ?? 'desktop',
orchestration: run.effectiveOrchestration,
toolMode: run.toolMode,
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 {
this.clearInteractionRequestOwners(continuation.sessionId, run.turnId);
releaseExecutionAbort();
}
}
}
private async revalidateContinuationSafety(
continuation: RuntimeContinuation,
availableToolNames?: readonly string[],
): Promise<void> {
if (!this.deps.inspectContinuationSafety) {
throw new Error('Runtime continuation requires an authoritative safety inspector');
}
const observation = await this.deps.inspectContinuationSafety(continuation.sessionId);
assertContinuationSafetyUnchanged(
continuation,
availableToolNames ? { ...observation, availableToolNames } : observation,
);
}
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: DeliveryAckQueue<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 === 'user_question_request' || event.type === 'sandbox_boundary_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;
}
// The Session projection above now reads as aborted. The ledger has to say
// the same thing before this stop reports success: a Run left non-terminal
// here stays that way, because the stream that would have finalized it is
// exactly the one the stop could not wake.
//
// Without a Host interaction authority, Runtime owns terminal settlement.
// A Hosted Run's terminal fact belongs to the Host, which also parks
// provider-indeterminate Runs that a stop must not resolve on its behalf.
for (const target of stoppedRuns.values()) {
if (!this.deps.interactionAuthority) {
try {
await target.run?.settleStopTerminal();
} catch (error) {
// Leave the target unfinished so the operation stays pending and a
// retried stop settles it again, the same way a failed projection
// write is retried above.
failures.add(error);
continue;
}
}
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 respondToSandboxBoundary(
sessionId: string,
response: SandboxBoundaryResponse,
): Promise<void> {
const key = interactionOwnerKey(sessionId, response.requestId);
const owner = this.interactionRequestOwners.get(key);
if (owner?.request.type !== 'sandbox_boundary_request') {
throw new Error(`No pending sandbox boundary request ${response.requestId}`);
}
const active = this.backendGenerations.get(owner.generation);
if (
!active ||
active.sessionId !== sessionId ||
active.phase === 'terminated' ||
active.phase === 'failed'
) {
this.interactionRequestOwners.delete(key);
throw new Error(`Sandbox boundary request owner is unavailable: ${response.requestId}`);
}
await active.backend.respondToSandboxBoundary(response);
}
listActiveInteractions(sessionId: string): ActiveInteractionRequestEvent[] {
return [...this.interactionRequestOwners.values()]
.filter((owner) => owner.sessionId === sessionId)
.sort((left, right) => left.request.ts - right.request.ts)
.map((owner) => owner.request);
}
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);
}
runningTurnIds(sessionId: string): string[] {
const turnIds: string[] = [];
for (const active of this.backendGenerationsFor(sessionId)) {
for (const run of active.activeRuns.values()) {
if (!turnIds.includes(run.turnId)) turnIds.push(run.turnId);
}
}
return turnIds;
}
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.historyCompactCoordinator.clear(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',
);
}
/**
* Track every request a session can park on until its settlement ack lands,
* so a surface that was not mounted when the request streamed by can still
* read it back and render the prompt (#2072).
*/
private observeInteractionEvent(
sessionId: string,
backend: AgentBackend,
event: SessionEvent,
): void {
if (
event.type !== 'sandbox_boundary_request' &&
event.type !== 'user_question_request' &&
event.type !== 'sandbox_boundary_decision_ack' &&
event.type !== 'user_question_answer_ack'
) {
return;
}
const key = interactionOwnerKey(sessionId, event.requestId);
if (
event.type === 'sandbox_boundary_decision_ack' ||
event.type === 'user_question_answer_ack'
) {
this.interactionRequestOwners.delete(key);
return;
}
const generation = [...this.backendGenerations.values()].find(
(candidate) =>
candidate.sessionId === sessionId &&
candidate.backend === backend &&
candidate.phase !== 'terminated',
);
if (!generation) {
throw new RuntimeInteractionInvariantError(
`Interaction request ${event.requestId} has no active backend owner`,
);
}
const existing = this.interactionRequestOwners.get(key);
if (
existing &&
(existing.generation !== generation.generation || existing.turnId !== event.turnId)
) {
throw new RuntimeInteractionInvariantError(
`Interaction request ${event.requestId} has conflicting owners`,
);
}
this.interactionRequestOwners.set(key, {
sessionId,
turnId: event.turnId,
generation: generation.generation,
request: event,
});
}
private clearInteractionRequestOwners(sessionId: string, turnId: string): void {
for (const [key, owner] of this.interactionRequestOwners) {
if (owner.sessionId === sessionId && owner.turnId === turnId) {
this.interactionRequestOwners.delete(key);
}
}
}
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),
);
}
/**
* Builds the backend recorder hooks shared by `ensureActive` and
* `ensureChildActive`. The two paths are structurally identical except for
* which active-session map they resolve against — captured here by
* `resolveActive`, so each call site binds its own resolver. Callers retain
* the intentionally divergent fields (`allowMidTurnHistoryCompaction`,
* `shellRunContextSummary`, system prompt/tools source) at their own sites.
*/
private buildBackendRecorderHooks(input: {
resolveActive: () => BackendGeneration | undefined;
sessionId: string;
}): Pick<
BackendFactoryContext,
| 'recordRunTrace'
| 'recordProviderRequestCapture'
| 'recordProviderRequestAttempt'
| 'recordModelCallAttempt'
| 'recordRunComposition'
| 'loadHistoryCompactCheckpoint'
| 'recordHistoryCompactCheckpoint'
| 'loadTurnRuntimeEvents'
| 'recordActiveFullCompactBlock'
| 'recordSemanticCompactBlock'
> {
const { resolveActive, sessionId } = input;
const runFor = (turnId: string): AgentRun | undefined => {
const active = resolveActive();
const runId = active?.turnToRunId.get(turnId);
return runId ? active?.activeRuns.get(runId) : undefined;
};
return {
recordRunTrace: (event) => {
runFor(event.turnId)?.recordRunTrace(event);
},
...(this.deps.runStore
? {
recordProviderRequestCapture: (capture) => {
const run = runFor(capture.turnId);
if (!run)
return Promise.reject(new Error('No active AgentRun for provider request capture'));
return run.recordProviderRequestCapture(capture);
},
recordProviderRequestAttempt: (attempt) => {
runFor(attempt.turnId)?.recordProviderRequestAttempt(attempt);
},
// Resolved by runId rather than turnId: the canonical record names
// the run it belongs to, so it needs no turn-to-run indirection.
recordModelCallAttempt: (attempt) => {
const run = resolveActive()?.activeRuns.get(attempt.runId);
return run?.recordModelCallAttempt(attempt) ?? Promise.resolve();
},
recordRunComposition: (runId, snapshot) => {
const run = resolveActive()?.activeRuns.get(runId);
if (!run) {
return Promise.reject(new Error('No active AgentRun for Run Composition'));
}
return run.recordRunComposition(snapshot);
},
loadHistoryCompactCheckpoint: () => this.historyCompactCoordinator.load(sessionId),
recordHistoryCompactCheckpoint: (
checkpoint: HistoryCompactCheckpoint,
turnId: string,
) => this.historyCompactCoordinator.record(sessionId, checkpoint, runFor(turnId)),
}
: {}),
...(this.deps.runtimeEventStore
? {
loadTurnRuntimeEvents: (turnId: string) => {
const run = runFor(turnId);
if (!run)
return Promise.reject(new Error('No active AgentRun for turn runtime events'));
return run.loadTurnRuntimeEvents();
},
}
: {}),
recordActiveFullCompactBlock: (block) => {
runFor(block.turnId)?.recordActiveFullCompactBlock(block);
},
recordSemanticCompactBlock: (block) => {
runFor(block.turnId)?.recordSemanticCompactBlock(block);
},
};
}
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,
}
: {}),
...this.buildBackendRecorderHooks({
resolveActive: () => this.active.get(sessionId),
sessionId,
}),
allowMidTurnHistoryCompaction: Boolean(this.deps.runtimeEventStore),
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');
}
const snapshotDefinition = {
id: snapshot.agentId,
permissionMode: header.permissionMode,
tools: snapshot.toolNames,
};
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,
): 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,
...this.buildBackendRecorderHooks({
resolveActive: () => this.childActive.get(activeKey),
sessionId,
}),
// A child-only ledger cannot claim coverage of the parent session prefix.
allowMidTurnHistoryCompaction: false,
});
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[],
run: AgentRun,
execution: PendingExecutionClaim,
): Promise<BackendGeneration> {
const active = await this.ensureChildActive(
activeKey,
sessionId,
header,
systemPrompt,
tools,
execution,
);
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 requireRuntimeContinuationAuthority(
store: RuntimeEventStore,
): RuntimeContinuationAuthorityStore {
const candidate = store as Partial<RuntimeContinuationAuthorityStore>;
if (
candidate.continuationAuthorityCapability !== 'runtime_continuation_authority_v1' ||
typeof candidate.readImmutableRuntimeEvents !== 'function' ||
typeof candidate.readImmutableRuntimePrefix !== 'function' ||
typeof candidate.claimContinuation !== 'function' ||
typeof candidate.readContinuationClaimStateByBoundary !== 'function' ||
typeof candidate.listContinuationClaimsForRecovery !== 'function' ||
typeof candidate.commitContinuationStart !== 'function' ||
typeof candidate.commitContinuationRepairStart !== 'function'
) {
throw new Error('Runtime continuation requires SQLite continuation authority');
}
return candidate as RuntimeContinuationAuthorityStore;
}
async function revalidateContinuationBoundary(
store: RuntimeContinuationAuthorityStore,
continuation: RuntimeContinuation,
): Promise<RuntimeEvent[]> {
if (
!continuation.boundary ||
!continuation.providerReplayDigest ||
continuation.providerProjectionVersion !== PROVIDER_REPLAY_PROJECTION_VERSION
) {
throw new RuntimeContinuationRevalidationError(
'source_identity_changed',
'Runtime continuation is missing its versioned immutable boundary',
);
}
const prefixes: ImmutableRuntimePrefixV1[] = [];
const immediateSourceIndex = continuation.boundary.segments.length - 1;
for (const [index, segment] of continuation.boundary.segments.entries()) {
const prefix = await store.readImmutableRuntimePrefix({
sessionId: segment.identity.sessionId,
runId: segment.identity.runId,
// Ancestor segments are immutable lineage pins. The immediate source is
// different: execution must observe its latest durable head so H+1
// cannot be hidden by rereading only the already-planned prefix.
...(index === immediateSourceIndex ? {} : { upToEventSeq: segment.position.lastEventSeq }),
});
if (
!isDeepStrictEqual(prefix.identity, segment.identity) ||
!isDeepStrictEqual(prefix.position, segment.position) ||
prefix.prefixDigest !== segment.prefixDigest
) {
throw new RuntimeContinuationRevalidationError(
'source_identity_changed',
`Runtime continuation boundary changed for ${segment.identity.runId}`,
);
}
prefixes.push(prefix);
}
const replay = buildContinuationReplayPlan({
prefixes: prefixes as [ImmutableRuntimePrefixV1, ...ImmutableRuntimePrefixV1[]],
providerProjectionVersion: continuation.providerProjectionVersion,
});
if (
replay.kind !== 'replayable' ||
replay.plan.boundary.manifestDigest !== continuation.boundary.manifestDigest ||
replay.plan.providerReplayDigest !== continuation.providerReplayDigest ||
!isDeepStrictEqual(replay.plan.runtimeContext, continuation.runtimeContext)
) {
throw new RuntimeContinuationRevalidationError(
'source_replay_changed',
'Runtime continuation replay changed after planning',
);
}
return [...prefixes.at(-1)!.events];
}
function continuationClaimForExecution(
continuation: RuntimeContinuation,
claimedAt: number,
targetRunHeader: AgentRunHeader,
): ContinuationClaimV1 {
if (
!continuation.claimId ||
!continuation.boundary ||
!continuation.providerReplayDigest ||
continuation.providerProjectionVersion !== PROVIDER_REPLAY_PROJECTION_VERSION
) {
throw new RuntimeContinuationRevalidationError(
'source_identity_changed',
'Runtime continuation is missing its durable claim identity',
);
}
return {
protocol: 'continuation_claim_v1',
claimId: continuation.claimId,
boundaryDigest: continuation.boundary.manifestDigest,
boundary: continuation.boundary,
providerProjectionVersion: continuation.providerProjectionVersion,
providerReplayDigest: continuation.providerReplayDigest,
target: {
sessionId: continuation.sessionId,
invocationId: continuation.invocationId,
runId: continuation.runId,
turnId: continuation.turnId,
},
targetRunHeader,
claimedAt,
};
}
function continuationTargetRunHeaderForExecution(input: {
continuation: RuntimeContinuation;
sessionHeader: SessionHeader;
userInput: UserMessageInput;
workspaceIdentity: string;
effectiveOrchestration: EffectiveOrchestration;
effectiveToolMode: ToolMode;
claimedAt: number;
}): AgentRunHeader {
const {
continuation,
sessionHeader,
userInput,
effectiveOrchestration,
effectiveToolMode,
claimedAt,
} = input;
if (!continuation.claimId || !continuation.boundary) {
throw new RuntimeContinuationRevalidationError(
'source_identity_changed',
'Runtime continuation is missing its durable target-header identity',
);
}
const source = continuation.boundary.segments.at(-1)!;
return {
runId: continuation.runId,
invocationId: continuation.invocationId,
sessionId: continuation.sessionId,
turnId: continuation.turnId,
status: 'created',
backendKind: sessionHeader.backend,
llmConnectionSlug: sessionHeader.llmConnectionSlug,
modelId: sessionHeader.model,
cwd: sessionHeader.cwd,
workspaceIdentity: input.workspaceIdentity,
permissionMode: sessionHeader.permissionMode,
collaborationMode: sessionHeader.collaborationMode ?? 'agent',
orchestrationMode: effectiveOrchestration.mode,
orchestrationSource: effectiveOrchestration.source,
agentSwarmAuthorization: effectiveOrchestration.agentSwarmAuthorization,
toolMode: effectiveToolMode,
createdAt: claimedAt,
updatedAt: claimedAt,
...(userInput.parentRunId ? { parentRunId: userInput.parentRunId } : {}),
...(userInput.resumedFromRunId ? { resumedFromRunId: userInput.resumedFromRunId } : {}),
...(userInput.retriedFromRunId ? { retriedFromRunId: userInput.retriedFromRunId } : {}),
...(userInput.parentTurnId ? { parentTurnId: userInput.parentTurnId } : {}),
...(userInput.retriedFromTurnId ? { retriedFromTurnId: userInput.retriedFromTurnId } : {}),
...(userInput.regeneratedFromTurnId
? { regeneratedFromTurnId: userInput.regeneratedFromTurnId }
: {}),
...(userInput.branchOfTurnId ? { branchOfTurnId: userInput.branchOfTurnId } : {}),
...(userInput.parentSessionId ? { parentSessionId: userInput.parentSessionId } : {}),
...(userInput.agentId ? { agentId: userInput.agentId } : {}),
...(userInput.agentName ? { agentName: userInput.agentName } : {}),
continuationSource: {
protocol: 'continuation_source_v2',
claimId: continuation.claimId,
boundaryDigest: continuation.boundary.manifestDigest,
sourceInvocationId: source.identity.invocationId,
sourceRunId: source.identity.runId,
sourceTurnId: source.identity.turnId,
sourceRuntimeEventHighWater: source.position.lastEventSeq,
sourcePrefixDigest: source.prefixDigest,
replayManifestDigest: continuation.boundary.manifestDigest,
},
};
}
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',
);
}
if (continuation.boundary) {
// Composite immutable-prefix and provider replay equality were already
// revalidated by revalidateContinuationBoundary().
return;
}
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 plannedToolNames = [...new Set(snapshot.availableToolNames)].sort();
const currentToolNames = [...new Set(observation.availableToolNames)].sort();
if (!isDeepStrictEqual(plannedToolNames, currentToolNames)) {
throw new RuntimeContinuationRevalidationError(
'tool_catalog_changed',
`Runtime continuation tool catalog changed after planning: planned [${plannedToolNames.join(
', ',
)}], current [${currentToolNames.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',
);
}
}
}
function snapshotRuntimeContinuation(continuation: RuntimeContinuation): RuntimeContinuation {
return deepFreezeContinuationValue(structuredClone(continuation));
}
function deepFreezeContinuationValue<T>(value: T): T {
if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value;
for (const nested of Object.values(value as Record<string, unknown>)) {
deepFreezeContinuationValue(nested);
}
return Object.freeze(value);
}
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;
}
async abandonUnstartedContinuation(error: unknown): Promise<never> {
this.settleReservedExecution({ ok: false, error });
this.releaseMessage();
await this.callbacks.finalizeExecution(async () => undefined);
throw error;
}
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);
}
function effectiveToolModeForRun(run: AgentRunHeader): ToolMode {
return run.toolMode ?? DEFAULT_TOOL_MODE;
}
async function interactionResumeAllowed(
interactionRun: RuntimeInteractionRunBinding | undefined,
event: SessionEvent,
): Promise<boolean> {
if (
!interactionRun ||
(event.type !== 'user_question_answer_ack' && event.type !== 'sandbox_boundary_decision_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);
}
}
function interactionOwnerKey(sessionId: string, requestId: string): string {
return `${sessionId}\0${requestId}`;
}
export type { AgentRunLineage };