packages/runtime/src/runtime-kernel.ts

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import type { WorkHubActionReceipt } from '@maka/core/workhub-action-result';
import type { AgentRunStore } from '@maka/core/agent-run';
import { agentRunCompositionFromEvents } from '@maka/core/agent-run';
import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation';
import {
  decodeRuntimeBoundaryCursor,
  type ContinuationClaimV1,
  type ImmutableRuntimePrefixV1,
} from '@maka/core/runtime-boundary';
import {
  isTerminalRuntimeEvent,
  type RuntimeEvent,
  type RuntimeEventInvocationOpenedContent,
  type ToolBoundaryProtocol,
} from '@maka/core/runtime-event';
import type {
  RuntimeContinuationAuthorityStore,
  RuntimeEventStore,
} from '@maka/core/runtime-event-store';
import {
  type ActiveInteractionRequestEvent,
  type CompleteEvent,
  type SessionEvent,
  type TokenUsageEvent,
} from '@maka/core/events';
import type {
  SessionBlockedReason,
  SessionHeader,
  SessionHeaderPatch,
  SessionStatus,
} from '@maka/core/session';
import { isDeepStrictEqual } from 'node:util';
import type { 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 AgentRunHooks,
  type AgentRunLineage,
  type RuntimeContinuationFailpoint,
} from './agent-run.js';
import {
  createSessionEventMapMemory,
  isLiveBackendSessionEvent,
  mapSessionEventToRuntimeEvent,
  type RuntimeEventMapContext,
} from './session-event-runtime-mapper.js';
import { cloneAndFreezeRuntimeSnapshot } from './runtime-snapshot.js';
import type {
  AgentBackend,
  BackendSendInput,
  HostedInteractionBridge,
  RuntimeContinuationMetadata,
} from '@maka/core/backend-types';
import type { MakaTool } from './tool-runtime.js';
import type {
  BackendFactoryContext,
  BackendRegistry,
  CompactSessionInput,
  PreparedBackendActivation,
  ResolvedChildToolActivation,
  SessionStore,
  StopSessionInput,
} from './session-manager.js';
import type { TurnShellPlan } from './shell-detect.js';
import type { ShellRunProcessManager } from './shell-run-manager.js';
import { buildStatusPatch, normalizeStopSessionSource } from './session-projection-helpers.js';
import { buildToolsForAgentDefinition } from './agent-catalog.js';
import { loadLatestHistoryCompactCheckpointFromRunLedger } from './history-compact-ledger.js';
import { loadModelProjectionTransitionsFromRunLedger } from './model-projection-transition-ledger.js';
import type { ModelProjectionTransition } from '@maka/core/model-projection-transition';
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,
  digestProviderReplayAdmission,
  type ContinuationReplayAdmissionRoute,
} from './continuation-replay.js';
import {
  admitProviderReasoningReplayItems,
  buildRuntimeEventModelReplayPlan,
  compatibleProviderReasoningReplayEventIds,
  PROVIDER_REPLAY_PROJECTION_VERSION,
} from './model-history.js';
import {
  consumeRuntimeContinuationStartAdmissionProof,
  type RuntimeContinuationStartAdmissionProof,
} from './runtime-continuation-admission.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,
  isHostedInteractionRequestEvent,
  isHostedInteractionSettlementAckEvent,
  isShutdownCancelledInteractionAdmission,
  type RuntimeInteractionAuthority,
  type RuntimeInteractionRunBinding,
  type RuntimeInteractionRunClosureReason,
} from './interaction-authority.js';
import { DeliveryAckQueue, isDeliveryAckQueueClosed } from './delivery-ack-queue.js';
import { runtimeHandoffPause, type RuntimeHandoffIntent } from '@maka/core/runtime-handoff';
import { preserveHandoffOpening } from './runtime-resume.js';
import { runtimeInvocationRouteForHeader } from './runtime-invocation-route.js';
import type { AgentRunHandoffRequest } from './agent-run.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>;
  runCoordinationOperation(
    sessionId: string,
    input: UserMessageInput,
    options: TurnStartOptions,
    execute: () => Promise<WorkHubActionReceipt>,
  ): AsyncIterable<SessionEvent>;
  compactSession(sessionId: string, input?: CompactSessionInput): AsyncIterable<SessionEvent>;
  preflightContextCompaction(sessionId: string): Promise<void>;
  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>;
  /** Compatibility surface; durable message admission belongs to Runtime Host. */
  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;
  requestRunHandoff?(
    sessionId: string,
    runId: string,
    pause: RuntimeHandoffIntent,
    signal: AbortSignal,
  ): AgentRunHandoffRequest | undefined;
  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 | null;
  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>;
  /** Original logical owner may have accepted Stop while its sealed attempt retired. */
  stopBeforeDispatch?: () => StopSessionInput | undefined;
}

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 type BackendActivationBoundary = <T>(operation: () => Promise<T> | T) => Promise<T>;

interface ChildToolActivation {
  readonly tools: readonly MakaTool[];
  readonly shell?: TurnShellPlan;
}

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[];
  resolveChildTools?: (sessionId: string) => Promise<ResolvedChildToolActivation>;
  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;
  phase: 'active' | 'stopping' | 'disposing' | 'failed' | 'terminated';
  backend: AgentBackend;
  providerStateIdentity?: `sha256:${string}`;
  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;
  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;
  hostOperation?: true;
  backendHeaderSnapshot?: { invalidated: boolean };
  backendPreparation?: PreparedBackendActivation;
  stopIntent?: SessionStopIntent;
  finalization?: ExecutionClaimOutcome;
}

type BackendDisposalOutcome = { ok: true } | { ok: false; error: unknown };

interface BackendInvalidationState {
  readonly outcome: Promise<BackendDisposalOutcome>;
  readonly activations: Set<PendingExecutionClaim>;
  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 backendGenerations = new Map<number, BackendGeneration>();
  private readonly backendActivationBuilds = new Map<string, Promise<BackendGeneration>>();
  private readonly backendActivations = new Set<PendingExecutionClaim>();
  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 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>(
    execution: PendingExecutionClaim,
    operation: () => Promise<T> | T,
  ): Promise<T> {
    const activate = async () => {
      this.backendActivations.add(execution);
      try {
        return await operation();
      } finally {
        this.backendActivations.delete(execution);
        this.backendInvalidations.get(execution.sessionId)?.activations.delete(execution);
        await this.flushBackendInvalidation(execution.sessionId);
      }
    };
    return await (this.deps.runBackendActivation?.(activate) ?? activate());
  }

  private readBackendHeader(execution: PendingExecutionClaim): Promise<SessionHeader> {
    // Register before the read: even the store may suspend after taking its
    // snapshot. This covers all preflight work before the policy activation gate.
    execution.backendHeaderSnapshot = { invalidated: false };
    return this.deps.store.readHeader(execution.sessionId);
  }

