import { spawn as spawnChild, type ChildProcess } from "node:child_process";
import { mkdir } from "node:fs/promises";
import { join } from "node:path";

import { sha256, writeWorkflowJson } from "./atomic-storage.js";
import { budgetExceeded, budgetValue } from "./budget.js";
import { NodePiSessionForker, type PiSessionForker } from "./pi-session-fork.js";
import { classifyProviderFallbackFailure } from "./model-fallback.js";
import { parseJsonCandidate } from "./json-candidate.js";
import {
  effectiveWorkerContract,
  BudgetExceededError,
  OUTPUT_BYTE_UNIT_NOTE,
  WorkflowRunStore,
} from "./run-store.js";
import { reportMatchesSchema } from "./worker-gate.js";
import { buildWorkerPrompt } from "./worker-prompt-capsule.js";
import {
  resolveWorkerLaunch,
  type PiProcessLaunchCommand,
  type WorkerLaunchAuthority,
} from "./worker-launch.js";
import {
  WorkflowLeaseManager,
  type ProcessIdentity,
  type ProcessIdentityInspector,
} from "./workflow-lease.js";
import type {
  AttemptId,
  AcceptanceCheck,
  ContractRevisionId,
  PeerMessageId,
  BudgetLimit,
  ReportSchemaV1,
  RunProjectionV2,
  WorkerAttemptProjection,
  WorkerId,
  WorkerProjection,
  WorkerRequest,
} from "./workflow-types.js";

export const WORKER_PROTOCOL_LINE_LIMIT_BYTES = 4 * 1024 * 1024;
export const WORKER_STDERR_TAIL_LIMIT_BYTES = 128 * 1024;

export class WorkerProtocolError extends Error {
  constructor(
    readonly code: "protocol_output_limit" | "protocol_invalid_json",
    message: string,
    readonly limitBytes?: number,
  ) {
    super(message);
    this.name = "WorkerProtocolError";
  }
}

export { BudgetExceededError } from "./run-store.js";
export type { PiProcessLaunchCommand } from "./worker-launch.js";

export interface ProcessExit {
  readonly code: number | null;
  readonly signal: string | null;
}

export interface ManagedWorkerProcess {
  readonly pid: number;
  readonly stdout: AsyncIterable<Uint8Array>;
  readonly stderr: AsyncIterable<Uint8Array>;
  readonly protocol: AsyncIterable<Uint8Array>;
  readonly exited: Promise<ProcessExit>;
  writeLine(line: string): Promise<void>;
  endInput(): Promise<void>;
  signalProcessGroup(signal: NodeJS.Signals): Promise<void>;
  closeParentLiveness(): Promise<void>;
}

export interface PiProcessFactory {
  spawn(command: PiProcessLaunchCommand): Promise<ManagedWorkerProcess>;
}

export interface WorkerTimerScheduler {
  schedule(delayMs: number, callback: () => void): { cancel(): void };
}

export interface WorkerProcessSupervisorOptions {
  readonly namespaceDirectory: string;
  readonly store: WorkflowRunStore;
  readonly leaseManager: WorkflowLeaseManager;
  readonly supervisorGeneration: string;
  readonly inspector: ProcessIdentityInspector;
  readonly processFactory: PiProcessFactory;
  readonly launchAuthority: WorkerLaunchAuthority;
  readonly timerScheduler?: WorkerTimerScheduler;
  readonly now?: () => Date;
  readonly cancellationGraceMs?: number;
  readonly sessionForker?: PiSessionForker;
}

export type SpawnWorkerResult =
  | {
      readonly kind: "started";
      readonly worker: WorkerProjection;
      readonly attempt: WorkerAttemptProjection;
      readonly settlement: Promise<RunProjectionV2>;
    }
  | {
      readonly kind: "existing" | "needs-reconcile";
      readonly worker: WorkerProjection;
      readonly attempt: WorkerAttemptProjection;
    };

export type CancelWorkerResult =
  | { readonly kind: "requested"; readonly projection: RunProjectionV2 }
  | { readonly kind: "indeterminate"; readonly projection: RunProjectionV2; readonly reason: string };

export class ResumeRefusedError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "ResumeRefusedError";
  }
}

interface ManagedAttempt {
  readonly requestKey: string;
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly leaseGeneration: string;
  readonly child: ManagedWorkerProcess;
  readonly deadline: { cancel(): void };
  inferenceActive: boolean;
  readonly pendingRevisions: Array<{
    readonly revisionId: ContractRevisionId;
    readonly prompt: string;
  }>;
  readonly pendingPeerMessages: Array<{
    readonly messageId: PeerMessageId;
    readonly senderWorkerId: WorkerId;
    readonly message: string;
    readonly correlation: string;
  }>;
  cancellationEscalation?: { cancel(): void };
}

interface MonitorState {
  outputBytes: number;
  tokens: number;
  costUsd: number;
  candidateValue: string | null;
  candidateHash: string | null;
  ready: boolean;
  stderrTail: Uint8Array;
  fallbackReason: string | null;
}

interface MonitorResult {
  readonly projection: RunProjectionV2;
  readonly fallbackReason: string | null;
}

class ModelFallbackEligibleError extends Error {
  constructor(readonly fallbackReason: string, message: string) {
    super(message);
    this.name = "ModelFallbackEligibleError";
  }
}

export class WorkerProcessSupervisor {
  readonly #namespaceDirectory: string;
  readonly #store: WorkflowRunStore;
  readonly #leaseManager: WorkflowLeaseManager;
  readonly #supervisorGeneration: string;
  readonly #inspector: ProcessIdentityInspector;
  readonly #processFactory: PiProcessFactory;
  #launchAuthority: WorkerLaunchAuthority;
  readonly #timers: WorkerTimerScheduler;
  readonly #now: () => Date;
  readonly #cancellationGraceMs: number;
  readonly #sessionForker: PiSessionForker;
  readonly #managed = new Map<WorkerId, ManagedAttempt>();

  constructor(options: WorkerProcessSupervisorOptions) {
    this.#namespaceDirectory = options.namespaceDirectory;
    this.#store = options.store;
    this.#leaseManager = options.leaseManager;
    this.#supervisorGeneration = options.supervisorGeneration;
    this.#inspector = options.inspector;
    this.#processFactory = options.processFactory;
    this.#launchAuthority = options.launchAuthority;
    this.#timers = options.timerScheduler ?? new NodeWorkerTimerScheduler();
    this.#now = options.now ?? (() => new Date());
    this.#cancellationGraceMs = options.cancellationGraceMs ?? 2_000;
    this.#sessionForker = options.sessionForker ?? new NodePiSessionForker();
  }

  updateLaunchAuthority(authority: WorkerLaunchAuthority): void {
    this.#launchAuthority = authority;
  }