  private invalidateBackendHeaderSnapshots(sessionId: string): void {
    for (const execution of this.executionClaims.get(sessionId) ?? []) {
      if (execution.backendHeaderSnapshot) execution.backendHeaderSnapshot.invalidated = true;
    }
  }

  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, execution.stopIntent.input.workHubActionId);
    }
  }

  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> {
    assertNoRemovedChildAgentRunLineage(input);
    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.readBackendHeader(execution);
      let workspaceIdentity: string | undefined;
      if (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,
        runStore: this.deps.runStore,
        runtimeEventStore: this.deps.runtimeEventStore,
        ...(this.deps.toolBoundaryProtocol
          ? { toolBoundaryProtocol: this.deps.toolBoundaryProtocol }
          : {}),
        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),
          ...this.messageProjectionHook(),
        },
      });
      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, run, execution, {
        steering: true,
        onRunStarted: options.onRunStarted,
        initialHeader: 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.readBackendHeader(execution);
    const sessionRuns = await this.deps.runtimeEventStore.listSessionInvocations(
      continuation.sessionId,
    );
    const sourceRun = sessionRuns.find((run) => run.runId === continuation.sourceRunId);
    if (!sourceRun) {
      throw new RuntimeContinuationRevalidationError(
        'source_identity_changed',
        'Runtime continuation source run no longer exists',
      );
    }
    const targetProviderStateIdentity = (
      await this.deps.backends.prepare(header.backend, {
        sessionId: continuation.sessionId,
        workspaceRoot: header.workspaceRoot,
        header,
        abortSignal: execution.abortController.signal,
      })
    ).providerStateIdentity;
    const admissionRoute: ContinuationReplayAdmissionRoute = {
      invocations: sessionRuns,
      targetProviderStateIdentity,
      targetModelId: header.model,
    };
    const sourceEvents = await revalidateContinuationBoundary(
      continuationAuthority,
      continuation,
      admissionRoute,
    );
    assertContinuationSourceUnchanged(continuation, sourceRun, sourceEvents);
    await this.revalidateContinuationSafety(continuation);
    if (!this.deps.store.hasExplicitSandboxBoundaryDenial) {
      throw new Error('Continuation requires authoritative sandbox boundary decision lookup');
    }
    const inheritedSandboxBoundaryDenied = await this.deps.store.hasExplicitSandboxBoundaryDenial(
      continuation.boundary!.segments.map((segment) => segment.identity),
    );

    const handoffSourceComposition =
      continuation.handoffRootRunId !== undefined
        ? agentRunCompositionFromEvents(
            await this.deps.runStore.readEvents(continuation.sessionId, sourceRun.runId),
          )
        : undefined;
    if (continuation.handoffRootRunId !== undefined && !handoffSourceComposition) {
      throw new RuntimeContinuationRevalidationError(
        'source_identity_changed',
        'Cooperative handoff source has no durable Run Composition',
      );
    }

    const userInput: UserMessageInput = {
      turnId: continuation.turnId,
      text: '',
      ...(continuation.handoffRootRunId === undefined
        ? { parentTurnId: continuation.sourceTurnId }
        : {}),
    };
    const effectiveOrchestration = effectiveOrchestrationForRun(sourceRun, header);
    const effectiveToolMode = effectiveToolModeForRun(sourceRun);
    const claimedAt = this.deps.now();
    const targetOpening = continuationTargetOpeningForExecution({
      sourceOpening: sourceRun.opening,
      continuation,
      sessionHeader: header,
      userInput,
      workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
      effectiveOrchestration,
      effectiveToolMode,
      targetProviderStateIdentity,
    });
    const claim = continuationClaimForExecution(continuation, claimedAt, targetOpening);
    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 existingClaim = sessionRuns.find((candidate) => {
      const source = candidate.opening.source;
      return (
        source.kind !== 'fresh' &&
        source.sourceRunId === continuation.sourceRunId &&
        source.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((candidate) => candidate.runId === continuation.runId);
    if (existingTarget) {
      throw new RuntimeContinuationRevalidationError(
        'target_run_conflict',
        'Runtime continuation target run already exists',
      );
    }

    const continuationToolBoundaryProtocol = this.deps.toolBoundaryProtocol;
    const run = new AgentRun({
      sessionId: continuation.sessionId,
      header,
      userInput,
      runLineage:
        continuation.handoffRootRunId !== undefined
          ? sourceRun.opening.lineage
          : { parentRunId: continuation.sourceRunId },
      runId: continuation.runId,
      invocationId: continuation.invocationId,
      runStore: this.deps.runStore,
      runtimeEventStore: this.deps.runtimeEventStore,
      ...(continuationToolBoundaryProtocol
        ? { toolBoundaryProtocol: continuationToolBoundaryProtocol }
        : {}),
      newId: this.deps.newId,
      now: this.deps.now,
      workspaceIdentity: continuation.safetySnapshot.workspaceIdentity,
      effectiveOrchestration,
      // Round-tripped through the claim on purpose: openInvocation compares it
      // against the opening it computes, so every continuation proves the claim
      // still authorises the run about to execute.
      claimedOpening: claim.targetOpening,
      ...(continuation.handoffRootRunId !== undefined
        ? { handoffSourceOpening: sourceRun.opening, handoffSourceComposition }
        : {}),
      claimedOpenedAt: claimedAt,
      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',
            modelVisibility: 'hidden',
            // The start event is event 1 of the target invocation, so it is
            // also where that invocation's opening fact lives.
            content: claim.targetOpening,
            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),
        ...this.messageProjectionHook(),
      },
    });

    this.attachExecutionClaim(execution, run);
    yield* this.runAgentContinuation(
      continuation,
      admissionRoute,
      run,
      execution,
      {
        sessionId: continuation.sessionId,
        turnId: continuation.turnId,
        runId: continuation.runId,
      },
      options,
      () => this.revalidateContinuationSafety(continuation),
      inheritedSandboxBoundaryDenied,
    );
  }

  /** Host coordination uses the same Run owner and terminal authority without a provider send. */
  async *runCoordinationOperation(
    sessionId: string,
    input: UserMessageInput,
    options: TurnStartOptions,
    execute: () => Promise<WorkHubActionReceipt>,
  ): AsyncIterable<SessionEvent> {
    const execution = this.takeExecutionClaim(sessionId);
    execution.hostOperation = true;
    try {
      await this.enterExecutionClaim(execution);
      const header = await this.deps.store.readHeader(sessionId);
      const run = new AgentRun({
        sessionId,
        header,
        userInput: input,
        runId: options.runId,
        userMessageId: options.userMessageId,
        durability: 'required',
        runStore: this.deps.runStore,
        runtimeEventStore: this.deps.runtimeEventStore,
        newId: this.deps.newId,
        now: this.deps.now,
        effectiveOrchestration: resolveEffectiveOrchestration('default', undefined),
        hooks: {
          reserveRun: async (id, nextHeader, activeRun) => {
            const active = await this.reserveParentRun(id, nextHeader, activeRun, execution);
            this.reserveExecutionClaim(execution, active, activeRun);
            return active;
          },
          unregisterRun: (active, activeRun) => this.unregisterParentRun(active, activeRun),
          updateHeader: (id, patch) => this.updateHeader(id, patch),
          updateStatus: (id, status, reason, ts) => this.updateStatus(id, status, reason, ts),
          ...this.messageProjectionHook(),
        },
      });
      this.attachExecutionClaim(execution, run);
      const owners = this.createRunOwnerScope(run, execution);
      try {
        owners.bindMessage(this.deps.messageAuthority, {
          sessionId,
          turnId: input.turnId,
          runId: run.runId,
        });
        // Keep the execution claim attached until finalization. Stop/drain can
        // therefore cancel and await this Run without a provider generation.
        await run.beginCoordination();
        await options.onRunStarted?.(run.runId, header);
      } catch (error) {
        await this.finalizeFailedRunStart(owners, run, execution, error);
        return;
      }
      try {
        if (run.isStopped()) return;
        const executed = await execute();
        const receipt: WorkHubActionReceipt = {
          ...executed,
          result:
            executed.result.disposition === 'clarify'
              ? { ...executed.result, coordinationTurnId: input.turnId }
              : executed.result,
        };
        const receiptEvent: RuntimeEvent = {
          id: this.deps.newId(),
          sessionId,
          turnId: input.turnId,
          runId: run.runId,
          invocationId: run.runId,
          ts: this.deps.now(),
          partial: false,
          role: 'system',
          author: 'host',
          modelVisibility: 'hidden',
          actions: { coordination: receipt },
        };
        await run.recordRuntimeEvents([receiptEvent], { requireDurableWrite: true });
        if (run.isStopped()) return;
        const complete: CompleteEvent = {
          type: 'complete',
          id: this.deps.newId(),
          turnId: input.turnId,
          ts: this.deps.now(),
          stopReason: 'end_turn',
        };
        await run.acceptMappedEvent(
          complete,
          mapSessionEventToRuntimeEvent(
            complete,
            this.runtimeEventMapContext({
              sessionId,
              invocationId: run.runId,
              runId: run.runId,
              turnId: input.turnId,
            }),
          ),
          { requireTerminalWrite: true },
        );
        yield complete;
      } 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(`Coordination cleanup failed for ${run.runId}`);
      }
    } finally {
      this.releaseExecutionClaim(execution);
      await this.flushBackendInvalidation(sessionId);
    }
  }

  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.readBackendHeader(execution);
      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 (!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.readBackendHeader(execution);
    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 } : {}),
      runStore: this.deps.runStore,
      runtimeEventStore: this.deps.runtimeEventStore,
      ...(this.deps.toolBoundaryProtocol
        ? { toolBoundaryProtocol: this.deps.toolBoundaryProtocol }
        : {}),
      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),
        ...this.messageProjectionHook(),
      },
    });

    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(execution, async () => {
        run.bindProviderStateIdentity(
          await this.prepareBackendForExecution(sessionId, header, execution),
        );
        return await run.beginOperation();
      });
      await input.hostedRoot?.onRunStarted?.();
      this.settleReservedExecutionClaim(execution, run, { ok: true });
    } catch (error) {
      await this.finalizeFailedRunStart(owners, run, execution, error);
      return;
    }

    let notedTerminal = false;
    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,
        runtimeContextInvocations: begin.runtimeContextInvocations,
      });
      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',
        contextCompactionOutcome: result.outcome,
      };
      const eventContext = this.runtimeEventMapContext({
        sessionId,
        invocationId: run.runId,
        runId: run.runId,
        turnId: run.turnId,
      });
      // Ahead of the usage row, because the ledger seals on its terminal fact:
      // a note queued behind one this compaction may already own would be
      // refused, and the reader would never learn the summary was skipped.
      if (result.outcome.kind === 'failed') {
        await run.recordSystemNote('context_compaction_failed_open').catch(() => {});
        notedTerminal = true;
      } else if (result.outcome.kind === 'compacted') {
        // Explicit compaction runs on its own turn and never enters the
        // send-flow note block, so the "compacted" note is written here. The
        // next user send passively replays this standalone checkpoint, which
        // `shouldAppendContextCompactedNote` suppresses, so there is no
        // duplicate.
        await run.recordSystemNote('context_compacted').catch(() => {});
        notedTerminal = true;
      }
      await run.acceptMappedEvent(
        tokenUsageEvent,
        mapSessionEventToRuntimeEvent(tokenUsageEvent, eventContext),
        { requireTerminalWrite: true },
      );
      if (run.isStopped()) return;
      yield tokenUsageEvent;
      if (run.isStopped()) return;
      await run.acceptMappedEvent(
        completeEvent,
        mapSessionEventToRuntimeEvent(completeEvent, eventContext),
        { requireTerminalWrite: true },
      );
      if (run.isStopped()) return;
      yield completeEvent;
    } catch (error) {
      // A thrown compaction still owns a fail-open note — but not when the throw
      // is a stop, which must leave no row. The note goes ahead of the failure
      // because the ledger seals on its terminal fact; the internal compaction
      // Turn has no user timeline for a failure banner.
      if (!notedTerminal && !run.isStopped()) {
        await run.recordSystemNote('context_compaction_failed_open').catch(() => {});
      }
      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(execution, () =>
      this.ensureActive(sessionId, header, execution),
    );
    if (!active.backend.compactHistory) {
      throw new RuntimeContextCompactError(
        'operation_unavailable',
        `Backend ${header.backend} does not support runtime compaction`,
      );
    }
    return active;
  }

  private async *runAgentTurn(
    sessionId: string,
    run: AgentRun,
    execution: PendingExecutionClaim,
    options: {
      steering?: boolean;
      onRunStarted?: (runId: string, initialHeader: SessionHeader) => void | Promise<void>;
      initialHeader?: SessionHeader;
      prepareBackendActivation?: () => Promise<void>;
    } = {},
  ): AsyncIterable<SessionEvent> {
    const { steering = false, onRunStarted, initialHeader, prepareBackendActivation } = options;
    const { abortController, release: releaseExecutionAbort } =
      this.inheritExecutionAbort(execution);
    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(execution, async () => {
        await prepareBackendActivation?.();
        run.bindProviderStateIdentity(
          await this.prepareBackendForExecution(sessionId, run.headerSnapshot(), execution),
        );
        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;
    }

    yield* this.streamAgentRun({
      sessionId,
      run,
      owners,
      backend: begin.backend,
      abortController,
      releaseExecutionAbort,
      requireTerminalWrite: Boolean(this.deps.runtimeEventStore),
      invocationId: begin.initialRuntimeEvent.invocationId,
      backendInput: {
        invocationId: begin.initialRuntimeEvent.invocationId,
        runId: run.runId,
        ...begin.backendInput,
        headAnchorRuntimeEvent: begin.initialRuntimeEvent,
        handoffBoundary: (signal, remainingSteps) =>
          run.reachHandoffBoundary(signal, remainingSteps),
        ...runtimeSteeringInput(owners.messageOwner),
      },
    });
  }

  private async *runAgentContinuation(
    continuation: RuntimeContinuation,
    admissionRoute: ContinuationReplayAdmissionRoute,
    run: AgentRun,
    execution: PendingExecutionClaim,
    messageOwner?: RuntimeMessageRunIdentity,
    options: ResumeContinuationOptions = {},
    revalidateSafety?: () => Promise<void>,
    inheritedSandboxBoundaryDenied = false,
  ): AsyncIterable<SessionEvent> {
    const { abortController, release: releaseExecutionAbort } =
      this.inheritExecutionAbort(execution);
    const owners = this.createRunOwnerScope(run, execution);
    let begin: Awaited<ReturnType<AgentRun['beginContinuation']>>;
    try {
      if (messageOwner) owners.bindMessage(this.deps.messageAuthority, messageOwner);
      begin = await this.runBackendActivation(execution, async () => {
        if (!revalidateSafety) {
          throw new Error('Durable continuation omitted final safety revalidation');
        }
        await revalidateSafety();
        run.bindProviderStateIdentity(
          await this.prepareBackendForExecution(
            continuation.sessionId,
            run.headerSnapshot(),
            execution,
          ),
        );
        const started = await run.beginContinuation(continuation);
        await owners.bindInteraction(this.deps.interactionAuthority, {
          sessionId: continuation.sessionId,
          turnId: run.turnId,
          runId: run.runId,
        });
        return started;
      });
      await options.onRunStarted?.();
      const stop = options.stopBeforeDispatch?.();
      if (stop) {
        run.stop(stop.source, stop.workHubActionId);
        releaseExecutionAbort();
        await owners.finalize();
        owners.releaseMessage();
        return;
      }
    } catch (error) {
      releaseExecutionAbort();
      if (error instanceof ContinuationStartCommitError) {
        await owners.abandonUnstartedContinuation(error);
        return;
      }
      await this.finalizeFailedRunStart(owners, run, execution, error);
      return;
    }

    let continuationMetadata: RuntimeContinuationMetadata;
    try {
      continuationMetadata = consumeAdmittedRuntimeContinuation({
        continuation,
        admissionRoute,
        startAdmission:
          'continuationStartAdmission' in begin
            ? begin.continuationStartAdmission
            : (() => {
                throw new Error('Durable continuation is missing its start admission');
              })(),
        ...(run.toolBoundaryProtocol ? { toolBoundaryProtocol: run.toolBoundaryProtocol } : {}),
      });
      continuationMetadata.sandboxBoundaryDenied = inheritedSandboxBoundaryDenied;
    } catch (error) {
      releaseExecutionAbort();
      await this.finalizeFailedRunStart(owners, run, execution, error);
      return;
    }
    yield* this.streamAgentRun({
      sessionId: continuation.sessionId,
      run,
      owners,
      backend: begin.backend,
      abortController,
      releaseExecutionAbort,
      requireTerminalWrite: true,
      recordUnobservedStreamFailure: true,
      invocationId: continuation.invocationId,
      ...(continuation.handoffRootRunId !== undefined
        ? {
            prepareBeforeDispatch: async () => {
              this.assertRunCanDispatch(run, begin.backend);
              if (!begin.backend.prepareRunComposition) {
                throw new Error(
                  'Backend does not support cooperative handoff composition preparation',
                );
              }
              await begin.backend.prepareRunComposition({ runId: run.runId, turnId: run.turnId });
              run.assertRunCompositionCommitted();
            },
          }
        : {}),
      backendInput: {
        invocationId: continuation.invocationId,
        runId: continuation.runId,
        turnId: continuation.turnId,
        orchestration: run.effectiveOrchestration,
        toolMode: run.toolMode,
        text: '',
        runtimeContext: continuation.runtimeContext,
        runtimeContextInvocations: admissionRoute.invocations,
        continuation: continuationMetadata,
        ...runtimeSteeringInput(owners.messageOwner),
        ...(continuation.handoffRootRunId !== undefined
          ? {
              maxSteps: continuation.handoffRemainingSteps,
              // The original user anchor remains in the authenticated replay;
              // a physical successor must not invent another user message.
              headAnchorRuntimeEvent: continuation.runtimeContext.find(
                (event) =>
                  event.runId === continuation.handoffRootRunId &&
                  event.turnId === continuation.turnId &&
                  event.role === 'user' &&
                  (event.author === 'user' || event.author === 'host') &&
                  event.content?.kind === 'text' &&
                  event.content.steering !== true,
              ),
            }
          : {}),
        handoffBoundary: (signal, remainingSteps) =>
          run.reachHandoffBoundary(signal, remainingSteps),
      },
    });
  }

  private async *streamAgentRun(input: {
    sessionId: string;
    invocationId: string;
    run: AgentRun;
    owners: RuntimeRunOwnerScope;
    backend: AgentBackend;
    backendInput: BackendSendInput;
    abortController: AbortController;
    releaseExecutionAbort: () => void;
    requireTerminalWrite: boolean;
    prepareBeforeDispatch?: () => Promise<void>;
    recordUnobservedStreamFailure?: boolean;
  }): AsyncIterable<SessionEvent> {
    const sessionEvents = new DeliveryAckQueue<SessionEvent>();
    const interactionRun = input.owners.interactionRun;
    const stopBackend = this.stopBackendFor(input.backend);
    let flowDone = false;
    let streamFailure: unknown;
    if (input.run.isStopped()) input.abortController.abort();
    const streamResult = this.runBackendEventStream({
      backend: input.backend,
      stopBackend,
      beforeDispatch: () => this.assertRunCanDispatch(input.run, input.backend),
      ...(input.prepareBeforeDispatch
        ? { prepareBeforeDispatch: input.prepareBeforeDispatch }
        : {}),
      hasCommittedHandoff: () => input.run.hasCommittedHandoff(),
      ...(interactionRun ? { hostedInteraction: interactionRun } : {}),
      abortSignal: input.abortController.signal,
      eventContext: this.runtimeEventMapContext({
        sessionId: input.sessionId,
        invocationId: input.invocationId,
        runId: input.run.runId,
        turnId: input.run.turnId,
      }),
      backendInput: input.backendInput,
      onSessionEvent: async (sessionEvent, runtimeEvent) => {
        this.assertInteractionPublication(interactionRun, sessionEvent);
        await input.run.acceptMappedEvent(sessionEvent, runtimeEvent, {
          requireTerminalWrite: input.requireTerminalWrite,
          allowInteractionResume: await interactionResumeAllowed(interactionRun, sessionEvent),
        });
        this.observeInteractionEvent(input.sessionId, input.backend, sessionEvent);
        await sessionEvents.push(sessionEvent);
      },
      onError: async (error) => {
        if (!isDeliveryAckQueueClosed(error)) {
          await input.run.recordFailure(error);
          sessionEvents.fail(error);
        }
      },
      onFinally: async () => {
        flowDone = true;
        try {
          await input.owners.finalize();
          // Release Runtime access before closing the event stream. Embedded
          // queues publish their final steering projection here; hosted owners
          // are only sealed, then the Host performs the handoff under its
          // Session admission gate.
          input.owners.releaseMessage();
          sessionEvents.close();
        } catch (error) {
          sessionEvents.fail(error);
          throw error;
        }
      },
    }).then(
      async (result) => {
        if (!flowDone) {
          try {
            flowDone = true;
            await input.owners.finalize();
            input.owners.releaseMessage();
            sessionEvents.close();
          } catch (error) {
            streamFailure = error;
            sessionEvents.fail(error);
            throw error;
          }
        }
        return result;
      },
      (error) => {
        streamFailure = error;
        sessionEvents.fail(error);
        throw error;
      },
    );

    try {
      for await (const event of sessionEvents) yield event;
      await streamResult;
    } finally {
      try {
        await this.cleanupRunExecution({
          run: input.run,
          stopBackend,
          flowDone,
          abortController: input.abortController,
          sessionEvents,
          streamResult,
          interactionRun,
          ...(input.recordUnobservedStreamFailure && streamFailure !== undefined
            ? { streamFailure }
            : {}),
          finalizeRun: () => input.owners.finalize(),
          releaseOwner: () => input.owners.releaseMessage(),
        });
      } finally {
        this.clearInteractionRequestOwners(input.sessionId, input.run.turnId);
        input.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> {
    // A draining authority refused the start because everything is stopping, not
    // because this run went wrong, so the run ends cancelled rather than failed.
    if (isShutdownCancelledInteractionAdmission(error)) run.stop(undefined);
    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;
    stopBackend: AgentBackend['stop'];
    flowDone: boolean;
    abortController: AbortController;
    sessionEvents: DeliveryAckQueue<SessionEvent>;
    streamResult: Promise<void>;
    interactionRun: RuntimeInteractionRunBinding | undefined;
    streamFailure?: 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.stopBackend('user_stop');
      input.abortController.abort();
      input.sessionEvents.close();
      await Promise.all([
        failures.capture(() => interactionClose),
        failures.capture(() => backendStop),
      ]);
      if (input.streamFailure !== undefined) {
        await failures.capture(() => input.run.recordFailure(input.streamFailure));
      }
    }

    await input.streamResult.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 && isHostedInteractionRequestEvent(event)) {
      binding.assertPendingAdmission(event);
    }
  }

  private runtimeEventMapContext(input: {
    sessionId: string;
    invocationId: string;
    runId: string;
    turnId: string;
  }): RuntimeEventMapContext {
    return {
      sessionId: input.sessionId,
      invocationId: input.invocationId,
      runId: input.runId,
      turnId: input.turnId,
      now: this.deps.now,
    };
  }

  private async runBackendEventStream(input: {
    backend: AgentBackend;
    stopBackend: AgentBackend['stop'];
    backendInput: BackendSendInput;
    hostedInteraction?: HostedInteractionBridge;
    abortSignal: AbortSignal;
    eventContext: RuntimeEventMapContext;
    prepareBeforeDispatch?: () => Promise<void>;
    beforeDispatch: () => void;
    hasCommittedHandoff?: () => boolean;
    onSessionEvent: (sessionEvent: SessionEvent, runtimeEvent: RuntimeEvent) => Promise<void>;
    onError: (error: unknown) => Promise<void>;
    onFinally: () => Promise<void>;
  }): Promise<void> {
    const backendInput = cloneAndFreezeRuntimeSnapshot(input.backendInput);
    if (input.eventContext.sessionId !== input.backend.sessionId) {
      throw new Error(
        `RuntimeKernel backend session mismatch: ${input.eventContext.sessionId} != ${input.backend.sessionId}`,
      );
    }
    if (input.abortSignal.aborted) {
      await input.onFinally();
      return;
    }

    const onAbort = (): void => {
      void input.stopBackend('user_stop').catch(() => {});
    };
    input.abortSignal.addEventListener('abort', onAbort, { once: true });

    const memory = createSessionEventMapMemory();
    let terminalSeen = false;
    let terminalAccepted = false;
    let errorSeen = false;
    try {
      if (input.prepareBeforeDispatch) await input.prepareBeforeDispatch();
      // Keep the final stop/admission check synchronous with backend.send.
      input.beforeDispatch();
      for await (const sessionEvent of input.backend.send({
        ...backendInput,
        ...(input.hostedInteraction ? { hostedInteraction: input.hostedInteraction } : {}),
      })) {
        if (terminalSeen || !isLiveBackendSessionEvent(sessionEvent)) continue;
        const runtimeEvent = mapSessionEventToRuntimeEvent(
          sessionEvent,
          input.eventContext,
          memory,
        );
        if (sessionEvent.type === 'error') errorSeen = true;
        terminalSeen = isTerminalRuntimeEvent(runtimeEvent);
        await input.onSessionEvent(sessionEvent, runtimeEvent);
        if (terminalSeen) terminalAccepted = true;
      }
      if (!terminalSeen && !input.hasCommittedHandoff?.()) {
        for (const sessionEvent of this.missingTerminalSessionEvents(
          input.eventContext.turnId,
          !errorSeen,
        )) {
          const runtimeEvent = mapSessionEventToRuntimeEvent(
            sessionEvent,
            input.eventContext,
            memory,
          );
          await input.onSessionEvent(sessionEvent, runtimeEvent);
          if (isTerminalRuntimeEvent(runtimeEvent)) terminalSeen = true;
        }
      }
    } catch (error) {
      if (terminalAccepted) return;
      await input.onError(error);
      throw error;
    } finally {
      input.abortSignal.removeEventListener('abort', onAbort);
      await input.onFinally();
    }
  }

  private missingTerminalSessionEvents(turnId: string, includeError: boolean): SessionEvent[] {
    const ts = this.deps.now();
    return [
      ...(includeError
        ? [
            {
              type: 'error' as const,
              id: this.deps.newId(),
              turnId,
              ts,
              recoverable: false,
              code: 'missing_terminal_event',
              reason: 'missing_terminal_event',
              message: 'backend exhausted without a terminal RuntimeEvent',
            },
          ]
        : []),
      {
        type: 'complete',
        id: this.deps.newId(),
        turnId,
        ts,
        stopReason: 'error',
      },
    ];
  }

  stopSession(sessionId: string, input: StopSessionInput = {}): Promise<void> {
    normalizeStopSessionSource(input.source, input.workHubActionId);
    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, input.workHubActionId);
    }
    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> {
    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, input.workHubActionId);
    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);

    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,
      });
    }
    return operation;
  }

  private buildStopOperation(input: StopSessionInput): StopOperation {
    const abortSource = normalizeStopSessionSource(input.source, input.workHubActionId);
    const ts = this.deps.now();
    return {
      abortSource,
      ts,
      statusProjected: 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;
    }
    // The ledger has to say this turn was aborted before the 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.
    // Nothing else records the abort — the transcript reads it back off this
    // terminal fact.
    //
    // 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.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}`);
  }

  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)),
    );
  }

  private activeRunsFor(sessionId: string): AgentRun[] {
    const runs = new Set<AgentRun>();
    for (const active of this.backendGenerationsFor(sessionId)) {
      for (const run of active.activeRuns.values()) runs.add(run);
    }
    for (const claim of this.executionClaims.get(sessionId) ?? []) {
      if (claim.hostOperation && claim.run) runs.add(claim.run);
    }
    return [...runs];
  }

  hasActiveRuns(sessionId: string): boolean {
    return this.activeRunsFor(sessionId).length > 0;
  }

  runningTurnIds(sessionId: string): string[] {
    return [...new Set(this.activeRunsFor(sessionId).map((run) => run.turnId))];
  }

  hasActiveRun(sessionId: string, runId: string, turnId?: string): boolean {
    return this.activeRunsFor(sessionId).some(
      (run) => run.runId === runId && (turnId === undefined || run.turnId === turnId),
    );
  }

  requestRunHandoff(
    sessionId: string,
    runId: string,
    pause: RuntimeHandoffIntent,
    signal: AbortSignal,
  ): AgentRunHandoffRequest | undefined {
    if (!this.deps.inspectContinuationSafety) return undefined;
    for (const active of this.backendGenerationsFor(sessionId)) {
      const run = active.activeRuns.get(runId);
      if (run) return run.requestHandoff(pause, signal);
    }
    return undefined;
  }

  updateCachedHeader(sessionId: string, header: SessionHeader): void {
    const active = this.active.get(sessionId);
    if (active) active.cachedHeader = header;
  }

  async invalidateBackend(sessionId: string): Promise<void> {
    this.invalidateBackendHeaderSnapshots(sessionId);
    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, claims] of this.executionClaims) {
      if ([...claims].some((execution) => execution.backendHeaderSnapshot))
        sessionIds.add(sessionId);
    }
    for (const execution of this.backendActivations) sessionIds.add(execution.sessionId);
    for (const sessionId of this.backendInvalidations.keys()) sessionIds.add(sessionId);
    await Promise.all(
      [...sessionIds].map(async (sessionId) => {
        this.invalidateBackendHeaderSnapshots(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.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);
  }

  private stopOperationReferences(active: BackendGeneration): boolean {
    return [...this.stopOperations.values()].some((operation) =>
      [...operation.targets.values()].some((target) => target.active === active),
    );
  }

  /** Every run this Session has opened, enumerated from the event spine. */
  private async sessionRunIds(sessionId: string): Promise<string[]> {
    const store = this.deps.runtimeEventStore;
    if (!store) return [];
    return (await store.listSessionInvocations(sessionId)).map((invocation) => invocation.runId);
  }

  private buildBackendRecorderHooks(input: {
    sessionId: string;
  }): Pick<
    BackendFactoryContext,
    | 'recordRunTrace'
    | 'recordSystemNote'
    | 'recordModelCallAttempt'
    | 'recordRunComposition'
    | 'recordRequestComposition'
    | 'loadHistoryCompactCheckpoint'
    | 'recordHistoryCompactCheckpoint'
    | 'loadModelProjectionTransitions'
    | 'recordModelProjectionTransition'
    | 'loadTurnRuntimeEvents'
  > {
    const { sessionId } = input;
    const runFor = (turnId: string): AgentRun | undefined => {
      const active = this.active.get(sessionId);
      const runId = active?.turnToRunId.get(turnId);
      return runId ? active?.activeRuns.get(runId) : undefined;
    };
    return {
      recordRunTrace: (event) => {
        runFor(event.turnId)?.recordRunTrace(event);
      },
      recordSystemNote: (kind, turnId, data) =>
        runFor(turnId)?.recordSystemNote(kind, data) ?? Promise.resolve(),
      ...(this.deps.runStore
        ? {
            // Resolved by runId rather than turnId: the canonical record names
            // the run it belongs to, so it needs no turn-to-run indirection.
            recordModelCallAttempt: (commit) => {
              const run = this.active.get(sessionId)?.activeRuns.get(commit.attempt.runId);
              return run?.recordModelCallAttempt(commit) ?? Promise.resolve();
            },
            recordRunComposition: (runId, snapshot) => {
              const run = this.active.get(sessionId)?.activeRuns.get(runId);
              if (!run) {
                return Promise.reject(new Error('No active AgentRun for Run Composition'));
              }
              return run.recordRunComposition(snapshot);
            },
            recordRequestComposition: (runId, snapshot) => {
              const run = this.active.get(sessionId)?.activeRuns.get(runId);
              if (!run) {
                return Promise.reject(new Error('No active AgentRun for Request Composition'));
              }
              return run.recordRequestComposition(snapshot);
            },
            loadHistoryCompactCheckpoint: () => this.historyCompactCoordinator.load(sessionId),
            recordHistoryCompactCheckpoint: (
              checkpoint: HistoryCompactCheckpoint,
              turnId: string,
            ) => this.historyCompactCoordinator.record(sessionId, checkpoint, runFor(turnId)),
            loadModelProjectionTransitions: async () =>
              loadModelProjectionTransitionsFromRunLedger(
                this.deps.runStore!,
                sessionId,
                await this.sessionRunIds(sessionId),
              ),
            recordModelProjectionTransition: (
              transition: ModelProjectionTransition,
              turnId: string,
            ) => {
              const run = runFor(turnId);
              if (!run) {
                return Promise.reject(
                  new Error('No active AgentRun for model projection transition'),
                );
              }
              return run.recordModelProjectionTransition(transition);
            },
          }
        : {}),
      ...(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();
            },
          }
        : {}),
    };
  }

  private async prepareBackendForExecution(
    sessionId: string,
    header: SessionHeader,
    execution: PendingExecutionClaim,
  ): Promise<`sha256:${string}` | undefined> {
    const existing = this.active.get(sessionId);
    if (existing) return existing.providerStateIdentity;
    const prepared = await this.deps.backends.prepare(header.backend, {
      sessionId,
      workspaceRoot: header.workspaceRoot,
      header,
      abortSignal: execution.abortController.signal,
    });
    execution.backendPreparation = prepared;
    return prepared.providerStateIdentity;
  }

  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 prepared =
        execution.backendPreparation ??
        (await this.deps.backends.prepare(header.backend, {
          sessionId,
          workspaceRoot: header.workspaceRoot,
          header,
          abortSignal: execution.abortController.signal,
        }));
      execution.run?.bindProviderStateIdentity(prepared.providerStateIdentity);
      const subagent = await this.resolveSubagentActivation(header);
      const backend = await prepared.build({
        sessionId,
        workspaceRoot: header.workspaceRoot,
        header,
        store: this.deps.store,
        abortSignal: execution.abortController.signal,
        ...(subagent
          ? {
              systemPrompt: subagent.systemPrompt,
              tools: subagent.tools,
              ...(subagent.shell ? { turnShellPlan: subagent.shell } : {}),
            }
          : {}),
        ...this.buildBackendRecorderHooks({
          sessionId,
        }),
        allowMidTurnHistoryCompaction: Boolean(this.deps.runtimeEventStore),
      });
      await this.rejectCancelledBackendActivation(backend, header, execution);
      const generation = this.createBackendGeneration(
        sessionId,
        backend,
        header,
        prepared.providerStateIdentity,
      );
      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 async resolveSubagentActivation(
    header: SessionHeader,
  ): Promise<{ systemPrompt: string; tools: MakaTool[]; shell?: TurnShellPlan } | undefined> {
    const snapshot = header.subagentRuntime;
    if (!snapshot) {
      if (header.subagentParent) {
        throw new Error('Linked child session is missing its durable runtime snapshot');
      }
      return undefined;
    }
    if (!header.subagentParent) {
      throw new Error('Subagent runtime snapshot requires a linked child session');
    }
    if (header.backend === 'plugin-executor') {
      return {
        systemPrompt: snapshot.systemPrompt,
        tools: [],
      };
    }
    const snapshotDefinition = {
      id: snapshot.agentId,
      permissionMode: header.permissionMode,
      tools: snapshot.toolNames,
    };
    const available = await this.childToolActivationForSession(header.id);
    const tools = buildToolsForAgentDefinition(available.tools, snapshotDefinition);
    if (tools.length !== snapshot.toolNames.length) {
      throw new Error('Subagent runtime tool snapshot is unavailable');
    }
    return {
      systemPrompt: snapshot.systemPrompt,
      tools,
      ...(available.shell ? { shell: available.shell } : {}),
    };
  }

  private async childToolActivationForSession(sessionId: string): Promise<ChildToolActivation> {
    if (!this.deps.resolveChildTools) return { tools: this.deps.childTools ?? [] };
    return await this.deps.resolveChildTools(sessionId);
  }

  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 createBackendGeneration(
    sessionId: string,
    backend: AgentBackend,
    header: SessionHeader,
    providerStateIdentity?: `sha256:${string}`,
  ): BackendGeneration {
    const active: BackendGeneration = {
      sessionId,
      generation: ++this.nextBackendGeneration,
      phase: 'active',
      backend,
      ...(providerStateIdentity ? { providerStateIdentity } : {}),
      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,
    execution: PendingExecutionClaim,
  ): Promise<void> {
    if (!execution.abortController.signal.aborted) return;
    const generation = this.createBackendGeneration(execution.sessionId, backend, header);
    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 this.active.get(active.sessionId) === 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 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 || invalidation.activations.size > 0 || 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);
      }
    }

    await this.waitForBackendDisposal(sessionId);
    if (execution.backendHeaderSnapshot?.invalidated) {
      // A refresh must not wait for preflight claims that may themselves be
      // waiting on the policy mutation gate. Remember their stale snapshots,
      // then re-arm invalidation inside activation, after any old disposal.
      this.ensureBackendInvalidation(sessionId);
    }
    const invalidation = this.backendInvalidations.get(sessionId);
    if (!invalidation) return;
    // This activation was already admitted when the refresh arrived. Let it
    // reserve its Run; invalidation must survive until that Run exits (or the
    // activation fails), rather than disposing a not-yet-reserved generation.
    if (invalidation.activations.has(execution)) return;
    await this.flushBackendInvalidation(sessionId);
    if (invalidation.activations.size > 0 || 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) {
      if (!existing.disposal) this.retainBackendActivations(sessionId, existing);
      return existing;
    }
    let resolve!: (outcome: BackendDisposalOutcome) => void;
    const outcome = new Promise<BackendDisposalOutcome>((resolvePromise) => {
      resolve = resolvePromise;
    });
    const invalidation: BackendInvalidationState = { outcome, resolve, activations: new Set() };
    this.retainBackendActivations(sessionId, invalidation);
    this.backendInvalidations.set(sessionId, invalidation);
    return invalidation;
  }

  private retainBackendActivations(
    sessionId: string,
    invalidation: BackendInvalidationState,
  ): void {
    // A cold backend has no generation or activeRuns yet. Retain the entire
    // prepare/build/reservation interval, not just the shared factory promise.
    for (const execution of this.backendActivations) {
      if (execution.sessionId === sessionId) invalidation.activations.add(execution);
    }
  }

  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: SessionHeaderPatch): Promise<SessionHeader> {
    const next = await this.deps.store.updateHeader(sessionId, patch);
    this.updateCachedHeader(sessionId, next);
    return next;
  }

  /** Present only when the store keeps a Session catalog to project into. */
  private messageProjectionHook(): Pick<AgentRunHooks, 'commitMessageProjection'> {
    const commit = this.deps.store.commitMessageCatalogProjection;
    if (!commit) return {};
    return {
      commitMessageProjection: async (sessionId, message) => {
        await commit.call(this.deps.store, sessionId, message);
        this.updateCachedHeader(sessionId, await this.deps.store.readHeader(sessionId));
      },
    };
  }
}

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,
  admissionRoute: ContinuationReplayAdmissionRoute,
): 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,
    admissionRoute,
  });
  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,
  targetOpening: RuntimeEventInvocationOpenedContent,
): 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,
    },
    targetOpening,
    claimedAt,
  };
}