  async spawnWorker(requestKey: string, input: {
    readonly request: WorkerRequest;
    readonly prompt: string;
    readonly depth: number;
    readonly parentWorkerId?: WorkerId | null;
    readonly continueWorkerId?: WorkerId;
  }): Promise<SpawnWorkerResult> {
    const run = await this.#requiredRun(requestKey);
    if (input.continueWorkerId !== undefined && input.request.context === "fork") {
      throw new TypeError("A worker cannot continue another worker session and fork driver context together");
    }
    const continuation = input.continueWorkerId === undefined
      ? undefined
      : await this.#continuationSource(run, input.continueWorkerId);
    const sessionPath = continuation?.sessionPath
      ?? this.#sessionDirectory(run.runId, input.request.requestKey);
    const forkedSessionFile = input.request.context === "fork"
      ? join(sessionPath, "session.jsonl")
      : undefined;
    const prompt = continuation === undefined
      ? input.prompt
      : buildWorkerPrompt({
          ...input.request,
          continuation: {
            priorWorkerId: continuation.priorWorkerId,
            priorAttemptId: continuation.priorAttemptId,
            sessionFile: continuation.sessionFile,
          },
        });
    // Pure preflight uses disposable runtime IDs. It proves launch authority before any ledger,
    // lease, filesystem, or process effect.
    resolveWorkerLaunch({
      contract: {
        ...input.request,
        context: input.request.context ?? "fresh",
        runId: run.runId,
        workerId: "worker_preflight",
        createdByDriverSessionId: run.source.driverSessionId,
      },
      attempt: {
        attemptId: "attempt_preflight",
        workerAttemptGeneration: "preflight-worker-generation",
        supervisorGeneration: this.#supervisorGeneration,
      },
      depth: input.depth,
      prompt,
      sessionDirectory: sessionPath,
      launchContractPath: join(this.#namespaceDirectory, "launch-contracts", "preflight.json"),
      ...(continuation ? { continuedSessionFile: continuation.sessionFile } : {}),
      ...(forkedSessionFile ? { forkedSessionFile } : {}),
      modelOverride: input.request.model,
      authority: this.#launchAuthority,
    });
    const prepared = await this.#store.prepareWorkerIntent(requestKey, {
      createdByDriverSessionId: run.source.driverSessionId,
      sessionPath,
      request: input.request,
      parentWorkerId: input.parentWorkerId ?? null,
      depth: input.depth,
      ...(continuation ? { continuation: {
        priorWorkerId: continuation.priorWorkerId,
        priorAttemptId: continuation.priorAttemptId,
        priorLeaseGeneration: continuation.priorLeaseGeneration,
        sessionFile: continuation.sessionFile,
      } } : {}),
    });
    const attempt = currentAttempt(prepared.worker);
    if (prepared.kind === "existing") {
      const managed = this.#managed.get(prepared.worker.workerId);
      if (managed?.attemptId === attempt.attemptId) {
        return { kind: "existing", worker: prepared.worker, attempt };
      }
      return { kind: "needs-reconcile", worker: prepared.worker, attempt };
    }
    if (prepared.worker.contract.context === "fork") {
      const forkSource = prepared.worker.contract.forkSource;
      if (!forkSource || !forkedSessionFile) throw new Error("Fork worker context is incomplete");
      try {
        await this.#sessionForker.fork({ ...forkSource, targetSessionFile: forkedSessionFile });
      } catch (error) {
        await this.#store.recordAttemptExit(requestKey, {
          workerId: prepared.worker.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          exitStatus: { code: 1, signal: null },
          candidateOutcome: { valueOrRef: null, hash: null },
          uncertainSideEffects: false,
          reason: `Pi session fork failed before process creation: ${toError(error).message}`,
        });
        throw error;
      }
    }
    return this.#startPreparedAttempt(requestKey, prepared.worker, attempt, prompt);
  }

  async resumeWorker(requestKey: string, input: {
    readonly workerId: WorkerId;
    readonly prompt: string;
  }): Promise<SpawnWorkerResult> {
    let run = await this.#requiredRun(requestKey);
    let worker = requiredWorker(run, input.workerId);
    let attempt = currentAttempt(worker);
    if (attempt.state === "ambiguous" || attempt.sideEffects.status === "uncertain") {
      throw new ResumeRefusedError(
        `Worker ${input.workerId} has unresolved side-effect ambiguity`,
      );
    }
    if (attempt.state === "running") {
      const inspection = await this.#inspectAttempt(attempt);
      if (inspection !== "dead") {
        throw new ResumeRefusedError(
          `Worker ${input.workerId} cannot resume because prior liveness is ${inspection}`,
        );
      }
      run = await this.#store.recordAttemptExit(requestKey, {
        workerId: worker.workerId,
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        exitStatus: { code: 1, signal: "proven-dead-before-resume" },
        candidateOutcome: attempt.candidateOutcome,
        uncertainSideEffects: false,
        reason: "Prior worker process is proven dead before explicit resume",
      });
      worker = requiredWorker(run, input.workerId);
      attempt = currentAttempt(worker);
    }
    if (attempt.state === "pending") {
      throw new ResumeRefusedError(
        `Worker ${input.workerId} has an unreconciled spawn intent; replay is forbidden`,
      );
    }
    const prepared = await this.#store.prepareReplacementAttempt(requestKey, {
      workerId: input.workerId,
    });
    return this.#startPreparedAttempt(
      requestKey,
      prepared.worker,
      prepared.attempt,
      input.prompt,
    );
  }

  async cancelWorker(requestKey: string, input: {
    readonly workerId: WorkerId;
    readonly reason: string;
  }): Promise<CancelWorkerResult> {
    let projection = await this.#store.requestWorkerCancellation(requestKey, input);
    const attempt = currentAttempt(requiredWorker(projection, input.workerId));
    const managed = this.#managed.get(input.workerId);
    if (!managed || managed.attemptId !== attempt.attemptId
      || managed.workerAttemptGeneration !== attempt.workerAttemptGeneration) {
      return {
        kind: "indeterminate",
        projection,
        reason: "No current supervised process is attached to the durable cancellation intent",
      };
    }
    if (attempt.launchContract?.channel === "rpc") {
      await managed.child.writeLine(JSON.stringify({
        id: `cancel-${attempt.attemptId}`,
        type: "abort",
      })).catch(() => {});
    }
    try {
      await managed.child.signalProcessGroup("SIGTERM");
    } catch (error) {
      projection = await this.#store.recordWorkerNeedsAttention(requestKey, {
        workerId: input.workerId,
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        reason: `Cancellation signal outcome is unknown: ${toError(error).message}`,
        ambiguous: true,
      });
      return { kind: "indeterminate", projection, reason: toError(error).message };
    }
    projection = await this.#store.recordCancellationSignalSent(requestKey, {
      workerId: input.workerId,
      attemptId: attempt.attemptId,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      signal: "SIGTERM",
    });
    managed.cancellationEscalation?.cancel();
    managed.cancellationEscalation = this.#timers.schedule(this.#cancellationGraceMs, () => {
      const current = this.#managed.get(input.workerId);
      if (current?.attemptId === attempt.attemptId) {
        void current.child.signalProcessGroup("SIGKILL").catch(() => {});
      }
    });
    return { kind: "requested", projection };
  }

  async steerWorker(requestKey: string, input: {
    readonly workerId: WorkerId;
    readonly message: string;
    readonly correlation?: string;
  }): Promise<RunProjectionV2> {
    const managed = this.#managed.get(input.workerId);
    if (!managed) throw new Error(`Worker ${input.workerId} has no attached RPC process`);
    const projection = await this.#requiredRun(requestKey);
    const worker = requiredWorker(projection, input.workerId);
    if (effectiveWorkerContract(worker).channel !== "rpc") {
      throw new Error("Only an RPC worker can be steered");
    }
    const message = nonEmpty(input.message, "worker steer message");
    const correlation = input.correlation === undefined
      ? null
      : boundedProtocolText(input.correlation, "worker steer correlation", 256);
    assertBlockedWakeCorrelation(worker, correlation);
    const command = managed.inferenceActive ? "steer" : "prompt";
    await managed.child.writeLine(JSON.stringify({
      id: `steer-${managed.attemptId}`,
      type: command,
      message,
    }));
    managed.inferenceActive = true;
    try {
      return await this.#store.recordSteerDelivered(requestKey, {
        workerId: input.workerId,
        messageHash: sha256(message),
        ...(correlation === null ? {} : { correlation }),
      });
    } catch (error) {
      await this.#store.recordWorkerNeedsAttention(requestKey, {
        workerId: worker.workerId,
        attemptId: managed.attemptId,
        workerAttemptGeneration: managed.workerAttemptGeneration,
        reason: `Steer reached the worker but correlated delivery ledgering failed: ${toError(error).message}`,
        ambiguous: true,
      }).catch(() => {});
      throw error;
    }
  }

  async reviseWorkerContract(requestKey: string, input: {
    readonly workerId: WorkerId;
    readonly correlation?: string;
    readonly goal?: string;
    readonly acceptance?: readonly string[];
    readonly constraints?: readonly string[];
    readonly report_schema?: ReportSchemaV1;
    readonly requires_human_approval?: boolean;
    readonly acceptance_check?: AcceptanceCheck;
  }): Promise<RunProjectionV2> {
    const before = await this.#requiredRun(requestKey);
    const workerBefore = requiredWorker(before, input.workerId);
    if (effectiveWorkerContract(workerBefore).channel !== "rpc") {
      throw new Error("Contract revision delivery requires an RPC worker");
    }
    let projection = await this.#store.requestContractRevision(requestKey, input);
    const worker = requiredWorker(projection, input.workerId);
    const revision = worker.contractRevisions.at(-1);
    if (!revision || revision.status !== "pending") {
      throw new Error("Contract revision was not recorded as pending");
    }
    const managed = this.#managed.get(input.workerId);
    if (!managed) return projection;
    const prompt = [
      `Apply immutable contract revision ${revision.revision} at this inference boundary.`,
      buildWorkerPrompt(revision.contract),
    ].join("\n\n");
    if (managed.inferenceActive) {
      managed.pendingRevisions.push({ revisionId: revision.revisionId, prompt });
      return projection;
    }
    await managed.child.writeLine(JSON.stringify({
      id: `revision-${revision.revisionId}`,
      type: "prompt",
      message: prompt,
    }));
    managed.inferenceActive = true;
    projection = await this.#recordRevisionDelivered(managed, revision.revisionId);
    return projection;
  }

  async reconcileRun(requestKey: string): Promise<RunProjectionV2> {
    let projection = await this.#requiredRun(requestKey);
    const reconciledPeerMessages = new Set<PeerMessageId>();
    for (const worker of Object.values(projection.workers)) {
      for (const peer of worker.peerMessages) {
        if (peer.status !== "pending" || reconciledPeerMessages.has(peer.messageId)) continue;
        reconciledPeerMessages.add(peer.messageId);
        const recipient = requiredWorker(projection, peer.recipientWorkerId);
        const recipientAttempt = currentAttempt(recipient);
        projection = await this.#store.recordPeerMessageDelivery(requestKey, {
          messageId: peer.messageId,
          recipientWorkerId: recipient.workerId,
          recipientAttemptId: recipientAttempt.attemptId,
          recipientWorkerAttemptGeneration: recipientAttempt.workerAttemptGeneration,
          status: "indeterminate",
          diagnostic: "Pending peer-message body was not persisted and cannot be replayed after supervisor restart",
        });
      }
    }
    for (const worker of Object.values(projection.workers)) {
      const attempt = currentAttempt(worker);
      if (attempt.state === "pending") {
        projection = await this.#store.recordWorkerNeedsAttention(requestKey, {
          workerId: worker.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          reason: "Recovered spawn intent has no durable process-start evidence; no replay performed",
          ambiguous: true,
        });
        continue;
      }
      if (attempt.state !== "running") continue;
      const liveness = await this.#inspectAttempt(attempt);
      if (liveness === "dead") {
        projection = await this.#store.recordAttemptExit(requestKey, {
          workerId: worker.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          exitStatus: { code: 1, signal: "proven-dead-on-recovery" },
          candidateOutcome: attempt.candidateOutcome,
          uncertainSideEffects: attempt.sideEffects.policy === "external"
            || attempt.sideEffects.policy === "reconcilable",
          reason: "Worker was proven dead during restart reconciliation",
        });
      } else {
        projection = await this.#store.recordWorkerNeedsAttention(requestKey, {
          workerId: worker.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          reason: liveness === "live"
            ? "Worker remains live but its control pipe was not reattached"
            : "Worker liveness could not be verified after restart",
          ambiguous: liveness === "unknown",
        });
      }
    }
    return projection;
  }

  async #startPreparedAttempt(
    requestKey: string,
    worker: WorkerProjection,
    attempt: WorkerAttemptProjection,
    prompt: string,
  ): Promise<SpawnWorkerResult> {
    const launchContractPath = join(
      this.#namespaceDirectory,
      "launch-contracts",
      worker.workerId,
      `${attempt.attemptId}.json`,
    );
    const resolved = resolveWorkerLaunch({
      contract: worker.contract,
      attempt: {
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        supervisorGeneration: attempt.supervisorGeneration,
      },
      depth: worker.depth,
      prompt,
      sessionDirectory: attempt.session.path,
      launchContractPath,
      ...(worker.continuation
        ? { continuedSessionFile: worker.continuation.sessionFile }
        : {}),
      ...(worker.contract.context === "fork"
        ? { forkedSessionFile: join(attempt.session.path, "session.jsonl") }
        : {}),
      modelOverride: attempt.modelSelection?.model ?? worker.contract.model,
      authority: this.#launchAuthority,
    });
    await mkdir(attempt.session.path, { recursive: true, mode: 0o700 });
    await mkdir(join(this.#namespaceDirectory, "launch-contracts", worker.workerId), {
      recursive: true,
      mode: 0o700,
    });
    await writeWorkflowJson(launchContractPath, resolved.contract, { createOnly: true });
    const reservation = await this.#leaseManager.reserveWorkerSession({
      sessionPath: attempt.session.path,
      runId: worker.contract.runId,
      workerId: worker.workerId,
      attemptId: attempt.attemptId,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      supervisorGeneration: this.#supervisorGeneration,
    });
    if (reservation.kind !== "acquired") {
      await this.#store.recordWorkerNeedsAttention(requestKey, {
        workerId: worker.workerId,
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        reason: reservation.kind === "held"
          ? `Worker session is held by generation ${reservation.owner.generation}`
          : reservation.diagnostic,
        ambiguous: reservation.kind !== "held",
      });
      throw new ResumeRefusedError("Worker session could not be reserved before process creation");
    }
    let child: ManagedWorkerProcess;
    try {
      child = await this.#processFactory.spawn(resolved.command);
    } catch (error) {
      await this.#leaseManager.releaseWorkerSession({
        sessionPath: attempt.session.path,
        generation: reservation.owner.generation,
        supervisorGeneration: this.#supervisorGeneration,
      });
      await this.#store.recordAttemptExit(requestKey, {
        workerId: worker.workerId,
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        exitStatus: { code: 127, signal: null },
        candidateOutcome: { valueOrRef: null, hash: null },
        uncertainSideEffects: false,
        reason: `Pi process creation failed: ${toError(error).message}`,
      });
      throw error;
    }
    const inspection = await this.#inspector.inspect(child.pid);
    if (inspection.kind !== "live") {
      await child.signalProcessGroup("SIGKILL").catch(() => {});
      const postKillInspection = await this.#inspector.inspect(child.pid);
      if (postKillInspection.kind === "absent") {
        await this.#leaseManager.releaseWorkerSession({
          sessionPath: attempt.session.path,
          generation: reservation.owner.generation,
          supervisorGeneration: this.#supervisorGeneration,
        });
      }
      await this.#store.recordWorkerNeedsAttention(requestKey, {
        workerId: worker.workerId,
        attemptId: attempt.attemptId,
        workerAttemptGeneration: attempt.workerAttemptGeneration,
        reason: postKillInspection.kind === "unknown"
          ? postKillInspection.diagnostic
          : "Spawned child disappeared before process identity could be recorded",
        ambiguous: true,
      });
      throw new Error("Spawned worker identity could not be proven");
    }
    const activeLease = await this.#leaseManager.activateWorkerSession({
      sessionPath: attempt.session.path,
      generation: reservation.owner.generation,
      supervisorGeneration: this.#supervisorGeneration,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      processIdentity: inspection.identity,
    });
    let projection = await this.#store.recordAttemptStarted(requestKey, {
      workerId: worker.workerId,
      attemptId: attempt.attemptId,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      leaseGeneration: activeLease.generation,
      processIdentity: inspection.identity,
      launchContract: resolved.contract,
    });
    worker = requiredWorker(projection, worker.workerId);
    attempt = currentAttempt(worker);

    let resolveReady!: () => void;
    let rejectReady!: (error: Error) => void;
    const ready = new Promise<void>((resolve, reject) => {
      resolveReady = resolve;
      rejectReady = reject;
    });
    const state: MonitorState = {
      outputBytes: 0,
      tokens: 0,
      costUsd: 0,
      candidateValue: null,
      candidateHash: null,
      ready: false,
      stderrTail: new Uint8Array(),
      fallbackReason: null,
    };
    const wallClockMs = budgetValue(worker.contract.budget.maxWallClockMs);
    const deadline = wallClockMs === undefined ? { cancel(): void {} } : this.#timers.schedule(
      wallClockMs,
      () => {
        void this.cancelWorker(requestKey, {
          workerId: worker.workerId,
          reason: "worker wall-clock budget exhausted",
        }).catch(() => {});
      },
    );
    const managed: ManagedAttempt = {
      requestKey,
      workerId: worker.workerId,
      attemptId: attempt.attemptId,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      leaseGeneration: activeLease.generation,
      child,
      deadline,
      inferenceActive: false,
      pendingRevisions: [],
      pendingPeerMessages: [],
    };
    this.#managed.set(worker.workerId, managed);
    const monitored = this.#monitorAttempt({
      requestKey,
      worker,
      attempt,
      child,
      leaseGeneration: activeLease.generation,
      state,
      resolveReady,
      rejectReady,
    });
    try {
      if (worker.contract.channel === "json-result") await child.endInput();
      await ready;
    } catch (error) {
      const result = await monitored.catch(() => undefined);
      if (result?.fallbackReason && this.#nextFallbackModel(result.projection, worker.workerId, attempt)) {
        return this.#startModelFallback(
          requestKey,
          result.projection,
          worker.workerId,
          attempt,
          result.fallbackReason,
          prompt,
        );
      }
      throw error;
    }
    if (worker.contract.channel === "rpc") {
      await child.writeLine(JSON.stringify({
        id: `prompt-${attempt.attemptId}`,
        type: "prompt",
        message: prompt,
      }));
      managed.inferenceActive = true;
    }
    projection = await this.#requiredRun(requestKey);
    worker = requiredWorker(projection, worker.workerId);
    attempt = currentAttempt(worker);
    const settlement = monitored.then((result) => this.#settleModelFallback(
      requestKey,
      result,
      worker.workerId,
      attempt,
      prompt,
    ));
    return { kind: "started", worker, attempt, settlement };
  }

  async #settleModelFallback(
    requestKey: string,
    result: MonitorResult,
    workerId: WorkerId,
    attempt: WorkerAttemptProjection,
    prompt: string,
  ): Promise<RunProjectionV2> {
    if (!result.fallbackReason) return result.projection;
    const next = this.#nextFallbackModel(result.projection, workerId, attempt);
    if (!next) return result.projection;
    try {
      const fallback = await this.#startModelFallback(
        requestKey,
        result.projection,
        workerId,
        attempt,
        result.fallbackReason,
        prompt,
      );
      return fallback.kind === "started"
        ? fallback.settlement
        : this.#requiredRun(requestKey);
    } catch {
      // A replacement attempt's provider error can race its own ready signal on a separate
      // stream and surface here as a throw, even though the attempt's exit is already durably
      // ledgered by its monitor. Settlement reports durable state; it never rejects on the
      // ordering of that race.
      return this.#requiredRun(requestKey);
    }
  }

  async #startModelFallback(
    requestKey: string,
    projection: RunProjectionV2,
    workerId: WorkerId,
    attempt: WorkerAttemptProjection,
    reason: string,
    prompt: string,
  ): Promise<SpawnWorkerResult> {
    const next = this.#nextFallbackModel(projection, workerId, attempt);
    if (!next) throw new Error(`Model fallback chain exhausted after ${reason}`);
    const prepared = await this.#store.prepareReplacementAttempt(requestKey, {
      workerId,
      model: next,
      fallbackReason: reason,
    });
    return this.#startPreparedAttempt(
      requestKey,
      prepared.worker,
      prepared.attempt,
      prompt,
    );
  }

  #nextFallbackModel(
    projection: RunProjectionV2,
    workerId: WorkerId,
    attempt: WorkerAttemptProjection,
  ) {
    const worker = requiredWorker(projection, workerId);
    const chain = [worker.contract.model, ...(worker.contract.modelFallbacks ?? [])];
    const attempted = attempt.modelSelection?.model ?? attempt.launchContract?.model ?? worker.contract.model;
    const index = chain.findIndex((model) =>
      model.provider === attempted.provider && model.id === attempted.id);
    return index < 0 ? undefined : chain[index + 1];
  }

  async #monitorAttempt(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly attempt: WorkerAttemptProjection;
    readonly child: ManagedWorkerProcess;
    readonly leaseGeneration: string;
    readonly state: MonitorState;
    readonly resolveReady: () => void;
    readonly rejectReady: (error: Error) => void;
  }): Promise<MonitorResult> {
    let fatalStreamError: Error | undefined;
    let fallbackStreamError: ModelFallbackEligibleError | undefined;
    let fatalCancellationStarted = false;
    const captureStreamError = (error: unknown): Promise<void> | void => {
      const observed = toError(error);
      if (observed instanceof ModelFallbackEligibleError) {
        fallbackStreamError ??= observed;
        if (fatalStreamError === undefined) {
          input.state.fallbackReason ??= observed.fallbackReason;
        }
        input.rejectReady(observed);
        return;
      }
      fatalStreamError ??= observed;
      input.state.fallbackReason = null;
      input.rejectReady(observed);
      if (fatalCancellationStarted) return;
      fatalCancellationStarted = true;
      return this.cancelWorker(input.requestKey, {
        workerId: input.worker.workerId,
        reason: `worker protocol failed: ${observed.message}`,
      }).then(() => {}).catch(() => {});
    };
    const stdout = this.#consumeStdout(input).catch(captureStreamError);
    const protocol = this.#consumeProtocol(input).catch(captureStreamError);
    const stderr = this.#consumeStderr(input);
    const exit = await input.child.exited;
    await Promise.all([stdout, protocol, stderr]);
    const streamError = fatalStreamError ?? fallbackStreamError;
    if (!input.state.ready && streamError === undefined) {
      input.rejectReady(new Error(
        `Worker exited before launch parity was ready after protocol drain: ${formatExit(exit)}`,
      ));
    }
    const managed = this.#managed.get(input.worker.workerId);
    managed?.deadline.cancel();
    managed?.cancellationEscalation?.cancel();
    await input.child.closeParentLiveness().catch(() => {});
    let projection = await this.#store.recordAttemptUsage(input.requestKey, {
      workerId: input.worker.workerId,
      attemptId: input.attempt.attemptId,
      workerAttemptGeneration: input.attempt.workerAttemptGeneration,
      tokens: input.state.tokens,
      costUsd: input.state.costUsd,
      outputBytes: input.state.outputBytes,
    });
    const uncertain = input.worker.contract.sideEffectPolicy === "external"
      || input.worker.contract.sideEffectPolicy === "reconcilable";
    projection = await this.#store.recordAttemptExit(input.requestKey, {
      workerId: input.worker.workerId,
      attemptId: input.attempt.attemptId,
      workerAttemptGeneration: input.attempt.workerAttemptGeneration,
      exitStatus: streamError ? { code: 1, signal: exit.signal } : exit,
      candidateOutcome: {
        valueOrRef: input.state.candidateValue,
        hash: input.state.candidateHash,
      },
      uncertainSideEffects: uncertain && (exit.code !== 0 || streamError !== undefined),
      reason: streamError
        ? `Worker protocol failed: ${protocolErrorDetails(streamError)}`
        : `Worker process exited: ${formatExit(exit)}${
            exit.code === 0 || input.state.stderrTail.byteLength === 0
              ? ""
              : `; stderr_tail=${JSON.stringify(new TextDecoder().decode(input.state.stderrTail))}`
          }`,
    });
    projection = await this.#recordAutomaticGateEvaluation(input.requestKey, projection, input.worker.workerId);
    await this.#leaseManager.releaseWorkerSession({
      sessionPath: input.attempt.session.path,
      generation: input.leaseGeneration,
      supervisorGeneration: this.#supervisorGeneration,
    });
    if (this.#managed.get(input.worker.workerId)?.attemptId === input.attempt.attemptId) {
      this.#managed.delete(input.worker.workerId);
    }
    return { projection, fallbackReason: input.state.fallbackReason };
  }

  async #recordAutomaticGateEvaluation(
    requestKey: string,
    projection: RunProjectionV2,
    workerId: WorkerId,
  ): Promise<RunProjectionV2> {
    const worker = projection.workers[workerId];
    const attempt = worker?.attempts.at(-1);
    if (!worker || !attempt || attempt.state !== "running") return projection;
    const evidenceExists = attempt.candidateOutcome.valueOrRef !== null
      || attempt.candidateOutcome.hash !== null;
    if (!evidenceExists) return projection;

    const contract = effectiveWorkerContract(worker);
    const reportMatches = contract.report_schema
      ? reportMatchesSchema(contract.report_schema, parseJsonCandidate(attempt.candidateOutcome.valueOrRef))
      : null;
    return this.#store.recordGateEvaluation(requestKey, {
      workerId,
      reportMatches,
      acceptanceCheckExitCode: null,
      humanApproved: null,
      evidenceExists,
    });
  }

  async #consumeStdout(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly attempt: WorkerAttemptProjection;
    readonly child: ManagedWorkerProcess;
    readonly state: MonitorState;
    readonly resolveReady: () => void;
  }): Promise<void> {
    await this.#consumeJsonLines(input, input.child.stdout, "pi-output");
  }

  async #consumeProtocol(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly attempt: WorkerAttemptProjection;
    readonly child: ManagedWorkerProcess;
    readonly state: MonitorState;
    readonly resolveReady: () => void;
  }): Promise<void> {
    await this.#consumeJsonLines(input, input.child.protocol, "worker-control");
  }

  async #consumeJsonLines(
    input: {
      readonly requestKey: string;
      readonly worker: WorkerProjection;
      readonly attempt: WorkerAttemptProjection;
      readonly child: ManagedWorkerProcess;
      readonly state: MonitorState;
      readonly resolveReady: () => void;
    },
    stream: AsyncIterable<Uint8Array>,
    source: "pi-output" | "worker-control",
  ): Promise<void> {
    const decoder = new TextDecoder();
    let buffer = "";
    for await (const chunk of stream) {
      // maxOutputBytes budgets the worker's own output. Control envelopes are supervisor
      // bookkeeping, so counting them would silently tighten a declared budget.
      if (source === "pi-output") input.state.outputBytes += chunk.byteLength;
      buffer += decoder.decode(chunk, { stream: true });
      if (!buffer.includes("\n") && Buffer.byteLength(buffer, "utf8") > WORKER_PROTOCOL_LINE_LIMIT_BYTES) {
        throw protocolOutputLimit();
      }
      await this.#consumeLines(buffer, false, async (line, rest) => {
        buffer = rest;
        await this.#handleJsonLine(input, line, source);
      });
      await this.#enforceObservedBudgets(input);
    }
    buffer += decoder.decode();
    if (buffer.length > 0) {
      await this.#handleJsonLine(input, stripCarriageReturn(buffer), source);
    }
  }

  async #consumeStderr(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly attempt: WorkerAttemptProjection;
    readonly child: ManagedWorkerProcess;
    readonly state: MonitorState;
  }): Promise<void> {
    for await (const chunk of input.child.stderr) {
      input.state.outputBytes += chunk.byteLength;
      input.state.stderrTail = appendTail(
        input.state.stderrTail,
        chunk,
        WORKER_STDERR_TAIL_LIMIT_BYTES,
      );
      await this.#enforceObservedBudgets(input);
    }
  }

  async #consumeLines(
    initial: string,
    flush: boolean,
    consume: (line: string, rest: string) => Promise<void>,
  ): Promise<void> {
    let buffer = initial;
    while (true) {
      const index = buffer.indexOf("\n");
      if (index < 0) break;
      const line = stripCarriageReturn(buffer.slice(0, index));
      const rest = buffer.slice(index + 1);
      await consume(line, rest);
      buffer = rest;
    }
    if (flush && buffer.length > 0) await consume(stripCarriageReturn(buffer), "");
  }

  async #handleJsonLine(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly attempt: WorkerAttemptProjection;
    readonly child: ManagedWorkerProcess;
    readonly state: MonitorState;
    readonly resolveReady: () => void;
  }, line: string, source: "pi-output" | "worker-control"): Promise<void> {
    if (line.length === 0) return;
    if (Buffer.byteLength(line, "utf8") > WORKER_PROTOCOL_LINE_LIMIT_BYTES) {
      throw protocolOutputLimit();
    }
    let value: unknown;
    try {
      value = JSON.parse(line) as unknown;
    } catch (error) {
      throw new WorkerProtocolError(
        "protocol_invalid_json",
        `Worker protocol line is invalid JSON: ${toError(error).message}`,
      );
    }
    if (!isRecord(value)) throw new Error("Worker protocol line must be a JSON object");
    const type = value["type"];
    if (source === "pi-output") {
      if (type === "message_end") await this.#handleAssistantMessageEnd(input, value);
      return;
    }
    if (typeof type !== "string" || !type.startsWith("pi_amq_worker_")) {
      throw new Error("Worker control channel accepts only pi_amq worker envelopes");
    }
    if (value["type"] === "pi_amq_worker_refused") {
      const diagnostic = String(value["error"] ?? "unknown reason");
      throw new ModelFallbackEligibleError(
        `launch-validation: ${diagnostic}`,
        `Child launch refused: ${diagnostic}`,
      );
    }
    if (value["type"] === "pi_amq_worker_provider_failure") {
      assertProtocolIdentity(value, input.worker, input.attempt, this.#supervisorGeneration);
      const reason = typeof value["reason"] === "string" ? value["reason"] : "";
      if (!/^provider-(auth|quota|capacity): /.test(reason)) {
        throw new Error("Worker provider-failure envelope has an invalid reason");
      }
      throw new ModelFallbackEligibleError(reason, `Provider failure: ${reason}`);
    }
    if (value["type"] === "pi_amq_worker_wait") {
      assertProtocolIdentity(value, input.worker, input.attempt, this.#supervisorGeneration);
      const reason = typeof value["reason"] === "string"
        ? boundedProtocolText(value["reason"], "worker wait reason", 2_048)
        : "";
      const correlation = typeof value["correlation"] === "string"
        ? boundedProtocolText(value["correlation"], "worker wait correlation", 256)
        : "";
      await this.#store.recordWorkerWaitDeclared(input.requestKey, {
        workerId: input.worker.workerId,
        attemptId: input.attempt.attemptId,
        workerAttemptGeneration: input.attempt.workerAttemptGeneration,
        reason,
        correlation,
      });
      return;
    }
    if (value["type"] === "pi_amq_worker_peer_message") {
      assertProtocolIdentity(value, input.worker, input.attempt, this.#supervisorGeneration);
      const target = typeof value["target"] === "string" ? nonEmpty(value["target"], "peer target") : "";
      const message = typeof value["message"] === "string" ? nonEmpty(value["message"], "peer message") : "";
      const correlation = typeof value["correlation"] === "string"
        ? nonEmpty(value["correlation"], "peer correlation")
        : "";
      if (message.length > 32_768 || target.length > 256 || correlation.length > 256) {
        throw new Error("Peer message exceeds its bounded protocol fields");
      }
      await this.#routePeerMessage(input.requestKey, input.worker.workerId, target, message, correlation);
      return;
    }
    if (value["type"] === "pi_amq_worker_ready") {
      assertReadyEnvelope(value, input.worker, input.attempt, this.#supervisorGeneration);
      const sessionFile = value["sessionFile"];
      if (typeof sessionFile !== "string") {
        throw new Error("Worker ready envelope requires a session file");
      }
      await this.#store.recordWorkerReady(input.requestKey, {
        workerId: input.worker.workerId,
        attemptId: input.attempt.attemptId,
        workerAttemptGeneration: input.attempt.workerAttemptGeneration,
        sessionFile,
      });
      input.state.ready = true;
      input.resolveReady();
      return;
    }
    if (value["type"] === "pi_amq_worker_settled") {
      assertProtocolIdentity(value, input.worker, input.attempt, this.#supervisorGeneration);
      const managed = this.#managed.get(input.worker.workerId);
      if (managed?.attemptId === input.attempt.attemptId) {
        managed.inferenceActive = false;
        const revisionDelivered = await this.#deliverNextRevision(managed);
        if (!revisionDelivered) await this.#deliverNextPeerMessage(managed);
      }
      if (input.worker.contract.channel === "json-result") {
        await input.child.closeParentLiveness();
      }
    }
  }

  async #handleAssistantMessageEnd(
    input: {
      readonly requestKey: string;
      readonly worker: WorkerProjection;
      readonly attempt: WorkerAttemptProjection;
      readonly state: MonitorState;
    },
    value: Record<string, unknown>,
  ): Promise<void> {
    const message = isRecord(value["message"]) ? value["message"] : undefined;
    if (!message || message["role"] !== "assistant") return;
    const usage = isRecord(message["usage"]) ? message["usage"] : undefined;
    const tokens = usage?.["totalTokens"];
    if (typeof tokens === "number" && Number.isFinite(tokens) && tokens >= 0) {
      input.state.tokens += Math.floor(tokens);
    }
    const cost = isRecord(usage?.["cost"]) ? usage["cost"]["total"] : undefined;
    if (typeof cost === "number" && Number.isFinite(cost) && cost >= 0) {
      input.state.costUsd += cost;
    }
    const text = assistantText(message["content"]);
    if (text !== null) {
      input.state.candidateValue = text;
      input.state.candidateHash = sha256(text);
    }
    await this.#store.recordAttemptUsage(input.requestKey, {
      workerId: input.worker.workerId,
      attemptId: input.attempt.attemptId,
      workerAttemptGeneration: input.attempt.workerAttemptGeneration,
      tokens: input.state.tokens,
      costUsd: input.state.costUsd,
      outputBytes: input.state.outputBytes,
    });
    const providerFailure = providerFallbackReason(message);
    if (providerFailure) {
      throw new ModelFallbackEligibleError(providerFailure, `Provider failure: ${providerFailure}`);
    }
  }

  async #routePeerMessage(
    requestKey: string,
    senderWorkerId: WorkerId,
    target: string,
    message: string,
    correlation: string,
  ): Promise<void> {
    const projection = await this.#requiredRun(requestKey);
    const sender = requiredWorker(projection, senderWorkerId);
    const grant = `message_worker:${target}`;
    if (!effectiveWorkerContract(sender).capabilities.includes(grant)) {
      throw new Error(`Peer message target is not directionally granted: ${target}`);
    }
    const candidates = target.startsWith("worker_")
      ? [projection.workers[target as WorkerId]].filter((worker): worker is WorkerProjection => worker !== undefined)
      : Object.values(projection.workers).filter((worker) =>
          effectiveWorkerContract(worker).role?.name === target);
    if (candidates.length !== 1) {
      throw new Error(`Peer message target ${target} resolved to ${candidates.length} workers`);
    }
    const recipient = candidates[0]!;
    if (recipient.workerId === senderWorkerId) throw new Error("Peer message cannot target its sender");
    const recipientAttempt = currentAttempt(recipient);
    const managed = this.#managed.get(recipient.workerId);
    if (!managed
      || managed.attemptId !== recipientAttempt.attemptId
      || managed.workerAttemptGeneration !== recipientAttempt.workerAttemptGeneration) {
      throw new Error(`Peer message recipient ${recipient.workerId} has no current attached process`);
    }
    const requested = await this.#store.recordPeerMessageRequested(requestKey, {
      senderWorkerId,
      recipientWorkerId: recipient.workerId,
      grant,
      correlation,
      messageHash: sha256(message),
    });
    const pending = {
      messageId: requested.message.messageId,
      senderWorkerId,
      message,
      correlation,
    };
    if (managed.inferenceActive) {
      managed.pendingPeerMessages.push(pending);
      return;
    }
    await this.#deliverPeerMessage(managed, pending);
  }

  async #deliverNextPeerMessage(managed: ManagedAttempt): Promise<boolean> {
    const projection = await this.#requiredRun(managed.requestKey);
    const worker = requiredWorker(projection, managed.workerId);
    const index = managed.pendingPeerMessages.findIndex((peer) =>
      correlationCanWake(worker, peer.correlation));
    if (index < 0) return false;
    const [next] = managed.pendingPeerMessages.splice(index, 1);
    if (!next) return false;
    await this.#deliverPeerMessage(managed, next);
    return true;
  }

  async #deliverPeerMessage(
    managed: ManagedAttempt,
    peer: ManagedAttempt["pendingPeerMessages"][number],
  ): Promise<void> {
    const prompt = [
      "<peer-message>",
      JSON.stringify({
        senderWorkerId: peer.senderWorkerId,
        correlation: peer.correlation,
        message: peer.message,
      }),
      "</peer-message>",
    ].join("\n");
    try {
      await managed.child.writeLine(JSON.stringify({
        id: `peer-${peer.messageId}`,
        type: "prompt",
        message: prompt,
      }));
      managed.inferenceActive = true;
      await this.#store.recordPeerMessageDelivery(managed.requestKey, {
        messageId: peer.messageId,
        recipientWorkerId: managed.workerId,
        recipientAttemptId: managed.attemptId,
        recipientWorkerAttemptGeneration: managed.workerAttemptGeneration,
        status: "delivered",
      });
    } catch (error) {
      await this.#store.recordPeerMessageDelivery(managed.requestKey, {
        messageId: peer.messageId,
        recipientWorkerId: managed.workerId,
        recipientAttemptId: managed.attemptId,
        recipientWorkerAttemptGeneration: managed.workerAttemptGeneration,
        status: "indeterminate",
        diagnostic: toError(error).message,
      }).catch(() => {});
      throw error;
    }
  }

  async #deliverNextRevision(managed: ManagedAttempt): Promise<boolean> {
    const next = managed.pendingRevisions[0];
    if (!next) return false;
    const projection = await this.#requiredRun(managed.requestKey);
    const worker = requiredWorker(projection, managed.workerId);
    const revision = worker.contractRevisions.find(
      (candidate) => candidate.revisionId === next.revisionId,
    );
    if (!revision) throw new Error(`Unknown queued contract revision ${next.revisionId}`);
    if (!correlationCanWake(worker, revision.correlation)) return false;
    managed.pendingRevisions.shift();
    await managed.child.writeLine(JSON.stringify({
      id: `revision-${next.revisionId}`,
      type: "prompt",
      message: next.prompt,
    }));
    managed.inferenceActive = true;
    await this.#recordRevisionDelivered(managed, next.revisionId);
    return true;
  }

  async #recordRevisionDelivered(
    managed: ManagedAttempt,
    revisionId: ContractRevisionId,
  ): Promise<RunProjectionV2> {
    try {
      return await this.#store.recordContractRevisionDelivered(managed.requestKey, {
        workerId: managed.workerId,
        revisionId,
        boundary: "next-inference",
      });
    } catch (error) {
      await this.#store.recordWorkerNeedsAttention(managed.requestKey, {
        workerId: managed.workerId,
        attemptId: managed.attemptId,
        workerAttemptGeneration: managed.workerAttemptGeneration,
        reason: `Contract revision reached the worker but ledger delivery is uncertain: ${toError(error).message}`,
        ambiguous: true,
      }).catch(() => {});
      throw error;
    }
  }

  async #enforceObservedBudgets(input: {
    readonly requestKey: string;
    readonly worker: WorkerProjection;
    readonly state: MonitorState;
  }): Promise<void> {
    const budget = input.worker.contract.budget;
    const tokenExceeded = budgetExceeded(input.state.tokens, budget.maxTokens);
    const costExceeded = budgetExceeded(input.state.costUsd, budget.maxCostUsd);
    const outputExceeded = budgetExceeded(input.state.outputBytes, budget.maxOutputBytes);
    if (!tokenExceeded && !costExceeded && !outputExceeded) return;
    const limit = (value: BudgetLimit): string =>
      value.kind === "unlimited" ? "unlimited" : String(value.value);
    const reason = tokenExceeded
      ? `worker token budget exhausted: used ${input.state.tokens} of ${limit(budget.maxTokens)}`
      : costExceeded
        ? `worker cost budget exhausted: used ${input.state.costUsd} of ${limit(budget.maxCostUsd)}`
        : `worker output-byte budget exhausted: used ${input.state.outputBytes} of `
          + `${limit(budget.maxOutputBytes)}. ${OUTPUT_BYTE_UNIT_NOTE}`;
    let current = await this.#requiredRun(input.requestKey);
    let attempt = currentAttempt(requiredWorker(current, input.worker.workerId));
    current = await this.#store.recordAttemptUsage(input.requestKey, {
      workerId: input.worker.workerId,
      attemptId: attempt.attemptId,
      workerAttemptGeneration: attempt.workerAttemptGeneration,
      tokens: input.state.tokens,
      costUsd: input.state.costUsd,
      outputBytes: input.state.outputBytes,
    });
    attempt = currentAttempt(requiredWorker(current, input.worker.workerId));
    if (attempt.cancellation.requestedAt === null) {
      await this.cancelWorker(input.requestKey, {
        workerId: input.worker.workerId,
        reason,
      });
    }
  }

  async #inspectAttempt(attempt: WorkerAttemptProjection): Promise<"live" | "dead" | "unknown"> {
    const pid = attempt.pidStartEvidence.pid;
    if (pid === null) return "unknown";
    const inspection = await this.#inspector.inspect(pid);
    // A PID lookup is lease/liveness evidence, not an authoritative child-exit event.
    // Absence can race platform process accounting and must stay fail-closed.
    if (inspection.kind === "absent") return "unknown";
    if (inspection.kind === "unknown") return "unknown";
    return processIdentityMatchesAttempt(inspection.identity, attempt)
      ? "live"
      : "dead";
  }

  async #continuationSource(run: RunProjectionV2, workerId: WorkerId): Promise<{
    readonly priorWorkerId: WorkerId;
    readonly priorAttemptId: AttemptId;
    readonly priorLeaseGeneration: string;
    readonly sessionPath: string;
    readonly sessionFile: string;
  }> {
    const worker = requiredWorker(run, workerId);
    const attempt = currentAttempt(worker);
    if (worker.state !== "done" && worker.state !== "failed" && worker.state !== "cancelled") {
      throw new ResumeRefusedError(
        `Continuation prior worker is not terminal: ${worker.state}`,
      );
    }
    if (attempt.pidStartEvidence.observedExitAt === null
      || attempt.pidStartEvidence.exitStatus === null
      || attempt.pidStartEvidence.pid === null) {
      throw new ResumeRefusedError("Continuation prior process death is not durably observed");
    }
    const inspection = await this.#inspector.inspect(attempt.pidStartEvidence.pid);
    if (inspection.kind === "unknown") {
      throw new ResumeRefusedError(
        `Continuation refused because prior liveness is unknown: ${inspection.diagnostic}`,
      );
    }
    if (inspection.kind === "live" && processIdentityMatchesAttempt(inspection.identity, attempt)) {
      throw new ResumeRefusedError("Continuation refused because prior liveness is live");
    }
    if (attempt.session.file === null || attempt.session.leaseGeneration === null) {
      throw new ResumeRefusedError("Continuation prior session file and lease generation are required");
    }
    return {
      priorWorkerId: worker.workerId,
      priorAttemptId: attempt.attemptId,
      priorLeaseGeneration: attempt.session.leaseGeneration,
      sessionPath: attempt.session.path,
      sessionFile: attempt.session.file,
    };
  }

  async #requiredRun(requestKey: string): Promise<RunProjectionV2> {
    const projection = await this.#store.readRun(requestKey);
    if (!projection) throw new Error(`Unknown run request ${requestKey}`);
    return projection;
  }

  #sessionDirectory(runId: string, workerRequestKey: string): string {
    return join(
      this.#namespaceDirectory,
      "worker-sessions",
      sha256(`${runId}\0${workerRequestKey}`).slice("sha256:".length),
    );
  }
}