/**
 * The opening fact the claim freezes for its target invocation.
 *
 * It has to be byte-identical to the one the target's own AgentRun computes:
 * the run compares them before it starts, so a claim can only admit the
 * execution it actually authorised.
 */
function continuationTargetOpeningForExecution(input: {
  continuation: RuntimeContinuation;
  sessionHeader: SessionHeader;
  userInput: UserMessageInput;
  workspaceIdentity: string;
  effectiveOrchestration: EffectiveOrchestration;
  effectiveToolMode: ToolMode;
  targetProviderStateIdentity: `sha256:${string}` | undefined;
  sourceOpening: RuntimeEventInvocationOpenedContent;
}): RuntimeEventInvocationOpenedContent {
  const { continuation, sessionHeader, userInput, effectiveOrchestration, effectiveToolMode } =
    input;
  if (!continuation.claimId || !continuation.boundary) {
    throw new RuntimeContinuationRevalidationError(
      'source_identity_changed',
      'Runtime continuation is missing its durable target-header identity',
    );
  }
  const lineage = {
    parentRunId: continuation.sourceRunId,
    ...(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 } : {}),
  };
  const opening: RuntimeEventInvocationOpenedContent = {
    kind: 'invocation_opened',
    protocol: 'invocation_opened_v1',
    route: runtimeInvocationRouteForHeader(sessionHeader, input.targetProviderStateIdentity),
    configuration: {
      cwd: sessionHeader.cwd,
      permissionMode: sessionHeader.permissionMode,
      collaborationMode: sessionHeader.collaborationMode ?? 'agent',
      orchestrationMode: effectiveOrchestration.mode,
      orchestrationSource: effectiveOrchestration.source,
      toolMode: effectiveToolMode,
      ...(effectiveOrchestration.agentSwarmAuthorization !== undefined
        ? { agentSwarmAuthorization: effectiveOrchestration.agentSwarmAuthorization }
        : {}),
      workspaceIdentity: input.workspaceIdentity,
    },
    root: { kind: 'user' },
    source: {
      kind: 'continuation',
      sourceInvocationId: continuation.sourceInvocationId,
      sourceRunId: continuation.sourceRunId,
      sourceTurnId: continuation.sourceTurnId,
      sourceRuntimeEventHighWater: continuation.sourceRuntimeEventHighWater,
      claimId: continuation.claimId,
      boundaryDigest: continuation.boundary.manifestDigest,
    },
    lineage,
  };
  return preserveHandoffOpening(continuation, opening, input.sourceOpening);
}