export class NodePiProcessFactory implements PiProcessFactory {
  async spawn(command: PiProcessLaunchCommand): Promise<ManagedWorkerProcess> {
    const child = spawnChild(command.executable, [...command.args], {
      cwd: command.cwd,
      env: { ...command.env },
      shell: false,
      detached: process.platform !== "win32",
      stdio: ["pipe", "pipe", "pipe", "pipe", "pipe"],
    });
    const exited = new Promise<ProcessExit>((resolve) => {
      child.once("exit", (code, signal) => resolve({ code, signal }));
    });
    await waitForSpawn(child);
    const protocol = child.stdio[command.protocolFd];
    if (child.pid === undefined
      || child.stdout === null
      || child.stderr === null
      || child.stdin === null
      || protocol === null
      || protocol === undefined
      || !(Symbol.asyncIterator in protocol)) {
      throw new Error("Pi child did not expose the required process pipes");
    }
    const pid = child.pid;
    return {
      pid,
      stdout: child.stdout,
      stderr: child.stderr,
      protocol: protocol as AsyncIterable<Uint8Array>,
      exited,
      writeLine: async (line) => writeToChild(child, `${line}\n`),
      endInput: async () => endChildInput(child),
      signalProcessGroup: async (signal) => signalGroup(child, pid, signal),
      closeParentLiveness: async () => { child.stdio[command.parentLivenessFd]?.destroy(); },
    };
  }
}

class NodeWorkerTimerScheduler implements WorkerTimerScheduler {
  schedule(delayMs: number, callback: () => void): { cancel(): void } {
    const timer = setTimeout(callback, Math.max(0, delayMs));
    timer.unref();
    return { cancel: () => clearTimeout(timer) };
  }
}

async function waitForSpawn(child: ChildProcess): Promise<void> {
  await new Promise<void>((resolve, reject) => {
    const onSpawn = (): void => { cleanup(); resolve(); };
    const onError = (error: Error): void => { cleanup(); reject(error); };
    const cleanup = (): void => {
      child.off("spawn", onSpawn);
      child.off("error", onError);
    };
    child.once("spawn", onSpawn);
    child.once("error", onError);
  });
}

async function writeToChild(child: ChildProcess, value: string): Promise<void> {
  if (!child.stdin || child.stdin.destroyed) throw new Error("Pi child stdin is unavailable");
  await new Promise<void>((resolve, reject) => {
    child.stdin!.write(value, "utf8", (error) => error ? reject(error) : resolve());
  });
}

async function endChildInput(child: ChildProcess): Promise<void> {
  const stdin = child.stdin;
  if (!stdin || stdin.destroyed || stdin.writableEnded) return;
  await new Promise<void>((resolve, reject) => {
    const onError = (error: Error): void => { cleanup(); reject(error); };
    const onFinish = (): void => { cleanup(); resolve(); };
    const cleanup = (): void => {
      stdin.off("error", onError);
      stdin.off("finish", onFinish);
    };
    stdin.once("error", onError);
    stdin.once("finish", onFinish);
    stdin.end();
  });
}