function consumeAdmittedRuntimeContinuation(input: {
  continuation: RuntimeContinuation;
  admissionRoute: ContinuationReplayAdmissionRoute;
  startAdmission: RuntimeContinuationStartAdmissionProof;
  toolBoundaryProtocol?: ToolBoundaryProtocol;
}): RuntimeContinuationMetadata {
  const { continuation } = input;
  assertRuntimeContinuationEnvelope(continuation);
  const startAdmissionIdentity = consumeRuntimeContinuationStartAdmissionProof(
    input.startAdmission,
  );
  const boundary = continuation.boundary
    ? decodeRuntimeBoundaryCursor(continuation.boundary)
    : undefined;
  if (
    !continuation.claimId ||
    !boundary ||
    continuation.providerProjectionVersion !== PROVIDER_REPLAY_PROJECTION_VERSION ||
    !continuation.providerReplayDigest ||
    !/^sha256:[0-9a-f]{64}$/.test(continuation.providerReplayDigest) ||
    !isDeepStrictEqual(startAdmissionIdentity, {
      startEventId: startAdmissionIdentity.startEventId,
      claimId: continuation.claimId,
      boundaryDigest: boundary.manifestDigest,
      providerProjectionVersion: continuation.providerProjectionVersion,
      providerReplayDigest: continuation.providerReplayDigest,
      ...(input.toolBoundaryProtocol ? { toolBoundaryProtocol: input.toolBoundaryProtocol } : {}),
      target: {
        sessionId: continuation.sessionId,
        invocationId: continuation.invocationId,
        runId: continuation.runId,
        turnId: continuation.turnId,
      },
    })
  ) {
    throw new Error('Runtime continuation durable admission identity is incomplete');
  }
  const immediateSource = boundary.segments.at(-1)!;
  if (
    boundary.manifestDigest !== continuation.boundary?.manifestDigest ||
    immediateSource.identity.sessionId !== continuation.sessionId ||
    immediateSource.identity.invocationId !== continuation.sourceInvocationId ||
    immediateSource.identity.runId !== continuation.sourceRunId ||
    immediateSource.identity.turnId !== continuation.sourceTurnId ||
    immediateSource.position.lastEventSeq !== continuation.sourceRuntimeEventHighWater
  ) {
    throw new Error('Runtime continuation durable admission boundary is inconsistent');
  }
  const replay = buildRuntimeEventModelReplayPlan(continuation.runtimeContext);
  const providerReasoningReplayEventIds = compatibleProviderReasoningReplayEventIds(
    continuation.runtimeContext,
    input.admissionRoute.invocations,
    input.admissionRoute.targetProviderStateIdentity,
    input.admissionRoute.targetModelId,
  );
  const admittedItems = admitProviderReasoningReplayItems(
    replay.items,
    providerReasoningReplayEventIds,
  );
  if (
    digestProviderReplayAdmission({
      providerProjectionVersion: continuation.providerProjectionVersion,
      targetProviderStateIdentity: input.admissionRoute.targetProviderStateIdentity,
      targetModelId: input.admissionRoute.targetModelId,
      items: admittedItems,
    }) !== continuation.providerReplayDigest
  ) {
    throw new Error('Runtime continuation provider replay identity changed after admission');
  }
  return {
    sourceInvocationId: continuation.sourceInvocationId,
    sourceRunId: continuation.sourceRunId,
    sourceTurnId: continuation.sourceTurnId,
    sourceRuntimeEventHighWater: continuation.sourceRuntimeEventHighWater,
  };
}