async function signalGroup(child: ChildProcess, pid: number, signal: NodeJS.Signals): Promise<void> {
  try {
    if (process.platform === "win32") child.kill(signal);
    else process.kill(-pid, signal);
  } catch (error) {
    if (!isNodeError(error) || error.code !== "ESRCH") throw error;
  }
}

function assertReadyEnvelope(
  value: Record<string, unknown>,
  worker: WorkerProjection,
  attempt: WorkerAttemptProjection,
  supervisorGeneration: string,
): void {
  assertProtocolIdentity(value, worker, attempt, supervisorGeneration);
}

function assertProtocolIdentity(
  value: Record<string, unknown>,
  worker: WorkerProjection,
  attempt: WorkerAttemptProjection,
  supervisorGeneration: string,
): void {
  if (value["runId"] !== worker.contract.runId
    || value["workerId"] !== worker.workerId
    || value["attemptId"] !== attempt.attemptId
    || value["workerAttemptGeneration"] !== attempt.workerAttemptGeneration
    || value["supervisorGeneration"] !== supervisorGeneration) {
    throw new Error("Worker protocol envelope generation or identity mismatch");
  }
}

function assistantText(value: unknown): string | null {
  if (typeof value === "string") return value;
  if (!Array.isArray(value)) return null;
  const text = value.flatMap((item) =>
    isRecord(item) && item["type"] === "text" && typeof item["text"] === "string"
      ? [item["text"]]
      : []).join("");
  return text.length === 0 ? null : text;
}

function providerFallbackReason(message: Record<string, unknown>): string | null {
  return classifyProviderFallbackFailure(message["stopReason"], message["errorMessage"]);
}

function currentAttempt(worker: WorkerProjection): WorkerAttemptProjection {
  const attempt = worker.attempts.at(-1);
  if (!attempt) throw new Error(`Worker ${worker.workerId} has no attempt`);
  return attempt;
}

function processIdentityMatchesAttempt(
  identity: ProcessIdentity,
  attempt: WorkerAttemptProjection,
): boolean {
  const recorded = attempt.pidStartEvidence;
  return identity.processStartIdentity === recorded.processStartIdentity
    && identity.bootIdentity === recorded.bootIdentity
    && identity.executable === recorded.executable
    && identity.hostname === recorded.hostname;
}

function requiredWorker(projection: RunProjectionV2, workerId: WorkerId): WorkerProjection {
  const worker = projection.workers[workerId];
  if (!worker) throw new Error(`Unknown worker ${workerId}`);
  return worker;
}

function formatExit(exit: ProcessExit): string {
  return exit.signal ?? `code ${exit.code ?? "unknown"}`;
}

function stripCarriageReturn(line: string): string {
  return line.endsWith("\r") ? line.slice(0, -1) : line;
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function isNodeError(error: unknown): error is NodeJS.ErrnoException {
  return error instanceof Error && "code" in error;
}

function toError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error));
}