function assertRuntimeContinuationEnvelope(continuation: RuntimeContinuation): void {
  const sourceRuntimeContext = continuation.sourceRuntimeContext ?? continuation.runtimeContext;
  if (continuation.sourceRuntimeEventHighWater < sourceRuntimeContext.length) {
    throw new Error('Runtime continuation high-water is behind its replay context');
  }
  if (continuation.runtimeContext.length === 0) {
    throw new Error('Runtime continuation replay context must not be empty');
  }
  const mismatched = sourceRuntimeContext.find(
    (event) =>
      event.sessionId !== continuation.sessionId ||
      event.invocationId !== continuation.sourceInvocationId ||
      event.runId !== continuation.sourceRunId ||
      event.turnId !== continuation.sourceTurnId,
  );
  if (mismatched) {
    throw new Error(`Runtime continuation replay identity mismatch at event ${mismatched.id}`);
  }
  if (
    !isDeepStrictEqual(
      continuation.runtimeContext.slice(-sourceRuntimeContext.length),
      sourceRuntimeContext,
    )
  ) {
    throw new Error('Runtime continuation source replay is not the tail of provider history');
  }
  if (
    continuation.invocationId === continuation.sourceInvocationId ||
    continuation.runId === continuation.sourceRunId ||
    (continuation.handoffRootRunId === undefined
      ? continuation.turnId === continuation.sourceTurnId
      : continuation.turnId !== continuation.sourceTurnId)
  ) {
    throw new Error('Runtime continuation must use fresh invocation, run, and turn identities');
  }
}