function protocolOutputLimit(): WorkerProtocolError {
  return new WorkerProtocolError(
    "protocol_output_limit",
    `Worker protocol JSONL line exceeds ${WORKER_PROTOCOL_LINE_LIMIT_BYTES} bytes`,
    WORKER_PROTOCOL_LINE_LIMIT_BYTES,
  );
}

function protocolErrorDetails(error: Error): string {
  return error instanceof WorkerProtocolError
    ? JSON.stringify({
        type: "protocol_error",
        code: error.code,
        message: error.message,
        ...(error.limitBytes === undefined ? {} : { limitBytes: error.limitBytes }),
      })
    : error.message;
}

function appendTail(current: Uint8Array, chunk: Uint8Array, limit: number): Uint8Array {
  if (chunk.byteLength >= limit) return chunk.slice(chunk.byteLength - limit);
  const keep = Math.min(current.byteLength, limit - chunk.byteLength);
  const result = new Uint8Array(keep + chunk.byteLength);
  result.set(current.slice(current.byteLength - keep), 0);
  result.set(chunk, keep);
  return result;
}

function nonEmpty(value: string, name: string): string {
  if (value.length === 0) throw new TypeError(`${name} must not be empty`);
  return value;
}

function boundedProtocolText(value: string, name: string, maxLength: number): string {
  if (value.trim().length === 0) throw new TypeError(`${name} must not be empty`);
  if (value.length > maxLength) throw new TypeError(`${name} exceeds ${maxLength} characters`);
  return value;
}

function correlationCanWake(worker: WorkerProjection, correlation: string | null): boolean {
  if (worker.state === "running") return true;
  if (worker.state !== "blocked") return false;
  const active = worker.waits.filter((wait) => wait.status === "waiting");
  return active.length === 1
    && correlation !== null
    && active[0]!.correlation === correlation;
}

function assertBlockedWakeCorrelation(
  worker: WorkerProjection,
  correlation: string | null,
): void {
  if (!correlationCanWake(worker, correlation)) {
    const expected = worker.waits.filter((wait) => wait.status === "waiting").at(-1)?.correlation;
    throw new Error(
      `Input correlation does not match typed wait${expected ? ` ${expected}` : ""}`,
    );
  }
}