function assertContinuationSourceUnchanged(
  continuation: RuntimeContinuation,
  sourceRun: RuntimeInvocationRecord,
  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 pause = terminalEvents.length === 1 ? runtimeHandoffPause(terminalEvents[0]!) : undefined;
  if (
    terminalEvents.length !== 1 ||
    (continuation.handoffRootRunId === undefined
      ? Boolean(pause) || terminalRunStatusFromRuntimeEvent(terminalEvents[0]!) === undefined
      : !pause ||
        pause.rootRunId !== continuation.handoffRootRunId ||
        pause.successorRunId !== continuation.runId ||
        pause.successorInvocationId !== continuation.invocationId ||
        pause.claimId !== continuation.claimId ||
        pause.remainingSteps !== continuation.handoffRemainingSteps)
  ) {
    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>;
}

function runtimeSteeringInput(
  owner: RuntimeMessageRunOwner | undefined,
): Pick<BackendSendInput, 'pullSteering' | 'ackSteering' | 'nackSteering'> {
  return owner
    ? {
        pullSteering: () => owner.pull(),
        ackSteering: (ids) => owner.ack(ids),
        nackSteering: (ids) => owner.nack(ids),
      }
    : {};
}

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 effectiveOrchestrationForRun(
  run: RuntimeInvocationRecord,
  session: SessionHeader,
): EffectiveOrchestration {
  const configuration = run.opening.configuration;
  if (configuration.agentSwarmAuthorization !== undefined) {
    return {
      mode: configuration.orchestrationMode,
      source: configuration.orchestrationSource,
      agentSwarmAuthorization: configuration.agentSwarmAuthorization,
    };
  }
  return resolveEffectiveOrchestration(session.orchestrationMode, undefined);
}

function effectiveToolModeForRun(run: RuntimeInvocationRecord): ToolMode {
  return run.opening.configuration.toolMode;
}

function assertNoRemovedChildAgentRunLineage(input: UserMessageInput): void {
  const legacy = input as UserMessageInput & {
    parentRunId?: unknown;
    resumedFromRunId?: unknown;
    retriedFromRunId?: unknown;
  };
  if (
    legacy.parentRunId !== undefined ||
    legacy.resumedFromRunId !== undefined ||
    legacy.retriedFromRunId !== undefined
  ) {
    throw new Error('Live Turn cannot use removed child AgentRun lineage');
  }
}

async function interactionResumeAllowed(
  interactionRun: RuntimeInteractionRunBinding | undefined,
  event: SessionEvent,
): Promise<boolean> {
  if (!interactionRun || !isHostedInteractionSettlementAckEvent(event)) {
    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 };