import { randomUUID as nodeRandomUUID } from "node:crypto";
import {
  chmod,
  mkdir,
  mkdtemp,
  readdir,
  rename,
  rm,
  stat,
} from "node:fs/promises";
import { isAbsolute, join, normalize, relative, sep } from "node:path";

import {
  canonicalJson,
  isNodeError,
  readWorkflowJson,
  sha256,
  syncWorkflowDirectory,
  writeWorkflowJson,
  type WorkflowAtomicFaultInjector,
} from "./atomic-storage.js";
import type {
  AmbiguityId,
  AmbiguityProjection,
  AttemptId,
  AcceptanceCheck,
  ContentRef,
  ContractRevisionId,
  EvidenceNode,
  ModelSpec,
  ObjectiveContract,
  PeerMessageId,
  PeerMessageProjection,
  RunBudget,
  RunBudgetProjection,
  RunId,
  RunProjectionV2,
  RunSource,
  ReportSchemaV1,
  ReportSchemaPolicy,
  StepId,
  WorkerAttemptProjection,
  WorkerBudget,
  WorkerContract,
  WorkerContinuationProvenance,
  WorkerId,
  WorkerProjection,
  WorkerProposalProvenance,
  WorkerRequest,
  WorkerRoleContract,
  WorkerLaunchContract,
  WorkerLaunchContractV2,
  WorkspaceGrant,
} from "./workflow-types.js";
import { parseWorkerLaunchContract, parseWorkerLaunchContractV2 } from "./worker-launch.js";
import {
  UNLIMITED,
  budgetExceeded,
  budgetExhausted,
  budgetValue,
  parseBudgetLimit,
  remainingBudget,
  reserveBudget,
} from "./budget.js";
import {
  classifyObservedTermination,
  evaluateWorkerAttemptTransition,
  normalizeWorkerAttemptState,
  type WorkerAttemptState,
  type WorkerGateFacts,
} from "./worker-gate.js";
import { evaluateRunTransition, type RunGateFacts } from "./run-gate.js";
import { WorkflowLeaseManager, type ProcessIdentity } from "./workflow-lease.js";

const EVENT_KEYS = [
  "schema",
  "eventId",
  "seq",
  "runId",
  "type",
  "supervisorGeneration",
  "occurredAt",
  "previousHash",
  "payload",
  "checksum",
] as const;

type WorkflowEventType =
  | "run-opened"
  | "worker-spawn-requested"
  | "attempt-spawn-intent"
  | "attempt-start-observed"
  | "worker-ready-observed"
  | "attempt-usage-observed"
  | "cancel-requested"
  | "cancel-signal-sent"
  | "attempt-exit-observed"
  | "worker-needs-attention"
  | "contract-revision-requested"
  | "contract-revision-delivered"
  | "collect-observed"
  | "silent-stall-recorded"
  | "gate-evaluated"
  | "worker-wait-declared"
  | "steer-delivered"
  | "peer-message-requested"
  | "peer-message-delivery-observed"
  | "ambiguity-recorded"
  | "reconciliation-recorded";

interface WorkflowLedgerEventV1 {
  readonly schema: 1;
  readonly eventId: string;
  readonly seq: number;
  readonly runId: RunId;
  readonly type: WorkflowEventType;
  readonly supervisorGeneration: string;
  readonly occurredAt: string;
  readonly previousHash: string | null;
  readonly payload: Record<string, unknown>;
  readonly checksum: string;
}

export interface StartRunInput {
  readonly requestKey: string;
  readonly source: RunSource;
  readonly objective: ObjectiveContract;
}

export type StartRunResult =
  | { readonly kind: "created"; readonly projection: RunProjectionV2 }
  | { readonly kind: "existing"; readonly projection: RunProjectionV2 };

export interface WorkflowRunStoreOptions {
  readonly namespaceDirectory: string;
  readonly leaseManager: WorkflowLeaseManager;
  readonly supervisorGeneration: string;
  readonly randomUUID?: () => string;
  readonly now?: () => Date;
  readonly atomicFault?: WorkflowAtomicFaultInjector;
}

export type RunRecordDiagnostic =
  | {
      readonly kind: "healthy";
      readonly directory: string;
      readonly projection: RunProjectionV2;
    }
  | {
      readonly kind: "corrupt";
      readonly directory: string;
      readonly diagnostic: string;
    };

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

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

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

export interface PrepareWorkerIntentResult {
  readonly kind: "created" | "existing";
  readonly projection: RunProjectionV2;
  readonly worker: WorkerProjection;
}

export interface PrepareAttemptResult {
  readonly projection: RunProjectionV2;
  readonly worker: WorkerProjection;
  readonly attempt: WorkerAttemptProjection;
}

export class WorkflowRunStore {
  readonly #namespaceDirectory: string;
  readonly #leaseManager: WorkflowLeaseManager;
  readonly #supervisorGeneration: string;
  readonly #randomUUID: () => string;
  readonly #now: () => Date;
  readonly #atomicFault: WorkflowAtomicFaultInjector | undefined;
  readonly #listeners = new Set<(projection: RunProjectionV2) => void>();

  constructor(options: WorkflowRunStoreOptions) {
    if (!isAbsolute(options.namespaceDirectory)) {
      throw new TypeError("Workflow run-store namespace must be absolute");
    }
    this.#namespaceDirectory = normalize(options.namespaceDirectory);
    this.#leaseManager = options.leaseManager;
    this.#supervisorGeneration = requireNonEmpty(
      options.supervisorGeneration,
      "supervisor generation",
    );
    this.#randomUUID = options.randomUUID ?? nodeRandomUUID;
    this.#now = options.now ?? (() => new Date());
    this.#atomicFault = options.atomicFault;
  }

  async startRun(input: StartRunInput): Promise<StartRunResult> {
    const requestKey = validateRequestKey(input.requestKey);
    const source = parseRunSource(input.source);
    const objective = parseObjectiveContract(input.objective);
    const requestFingerprint = fingerprint({ source, objective });
    return this.#leaseManager.withCurrentSupervisor(this.#supervisorGeneration, async () => {
      await this.#initialize();
      const existing = await this.readRun(requestKey);
      if (existing) return existingRun(existing, requestFingerprint);

      const runId = this.#id("run") as RunId;
      const occurredAt = this.#now().toISOString();
      const event = createEvent({
        eventId: this.#id("event"),
        seq: 0,
        runId,
        type: "run-opened",
        supervisorGeneration: this.#supervisorGeneration,
        occurredAt,
        previousHash: null,
        payload: {
          requestKey,
          requestFingerprint,
          source,
          objective,
        },
      });
      const projection = foldEvents([event]);
      const installed = await this.#installInitialRun(requestKey, event, projection);
      if (installed) {
        this.#notify(projection);
        return { kind: "created", projection };
      }
      const raced = await this.readRun(requestKey);
      if (!raced) throw new LedgerIntegrityError("Run directory appeared without a readable ledger");
      return existingRun(raced, requestFingerprint);
    });
  }

  subscribe(listener: (projection: RunProjectionV2) => void): () => void {
    this.#listeners.add(listener);
    return () => this.#listeners.delete(listener);
  }

  async readRun(requestKey: string): Promise<RunProjectionV2 | undefined> {
    validateRequestKey(requestKey);
    const directory = runDirectoryForRequest(this.#namespaceDirectory, requestKey);
    try {
      await stat(directory);
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return undefined;
      throw error;
    }
    // snapshot.json is deliberately not read here: the immutable event chain is authority.
    const projection = foldEvents(await readEventChain(directory));
    if (projection.requestKey !== requestKey) {
      throw new LedgerIntegrityError("Run request-key hash collision or directory mismatch");
    }
    return projection;
  }

  async listRuns(): Promise<readonly RunProjectionV2[]> {
    const byRequest = this.#byRequestDirectory();
    let entries: string[];
    try {
      entries = await readdir(byRequest);
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return [];
      throw error;
    }
    const runs: RunProjectionV2[] = [];
    for (const entry of entries.sort()) {
      if (entry.startsWith(".")) continue;
      const directory = join(byRequest, entry);
      const chain = await readEventChain(directory);
      runs.push(foldEvents(chain));
    }
    return runs.sort((left, right) => left.runId.localeCompare(right.runId));
  }

  async diagnoseRunRecords(): Promise<readonly RunRecordDiagnostic[]> {
    const byRequest = this.#byRequestDirectory();
    let entries: string[];
    try {
      entries = await readdir(byRequest);
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return [];
      throw error;
    }
    const records: RunRecordDiagnostic[] = [];
    for (const entry of entries.sort()) {
      if (entry.startsWith(".")) continue;
      const directory = join(byRequest, entry);
      try {
        const projection = foldEvents(await readEventChain(directory));
        if (runDirectoryForRequest(this.#namespaceDirectory, projection.requestKey) !== directory) {
          throw new LedgerIntegrityError("Run request-key hash does not match its directory");
        }
        records.push({ kind: "healthy", directory, projection });
      } catch (error) {
        records.push({
          kind: "corrupt",
          directory,
          diagnostic: error instanceof Error ? error.message : String(error),
        });
      }
    }
    return records;
  }

  async recordWorkerIntent(inputRequestKey: string, input: {
    readonly createdByDriverSessionId: string;
    readonly sessionPath: string;
    readonly request: WorkerRequest;
  }): Promise<RunProjectionV2> {
    return (await this.prepareWorkerIntent(inputRequestKey, {
      ...input,
      parentWorkerId: null,
      depth: 1,
    })).projection;
  }

  async prepareWorkerIntent(inputRequestKey: string, input: {
    readonly createdByDriverSessionId: string;
    readonly sessionPath: string;
    readonly request: WorkerRequest;
    readonly parentWorkerId: WorkerId | null;
    readonly depth: number;
    readonly continuation?: WorkerContinuationProvenance;
  }): Promise<PrepareWorkerIntentResult> {
    // This state-only seam records an atomic call that already occurred. It does not create a
    // process or schedule future work.
    const requestKey = validateRequestKey(inputRequestKey);
    const createdByDriverSessionId = requireNonEmpty(
      input.createdByDriverSessionId,
      "driver session ID",
    );
    if (!isAbsolute(input.sessionPath)) throw new TypeError("Worker session path must be absolute");
    const sessionPath = normalize(input.sessionPath);
    const request = parseWorkerRequest(input.request);
    const depth = positiveIntegerValue(input.depth, "worker depth");
    const parentWorkerId = input.parentWorkerId;
    const continuation = input.continuation === undefined
      ? null
      : parseWorkerContinuationProvenance(input.continuation);
    const contractInputFingerprint = fingerprint({
      request,
      createdByDriverSessionId,
      sessionPath,
      parentWorkerId,
      depth,
      ...(continuation ? { continuation } : {}),
    });
    let disposition: "created" | "existing" = "existing";
    const projection = await this.#withRunMutation(requestKey, async (current, events) => {
      const existing = Object.values(current.workers).find(
        (worker) => worker.spawnRequestKey === request.requestKey,
      );
      if (existing) {
        if (existing.contractFingerprint !== contractInputFingerprint) {
          throw new RunConflictError(
            `Worker request key ${request.requestKey} already has a different contract`,
          );
        }
        return { kind: "existing" as const, projection: current };
      }
      assertWorkerCapabilitySubset(current, request);
      assertSpawnBudget(current, request.budget, depth, this.#now());
      if (parentWorkerId !== null && !current.workers[parentWorkerId]) {
        throw new TypeError(`Parent worker ${parentWorkerId} does not exist in this run`);
      }
      if (continuation !== null) {
        assertContinuationSource(current, continuation, sessionPath);
      }
      const workerId = this.#id("worker") as WorkerId;
      const stepId = this.#id("step") as StepId;
      const attemptId = this.#id("attempt") as AttemptId;
      const contract: WorkerContract = {
        ...request,
        context: request.context ?? "fresh",
        runId: current.runId,
        workerId,
        createdByDriverSessionId,
      };
      disposition = "created";
      const event = this.#nextEvent(current, "worker-spawn-requested", {
        stepId,
        workerId,
        contractFingerprint: contractInputFingerprint,
        contract,
        continuation,
        parentWorkerId,
        depth,
        attempt: initialAttempt(
          attemptId,
          1,
          this.#randomUUID(),
          this.#supervisorGeneration,
          sessionPath,
          request.sideEffectPolicy,
          request.model,
          null,
        ),
      });
      return { kind: "append" as const, event, events };
    });
    const worker = Object.values(projection.workers).find(
      (candidate) => candidate.spawnRequestKey === request.requestKey,
    );
    if (!worker) throw new LedgerIntegrityError("Worker intent committed without a projection");
    return { kind: disposition, projection, worker };
  }

  async prepareReplacementAttempt(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly model?: ModelSpec;
    readonly fallbackReason?: string;
  }): Promise<PrepareAttemptResult> {
    const requestKey = validateRequestKey(inputRequestKey);
    const workerId = requireNonEmpty(input.workerId, "worker ID") as WorkerId;
    const projection = await this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, workerId);
      const previous = requiredCurrentAttempt(worker);
      if (isAttemptProcessActive(previous)) {
        throw new Error(`Worker ${workerId} still has an active attempt`);
      }
      if (previous.state !== "failed" && previous.state !== "cancelled") {
        throw new Error(`Worker ${workerId} cannot retry from ${previous.state}`);
      }
      if (budgetExhausted(worker.attempts.length, worker.contract.budget.maxAttempts)) {
        throw new BudgetExceededError(`Worker ${workerId} exhausted its attempt budget`);
      }
      assertReplacementBudget(current, worker.contract.budget, worker.depth, this.#now());
      const model = input.model ?? previous.modelSelection?.model ?? worker.contract.model;
      if (![worker.contract.model, ...(worker.contract.modelFallbacks ?? [])].some(
        (candidate) => candidate.provider === model.provider && candidate.id === model.id,
      )) {
        throw new TypeError(`Replacement model is outside the immutable fallback chain: ${model.provider}/${model.id}`);
      }
      const attempt = initialAttempt(
        this.#id("attempt") as AttemptId,
        worker.attempts.length + 1,
        this.#randomUUID(),
        this.#supervisorGeneration,
        previous.session.path,
        worker.contract.sideEffectPolicy,
        model,
        input.fallbackReason ?? null,
      );
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "attempt-spawn-intent", { workerId, attempt }),
      };
    });
    const worker = requiredWorker(projection, workerId);
    return { projection, worker, attempt: requiredCurrentAttempt(worker) };
  }

  async recordAttemptStarted(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly leaseGeneration: string;
    readonly processIdentity: ProcessIdentity;
    /** Write side is schema 2 only; schema 1 stays readable history and never launches a child. */
    readonly launchContract: WorkerLaunchContractV2;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (attempt.state === "running") {
        if (attempt.pidStartEvidence.pid !== input.processIdentity.pid) {
          throw new RunConflictError(`Attempt ${input.attemptId} already has different PID evidence`);
        }
        return { kind: "existing" as const, projection: current };
      }
      if (attempt.state !== "pending") {
        throw new Error(`Attempt ${input.attemptId} cannot start from ${attempt.state}`);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "attempt-start-observed", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          leaseGeneration: requireNonEmpty(input.leaseGeneration, "lease generation"),
          processIdentity: parseProcessIdentity(input.processIdentity),
          launchContract: parseWorkerLaunchContractV2(input.launchContract),
        }),
      };
    });
  }

  async recordWorkerReady(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly sessionFile: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    if (!isAbsolute(input.sessionFile)) {
      throw new TypeError("Worker session file must be absolute");
    }
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { worker, attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      const sessionFile = normalize(input.sessionFile);
      const sessionRelative = relative(attempt.session.path, sessionFile);
      if (sessionRelative.length === 0
        || sessionRelative === ".."
        || sessionRelative.startsWith(`..${sep}`)
        || isAbsolute(sessionRelative)) {
        throw new TypeError("Worker session file must be inside its leased session directory");
      }
      if (worker.continuation !== null && worker.continuation.sessionFile !== sessionFile) {
        throw new RunConflictError("Continued worker did not open the proven prior session file");
      }
      if (attempt.session.file === sessionFile) return { kind: "existing" as const, projection: current };
      if (attempt.session.file !== null) {
        throw new RunConflictError(`Attempt ${input.attemptId} already has a different session file`);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "worker-ready-observed", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          sessionFile,
        }),
      };
    });
  }

  async recordAttemptUsage(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly tokens: number;
    readonly costUsd: number;
    readonly outputBytes: number;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const usage = parseUsage({
      tokens: input.tokens,
      costUsd: input.costUsd,
      outputBytes: input.outputBytes,
    });
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (usage.tokens < attempt.usage.tokens
        || usage.costUsd < attempt.usage.costUsd
        || usage.outputBytes < attempt.usage.outputBytes) {
        throw new TypeError("Attempt usage must be monotonic");
      }
      if (canonicalJson(usage) === canonicalJson(attempt.usage)) {
        return { kind: "existing" as const, projection: current };
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "attempt-usage-observed", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          usage,
        }),
      };
    });
  }

  async requestWorkerCancellation(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly reason: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const reason = requireNonEmpty(input.reason, "cancellation reason");
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      const attempt = requiredCurrentAttempt(worker);
      if (attempt.cancellation.requestedAt !== null) {
        if (attempt.cancellation.reason !== reason) {
          throw new RunConflictError(`Worker ${input.workerId} already has a different cancellation intent`);
        }
        return { kind: "existing" as const, projection: current };
      }
      if (!isAttemptProcessActive(attempt)) {
        throw new Error(`Worker ${input.workerId} has no active attempt to cancel`);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "cancel-requested", {
          workerId: input.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          reason,
        }),
      };
    });
  }

  async recordCancellationSignalSent(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly signal: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (attempt.cancellation.requestedAt === null) {
        throw new Error("Cancellation signal cannot precede durable cancellation intent");
      }
      if (attempt.cancellation.sentAt !== null) return { kind: "existing" as const, projection: current };
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "cancel-signal-sent", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          signal: requireNonEmpty(input.signal, "cancellation signal"),
        }),
      };
    });
  }

  async recordAttemptExit(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly exitStatus: { readonly code: number | null; readonly signal: string | null };
    readonly candidateOutcome: { readonly valueOrRef: string | null; readonly hash: string | null };
    readonly uncertainSideEffects: boolean;
    readonly reason: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const exitStatus = parseExitStatus(input.exitStatus);
    const candidateOutcome = parseCandidateOutcome(input.candidateOutcome);
    const reason = requireNonEmpty(input.reason, "attempt exit reason");
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (attempt.pidStartEvidence.observedExitAt !== null) {
        return { kind: "existing" as const, projection: current };
      }
      const ambiguityId = input.uncertainSideEffects
        ? (this.#id("ambiguity") as AmbiguityId)
        : null;
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "attempt-exit-observed", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          exitStatus,
          candidateOutcome,
          uncertainSideEffects: input.uncertainSideEffects,
          reason,
          ambiguityId,
        }),
      };
    });
  }

  async recordWorkerNeedsAttention(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly reason: string;
    readonly ambiguous: boolean;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const reason = requireNonEmpty(input.reason, "worker attention reason");
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { worker, attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (worker.state === "ambiguous"
        || (!input.ambiguous && attempt.control.status === "detached")) {
        return { kind: "existing" as const, projection: current };
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "worker-needs-attention", {
          workerId: input.workerId,
          attemptId: input.attemptId,
          workerAttemptGeneration: input.workerAttemptGeneration,
          reason,
          ambiguous: input.ambiguous,
          ambiguityId: input.ambiguous ? (this.#id("ambiguity") as AmbiguityId) : null,
        }),
      };
    });
  }

  async requestContractRevision(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly correlation?: string;
    readonly goal?: string;
    readonly acceptance?: readonly string[];
    readonly constraints?: readonly string[];
    readonly report_schema?: ReportSchemaV1 | null;
    readonly report_schema_policy?: ReportSchemaPolicy;
    readonly requires_human_approval?: boolean;
    readonly acceptance_check?: AcceptanceCheck | null;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const correlation = input.correlation === undefined
      ? null
      : requireBoundedText(input.correlation, "contract revision correlation", 256);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      if (worker.state === "done" || worker.state === "failed" || worker.state === "cancelled") {
        throw new Error(`Worker ${input.workerId} contract is terminal`);
      }
      if (worker.state === "blocked") {
        requiredMatchingWait(worker, correlation, false);
      }
      const previous = latestWorkerContract(worker);
      const revisedRequest = parseWorkerRequest({
        ...previous,
        goal: input.goal ?? previous.goal,
        acceptance: input.acceptance ?? previous.acceptance,
        constraints: input.constraints ?? previous.constraints,
        ...optionalRevisionField("report_schema", input, previous),
        ...optionalRevisionField("acceptance_check", input, previous),
        ...(input.report_schema_policy === undefined
          ? (previous.report_schema_policy === undefined ? {} : { report_schema_policy: previous.report_schema_policy })
          : { report_schema_policy: input.report_schema_policy }),
        ...(input.requires_human_approval === undefined
          ? (previous.requires_human_approval === undefined
              ? {}
              : { requires_human_approval: previous.requires_human_approval })
          : { requires_human_approval: input.requires_human_approval }),
      }, true);
      const contract: WorkerContract = {
        ...revisedRequest,
        context: revisedRequest.context ?? "fresh",
        runId: worker.contract.runId,
        workerId: worker.workerId,
        createdByDriverSessionId: worker.contract.createdByDriverSessionId,
      };
      if (canonicalJson(contract) === canonicalJson(previous)) {
        throw new TypeError("Contract revision must change at least one field");
      }
      const revisionId = this.#id("revision") as ContractRevisionId;
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "contract-revision-requested", {
          workerId: worker.workerId,
          revisionId,
          revision: worker.contractRevisions.length + 1,
          contract,
          correlation,
        }),
      };
    });
  }

  async recordContractRevisionDelivered(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly revisionId: ContractRevisionId;
    readonly boundary: "next-inference";
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      const revision = worker.contractRevisions.find(
        (candidate) => candidate.revisionId === input.revisionId,
      );
      if (!revision) throw new Error(`Unknown contract revision ${input.revisionId}`);
      if (revision.status === "delivered") return { kind: "existing" as const, projection: current };
      if (worker.contractRevisions.some(
        (candidate) => candidate.revision < revision.revision && candidate.status !== "delivered",
      )) {
        throw new Error("Contract revisions must be delivered in order");
      }
      if (input.boundary !== "next-inference") {
        throw new TypeError("Contract revisions may be delivered only at the next inference boundary");
      }
      if (worker.state === "blocked") {
        requiredMatchingWait(worker, revision.correlation, false);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "contract-revision-delivered", {
          workerId: worker.workerId,
          revisionId: revision.revisionId,
          boundary: input.boundary,
          correlation: revision.correlation,
        }),
      };
    });
  }

  async recordCollectObservation(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly structuredPayload: unknown;
    readonly consecutiveRounds: number;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const consecutiveRounds = positiveIntegerValue(
      input.consecutiveRounds,
      "no-progress consecutive rounds",
    );
    if (consecutiveRounds < 2) throw new TypeError("No-progress threshold must be at least 2");
    const payloadFingerprint = fingerprint(input.structuredPayload);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      if (worker.state !== "running") {
        throw new Error(`No-progress observation requires running worker; ${input.workerId} is ${worker.state}`);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "collect-observed", {
          workerId: input.workerId,
          payloadFingerprint,
          consecutiveRounds,
        }),
      };
    });
  }

  async recordSilentStall(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      if (worker.state !== "running") {
        throw new Error(`Silent-stall observation requires running worker; ${input.workerId} is ${worker.state}`);
      }
      if (worker.stall.silent.detected) {
        return { kind: "existing" as const, projection: current };
      }
      const expectedCadenceMs = expectedWorkerCadenceMs(effectiveWorkerContract(worker).budget);
      const silentForMs = this.#now().getTime() - new Date(worker.lastLedgerAt).getTime();
      if (silentForMs < expectedCadenceMs) {
        throw new Error(
          `Worker ${input.workerId} has been silent for ${silentForMs}ms; expected cadence is ${expectedCadenceMs}ms`,
        );
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "silent-stall-recorded", {
          workerId: input.workerId,
          expectedCadenceMs,
        }),
      };
    });
  }

  async recordGateEvaluation(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly reportMatches: boolean | null;
    readonly acceptanceCheckExitCode: number | null;
    readonly humanApproved: boolean | null;
    readonly evidenceExists?: boolean;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      const attempt = requiredCurrentAttempt(worker);
      const contract = effectiveWorkerContract(worker);
      const facts: WorkerGateFacts = {
        evidence: {
          exists: input.evidenceExists ?? (
            attempt.candidateOutcome.valueOrRef !== null
            || attempt.candidateOutcome.hash !== null
          ),
        },
        report: {
          schemaDeclared: contract.report_schema !== undefined,
          matches: input.reportMatches ?? false,
          policy: contract.report_schema_policy ?? "preferred",
        },
        acceptanceCheck: {
          declared: contract.acceptance_check !== undefined,
          exitCode: input.acceptanceCheckExitCode,
        },
        human: {
          approvalRequired: contract.requires_human_approval === true,
          approved: input.humanApproved === true,
        },
        budget: { exceeded: workerBudgetExceeded(attempt, contract.budget) },
        runtime: {
          supervisorLost: false,
          effectsUncertain: attempt.sideEffects.status === "uncertain",
        },
        reconcile: { decisionExists: false },
        input: { kind: null, correlationMatches: false },
        cancel: { verbReceived: false },
      };
      const result = evaluateWorkerAttemptTransition(attempt.state, "done", facts);
      const blockers = result.allowed ? [] : result.blockers;
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "gate-evaluated", {
          workerId: worker.workerId,
          attemptId: attempt.attemptId,
          fromState: attempt.state,
          targetState: "done",
          evidenceExists: facts.evidence.exists,
          reportMatches: input.reportMatches,
          reportSchemaPolicy: contract.report_schema === undefined ? null : (contract.report_schema_policy ?? "preferred"),
          acceptanceCheckExitCode: input.acceptanceCheckExitCode,
          humanApproved: input.humanApproved,
          budgetExceeded: facts.budget.exceeded,
          allowed: result.allowed,
          blockers,
        }),
      };
    });
  }

  async recordSteerDelivered(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly messageHash: string;
    readonly correlation?: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const messageHash = requireNonEmpty(input.messageHash, "steer message hash");
    const correlation = input.correlation === undefined
      ? null
      : requireBoundedText(input.correlation, "steer correlation", 256);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const worker = requiredWorker(current, input.workerId);
      if (worker.state === "blocked") requiredMatchingWait(worker, correlation, false);
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "steer-delivered", {
          workerId: input.workerId,
          messageHash,
          correlation,
        }),
      };
    });
  }

  async recordWorkerWaitDeclared(inputRequestKey: string, input: {
    readonly workerId: WorkerId;
    readonly attemptId: AttemptId;
    readonly workerAttemptGeneration: string;
    readonly reason: string;
    readonly correlation: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const reason = requireBoundedText(input.reason, "worker wait reason", 2_048);
    const correlation = requireBoundedText(input.correlation, "worker wait correlation", 256);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const { worker, attempt } = requiredWorkerAttempt(current, input.workerId, input.attemptId);
      assertAttemptGeneration(attempt, input.workerAttemptGeneration, this.#supervisorGeneration);
      if (effectiveWorkerContract(worker).channel !== "rpc") {
        throw new TypeError("Only an RPC worker can declare a typed wait");
      }
      if (worker.state !== "running" || attempt.state !== "running") {
        throw new RunConflictError(`Typed wait requires running worker; ${worker.workerId} is ${worker.state}`);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "worker-wait-declared", {
          workerId: worker.workerId,
          attemptId: attempt.attemptId,
          workerAttemptGeneration: attempt.workerAttemptGeneration,
          supervisorGeneration: attempt.supervisorGeneration,
          reason,
          correlation,
        }),
      };
    });
  }

  async recordPeerMessageRequested(inputRequestKey: string, input: {
    readonly senderWorkerId: WorkerId;
    readonly recipientWorkerId: WorkerId;
    readonly grant: string;
    readonly correlation: string;
    readonly messageHash: string;
  }): Promise<{ readonly projection: RunProjectionV2; readonly message: PeerMessageProjection }> {
    const requestKey = validateRequestKey(inputRequestKey);
    const grant = requireNonEmpty(input.grant, "peer-message grant");
    const correlation = requireNonEmpty(input.correlation, "peer-message correlation");
    const messageHash = requireSha256(input.messageHash, "peer-message hash");
    const messageId = this.#id("peer") as PeerMessageId;
    const projection = await this.#withRunMutation(requestKey, async (current, events) => {
      const sender = requiredWorker(current, input.senderWorkerId);
      const recipient = requiredWorker(current, input.recipientWorkerId);
      if (sender.workerId === recipient.workerId) throw new TypeError("Peer message cannot target its sender");
      if (!effectiveWorkerContract(sender).capabilities.includes(grant)) {
        throw new TypeError(`Sender contract does not grant ${grant}`);
      }
      const grantedRecipient = resolvePeerGrantRecipient(current, grant, false);
      if (grantedRecipient.workerId !== recipient.workerId) {
        throw new TypeError(`Peer grant ${grant} does not resolve to ${recipient.workerId}`);
      }
      if (effectiveWorkerContract(sender).channel !== "rpc"
        || effectiveWorkerContract(recipient).channel !== "rpc") {
        throw new TypeError("Peer messaging requires RPC sender and recipient contracts");
      }
      if (recipient.state !== "running" && recipient.state !== "blocked") {
        throw new RunConflictError(
          `Peer-message recipient must be running or blocked; ${recipient.workerId} is ${recipient.state}`,
        );
      }
      if (recipient.state === "blocked") requiredMatchingWait(recipient, correlation, false);
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "peer-message-requested", {
          workerId: sender.workerId,
          messageId,
          senderWorkerId: sender.workerId,
          recipientWorkerId: recipient.workerId,
          grant,
          correlation,
          messageHash,
        }),
      };
    });
    const message = projection.workers[input.senderWorkerId]?.peerMessages.find(
      (candidate) => candidate.messageId === messageId,
    );
    if (!message) throw new LedgerIntegrityError("Peer message committed without sender projection");
    return { projection, message };
  }

  async recordPeerMessageDelivery(inputRequestKey: string, input: {
    readonly messageId: PeerMessageId;
    readonly recipientWorkerId: WorkerId;
    readonly recipientAttemptId: AttemptId;
    readonly recipientWorkerAttemptGeneration: string;
    readonly status: "delivered" | "indeterminate";
    readonly diagnostic?: string;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const diagnostic = input.diagnostic === undefined
      ? null
      : requireNonEmpty(input.diagnostic, "peer-message delivery diagnostic");
    return this.#withRunMutation(requestKey, async (current, events) => {
      const recipient = requiredWorker(current, input.recipientWorkerId);
      const attempt = requiredCurrentAttempt(recipient);
      if (attempt.attemptId !== input.recipientAttemptId
        || attempt.workerAttemptGeneration !== input.recipientWorkerAttemptGeneration) {
        throw new RunConflictError("Peer-message recipient generation is stale");
      }
      const message = recipient.peerMessages.find((candidate) => candidate.messageId === input.messageId);
      if (!message) throw new TypeError(`Unknown peer message ${input.messageId}`);
      if (message.status !== "pending") {
        if (message.status === input.status) return { kind: "existing" as const, projection: current };
        throw new RunConflictError(`Peer message ${input.messageId} is already ${message.status}`);
      }
      if (input.status === "delivered" && recipient.state === "blocked") {
        requiredMatchingWait(recipient, message.correlation, false);
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "peer-message-delivery-observed", {
          workerId: message.senderWorkerId,
          messageId: message.messageId,
          senderWorkerId: message.senderWorkerId,
          recipientWorkerId: message.recipientWorkerId,
          recipientAttemptId: input.recipientAttemptId,
          recipientWorkerAttemptGeneration: input.recipientWorkerAttemptGeneration,
          status: input.status,
          diagnostic,
        }),
      };
    });
  }

  async markAmbiguous(inputRequestKey: string, input: {
    readonly ambiguityId?: AmbiguityId;
    readonly reason: string;
    readonly relatedIds: readonly string[];
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const reason = requireNonEmpty(input.reason, "ambiguity reason");
    const relatedIds = input.relatedIds.map((id) => requireNonEmpty(id, "related ID"));
    const ambiguityId = input.ambiguityId ?? (this.#id("ambiguity") as AmbiguityId);
    return this.#withRunMutation(requestKey, async (current, events) => {
      const existing = current.unresolvedAmbiguities.find(
        (ambiguity) => ambiguity.ambiguityId === ambiguityId,
      );
      if (existing) {
        if (existing.reason !== reason || canonicalJson(existing.relatedIds) !== canonicalJson(relatedIds)) {
          throw new RunConflictError(`Ambiguity ${ambiguityId} already has different evidence`);
        }
        return { kind: "existing" as const, projection: current };
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "ambiguity-recorded", {
          ambiguityId,
          reason,
          relatedIds,
          status: "needs-reconcile",
        }),
      };
    });
  }

  async recordReconciliation(inputRequestKey: string, input: {
    readonly ambiguityId: AmbiguityId;
    readonly decision: string;
    readonly evidenceRefs: readonly string[];
    readonly workerState?: Exclude<WorkerAttemptProjection["state"], "ambiguous">;
  }): Promise<RunProjectionV2> {
    const requestKey = validateRequestKey(inputRequestKey);
    const ambiguityId = requireNonEmpty(input.ambiguityId, "ambiguity ID") as AmbiguityId;
    const decision = requireNonEmpty(input.decision, "reconciliation decision");
    const evidenceRefs = input.evidenceRefs.map((ref) => requireNonEmpty(ref, "evidence reference"));
    return this.#withRunMutation(requestKey, async (current, events) => {
      const ambiguity = current.unresolvedAmbiguities.find(
        (ambiguity) => ambiguity.ambiguityId === ambiguityId,
      );
      if (!ambiguity) {
        const existing = events.find((event) => event.type === "reconciliation-recorded"
          && event.payload["ambiguityId"] === ambiguityId);
        if (existing
          && existing.payload["decision"] === decision
          && canonicalJson(existing.payload["evidenceRefs"]) === canonicalJson(evidenceRefs)
          && (existing.payload["workerState"] ?? null) === (input.workerState ?? null)) {
          return { kind: "existing" as const, projection: current };
        }
        throw new Error(`Ambiguity ${ambiguityId} is not awaiting reconciliation`);
      }
      const relatedWorker = Object.values(current.workers).find(
        (worker) => ambiguity.relatedIds.includes(worker.workerId),
      );
      const relatedAttempt = relatedWorker?.attempts.find(
        (attempt) => ambiguity.relatedIds.includes(attempt.attemptId),
      );
      if (relatedWorker && relatedAttempt?.state === "ambiguous") {
        if (input.workerState === undefined) {
          throw new TypeError("Worker ambiguity reconciliation requires an explicit workerState");
        }
        if (input.workerState === "blocked") {
          const activeWaits = relatedWorker.waits.filter((wait) =>
            wait.status === "waiting" && wait.attemptId === relatedAttempt.attemptId);
          if (activeWaits.length !== 1) {
            throw new TypeError("Reconciliation to blocked requires the attempt's existing typed wait");
          }
        }
        const contract = effectiveWorkerContract(relatedWorker);
        const result = evaluateWorkerAttemptTransition(
          "ambiguous",
          input.workerState,
          {
            evidence: { exists: relatedWorker.gate.evidenceExists },
            report: {
              schemaDeclared: contract.report_schema !== undefined,
              matches: relatedWorker.gate.reportMatches === true,
              policy: contract.report_schema_policy ?? "preferred",
            },
            acceptanceCheck: {
              declared: contract.acceptance_check !== undefined,
              exitCode: relatedWorker.gate.acceptanceCheckExitCode,
            },
            human: {
              approvalRequired: contract.requires_human_approval === true,
              approved: relatedWorker.gate.humanApproved === true,
            },
            budget: { exceeded: workerBudgetExceeded(relatedAttempt, contract.budget) },
            runtime: { supervisorLost: false, effectsUncertain: false, certain: true },
            failure: { observed: input.workerState === "failed" },
            termination: { observed: input.workerState === "cancelled" },
            wait: { declared: input.workerState === "blocked" },
            reconcile: { decisionExists: true, uncertaintyCleared: true },
            input: { kind: null, correlationMatches: false },
            cancel: { verbReceived: relatedAttempt.cancellation.requestedAt !== null },
          },
        );
        if (!result.allowed) {
          throw new Error(`Reconciled worker transition is blocked: ${result.blockers.join(", ")}`);
        }
      } else if (input.workerState !== undefined) {
        throw new TypeError("workerState is valid only for a related ambiguous worker attempt");
      }
      return {
        kind: "append" as const,
        events,
        event: this.#nextEvent(current, "reconciliation-recorded", {
          ambiguityId,
          decision,
          evidenceRefs,
          workerId: relatedWorker?.workerId ?? null,
          attemptId: relatedAttempt?.attemptId ?? null,
          workerState: input.workerState ?? null,
        }),
      };
    });
  }

  async #withRunMutation(
    requestKey: string,
    decide: (
      current: RunProjectionV2,
      events: readonly WorkflowLedgerEventV1[],
    ) => Promise<
      | { readonly kind: "existing"; readonly projection: RunProjectionV2 }
      | {
          readonly kind: "append";
          readonly event: WorkflowLedgerEventV1;
          readonly events: readonly WorkflowLedgerEventV1[];
        }
    >,
  ): Promise<RunProjectionV2> {
    return this.#leaseManager.withCurrentSupervisor(this.#supervisorGeneration, async () => {
      const directory = runDirectoryForRequest(this.#namespaceDirectory, requestKey);
      const events = await readEventChain(directory);
      const current = foldEvents(events);
      const decision = await decide(current, events);
      if (decision.kind === "existing") return decision.projection;
      await writeWorkflowJson(
        eventPath(directory, decision.event),
        decision.event,
        {
          createOnly: true,
          ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
        },
      );
      const projection = foldEvents([...decision.events, decision.event]);
      await writeWorkflowJson(join(directory, "snapshot.json"), projection, {
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      this.#notify(projection);
      return projection;
    });
  }

  #notify(projection: RunProjectionV2): void {
    for (const listener of this.#listeners) {
      try {
        listener(projection);
      } catch {
        // Projection consumers are advisory and cannot roll back a committed ledger event.
      }
    }
  }

  #nextEvent(
    current: RunProjectionV2,
    type: Exclude<WorkflowEventType, "run-opened">,
    payload: Record<string, unknown>,
  ): WorkflowLedgerEventV1 {
    return createEvent({
      eventId: this.#id("event"),
      seq: current.ledger.throughSeq + 1,
      runId: current.runId,
      type,
      supervisorGeneration: this.#supervisorGeneration,
      occurredAt: this.#now().toISOString(),
      previousHash: current.ledger.chainHash,
      payload,
    });
  }

  async #installInitialRun(
    requestKey: string,
    event: WorkflowLedgerEventV1,
    projection: RunProjectionV2,
  ): Promise<boolean> {
    const parent = this.#byRequestDirectory();
    const destination = runDirectoryForRequest(this.#namespaceDirectory, requestKey);
    const prepared = await mkdtemp(join(parent, ".run.prepare-"));
    await chmod(prepared, 0o700);
    let installed = false;
    try {
      const events = join(prepared, "events");
      await mkdir(events, { mode: 0o700 });
      await writeWorkflowJson(eventPath(prepared, event), event, {
        createOnly: true,
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      await syncWorkflowDirectory(events);
      await writeWorkflowJson(join(prepared, "snapshot.json"), projection, {
        createOnly: true,
        ...(this.#atomicFault ? { fault: this.#atomicFault } : {}),
      });
      await syncWorkflowDirectory(prepared);
      try {
        await rename(prepared, destination);
      } catch (error) {
        if (isNodeError(error) && (error.code === "EEXIST" || error.code === "ENOTEMPTY")) {
          return false;
        }
        throw error;
      }
      installed = true;
      await syncWorkflowDirectory(parent);
      return true;
    } finally {
      if (!installed) await rm(prepared, { recursive: true, force: true });
    }
  }

  async #initialize(): Promise<void> {
    await mkdir(this.#namespaceDirectory, { recursive: true, mode: 0o700 });
    await mkdir(join(this.#namespaceDirectory, "runs"), { recursive: true, mode: 0o700 });
    await mkdir(this.#byRequestDirectory(), { recursive: true, mode: 0o700 });
  }

  #byRequestDirectory(): string {
    return join(this.#namespaceDirectory, "runs", "by-request");
  }

  #id(prefix: "run" | "event" | "worker" | "step" | "attempt" | "ambiguity" | "revision" | "peer"): string {
    return `${prefix}_${requireNonEmpty(this.#randomUUID(), `${prefix} UUID`)}`;
  }
}

export function runDirectoryForRequest(namespaceDirectory: string, requestKey: string): string {
  if (!isAbsolute(namespaceDirectory)) throw new TypeError("Workflow namespace must be absolute");
  const key = validateRequestKey(requestKey);
  return join(
    normalize(namespaceDirectory),
    "runs",
    "by-request",
    sha256(key).slice("sha256:".length),
  );
}

async function readEventChain(runDirectory: string): Promise<readonly WorkflowLedgerEventV1[]> {
  let files: string[];
  try {
    files = (await readdir(join(runDirectory, "events")))
      .filter((entry) => entry.endsWith(".json") && !entry.startsWith("."))
      .sort();
  } catch (error) {
    if (isNodeError(error) && error.code === "ENOENT") {
      throw new LedgerIntegrityError("Run ledger events directory is missing");
    }
    throw error;
  }
  if (files.length === 0) throw new LedgerIntegrityError("Run ledger is empty");
  const events: WorkflowLedgerEventV1[] = [];
  let expectedPrevious: string | null = null;
  for (let index = 0; index < files.length; index += 1) {
    const file = files[index]!;
    const raw = await readWorkflowJson(join(runDirectory, "events", file));
    if (raw === undefined) throw new LedgerIntegrityError(`Run event ${file} disappeared during recovery`);
    const event = parseEvent(raw);
    if (event.seq !== index) {
      throw new LedgerIntegrityError(`Run ledger gap: expected seq ${index}, got ${event.seq}`);
    }
    if (!file.startsWith(`${formatSequence(event.seq)}.${safePathPart(event.eventId)}.`)) {
      throw new LedgerIntegrityError(`Run event filename does not match seq/event ID: ${file}`);
    }
    if (event.previousHash !== expectedPrevious) {
      throw new LedgerIntegrityError(`Run ledger chain diverged at seq ${event.seq}`);
    }
    expectedPrevious = event.checksum;
    events.push(event);
  }
  return events;
}

function foldEvents(events: readonly WorkflowLedgerEventV1[]): RunProjectionV2 {
  let projection: RunProjectionV2 | undefined;
  for (const event of events) {
    if (!projection) {
      if (event.seq !== 0 || event.type !== "run-opened" || event.previousHash !== null) {
        throw new LedgerIntegrityError("Run ledger must begin with run-opened at seq 0");
      }
      const payload = parseRunOpenedPayload(event.payload);
      projection = {
        schema: 2,
        runId: event.runId,
        requestKey: payload.requestKey,
        requestFingerprint: payload.requestFingerprint,
        source: payload.source,
        objective: payload.objective,
        budget: initialBudgetProjection(payload.objective.budget),
        state: "open",
        workers: {},
        unresolvedAmbiguities: [],
        ledger: { throughSeq: 0, chainHash: event.checksum },
        createdAt: event.occurredAt,
        updatedAt: event.occurredAt,
      };
      continue;
    }
    if (event.runId !== projection.runId) {
      throw new LedgerIntegrityError(`Run ID changed at seq ${event.seq}`);
    }
    if (event.type === "run-opened") throw new LedgerIntegrityError("run-opened may occur only once");
    switch (event.type) {
      case "worker-spawn-requested": {
        const payload = parseWorkerSpawnPayload(event.payload);
        if (projection.workers[payload.workerId]) {
          throw new LedgerIntegrityError(`Duplicate worker ID ${payload.workerId}`);
        }
        const worker: WorkerProjection = {
          workerId: payload.workerId,
          spawnRequestKey: payload.contract.requestKey,
          atomicCallStepId: payload.stepId,
          contractFingerprint: payload.contractFingerprint,
          contract: payload.contract,
          continuation: payload.continuation,
          parentWorkerId: payload.parentWorkerId,
          depth: payload.depth,
          state: "pending",
          lastLedgerAt: event.occurredAt,
          attempts: [payload.attempt],
          contractRevisions: [],
          peerMessages: [],
          waits: [],
          stall: {
            silent: { detected: false, expectedCadenceMs: null, recordedAt: null },
            noProgress: {
              detected: false,
              fingerprint: null,
              identicalRounds: 0,
              recordedAt: null,
            },
          },
          gate: {
            evidenceExists: false,
            reportMatches: null,
            reportSchemaPolicy: null,
            acceptanceCheckExitCode: null,
            humanApproved: null,
            blockers: [],
          },
          acceptedOutcome: null,
        };
        projection = reserveProjectionBudget(
          { ...projection, workers: { ...projection.workers, [worker.workerId]: worker } },
          worker.contract.budget,
          true,
        );
        break;
      }
      case "attempt-spawn-intent": {
        const payload = parseAttemptSpawnIntentPayload(event.payload);
        const worker = requiredWorker(projection, payload.workerId);
        if (worker.attempts.some((attempt) => attempt.attemptId === payload.attempt.attemptId)) {
          throw new LedgerIntegrityError(`Duplicate attempt ID ${payload.attempt.attemptId}`);
        }
        projection = reserveProjectionBudget(updateWorker(projection, payload.workerId, (current) => ({
          ...current,
          state: "pending",
          attempts: [...current.attempts, payload.attempt],
        })), worker.contract.budget, false);
        break;
      }
      case "attempt-start-observed": {
        const payload = parseAttemptStartPayload(event.payload);
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          return {
            ...attempt,
            state: gatedWorkerState(attempt.state, "running", workerGateFacts()),
            launchContract: payload.launchContract,
            session: { ...attempt.session, leaseGeneration: payload.leaseGeneration },
            pidStartEvidence: {
              pid: payload.processIdentity.pid,
              processStartIdentity: payload.processIdentity.processStartIdentity,
              bootIdentity: payload.processIdentity.bootIdentity,
              executable: payload.processIdentity.executable,
              hostname: payload.processIdentity.hostname,
              spawnedAt: event.occurredAt,
              observedExitAt: null,
              exitStatus: null,
            },
          };
        }, "running");
        break;
      }
      case "worker-ready-observed": {
        const payload = parseWorkerReadyPayload(event.payload);
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          return { ...attempt, session: { ...attempt.session, file: payload.sessionFile } };
        });
        break;
      }
      case "attempt-usage-observed": {
        const payload = parseAttemptUsagePayload(event.payload);
        const before = requiredWorkerAttempt(projection, payload.workerId, payload.attemptId).attempt.usage;
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          if (payload.usage.tokens < attempt.usage.tokens
            || payload.usage.costUsd < attempt.usage.costUsd
            || payload.usage.outputBytes < attempt.usage.outputBytes) {
            throw new LedgerIntegrityError("Attempt usage regressed in the ledger");
          }
          return { ...attempt, usage: payload.usage };
        });
        projection = spendProjectionBudget(projection, {
          tokens: payload.usage.tokens - before.tokens,
          costUsd: payload.usage.costUsd - before.costUsd,
          outputBytes: payload.usage.outputBytes - before.outputBytes,
        });
        break;
      }
      case "cancel-requested": {
        const payload = parseCancelRequestedPayload(event.payload);
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          return {
            ...attempt,
            cancellation: {
              ...attempt.cancellation,
              reason: payload.reason,
              requestedAt: event.occurredAt,
            },
          };
        });
        break;
      }
      case "cancel-signal-sent": {
        const payload = parseCancelSignalPayload(event.payload);
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          return {
            ...attempt,
            cancellation: { ...attempt.cancellation, sentAt: event.occurredAt },
          };
        });
        break;
      }
      case "attempt-exit-observed": {
        const payload = parseAttemptExitPayload(event.payload);
        const currentAttempt = requiredWorkerAttempt(
          projection,
          payload.workerId,
          payload.attemptId,
        ).attempt;
        assertFoldGeneration(
          currentAttempt,
          payload.workerAttemptGeneration,
          event.supervisorGeneration,
        );
        const wasCancelled = currentAttempt.cancellation.requestedAt !== null;
        const successful = payload.exitStatus.code === 0 && payload.exitStatus.signal === null;
        const workerState = classifyObservedTermination({
          runtimeCertain: !payload.uncertainSideEffects,
          effectsUncertain: payload.uncertainSideEffects,
          cancellationRequested: wasCancelled,
          terminationObserved: true,
          failureObserved: !successful && !wasCancelled,
        });
        const transitionFacts = workerGateFacts({
          runtimeCertain: !payload.uncertainSideEffects,
          effectsUncertain: payload.uncertainSideEffects,
          cancellationRequested: wasCancelled,
          terminationObserved: true,
          failureObserved: !successful && !wasCancelled,
        });
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => ({
          ...attempt,
          state: gatedWorkerState(attempt.state, workerState, transitionFacts),
          pidStartEvidence: {
            ...attempt.pidStartEvidence,
            observedExitAt: event.occurredAt,
            exitStatus: payload.exitStatus,
          },
          candidateOutcome: payload.candidateOutcome,
          sideEffects: {
            ...attempt.sideEffects,
            status: payload.uncertainSideEffects
              ? "uncertain"
              : attempt.sideEffects.policy === "none" ? "none" : "known",
          },
          cancellation: {
            ...attempt.cancellation,
            observedAt: wasCancelled ? event.occurredAt : attempt.cancellation.observedAt,
          },
        }), workerState);
        if (payload.ambiguityId !== null) {
          projection = appendAmbiguity(projection, {
            ambiguityId: payload.ambiguityId,
            status: "needs-reconcile",
            reason: payload.reason,
            relatedIds: [payload.workerId, payload.attemptId],
            recordedAt: event.occurredAt,
          });
        }
        break;
      }
      case "worker-needs-attention": {
        const payload = parseWorkerNeedsAttentionPayload(event.payload);
        projection = updateAttempt(projection, payload.workerId, payload.attemptId, (attempt) => {
          assertFoldGeneration(attempt, payload.workerAttemptGeneration, event.supervisorGeneration);
          return payload.ambiguous ? {
            ...attempt,
            state: gatedWorkerState(
              attempt.state,
              "ambiguous",
              workerGateFacts({ effectsUncertain: true, runtimeCertain: false }),
            ),
            sideEffects: { ...attempt.sideEffects, status: "uncertain" },
          } : {
            ...attempt,
            control: {
              status: "detached",
              reason: payload.reason,
              recordedAt: event.occurredAt,
            },
          };
        }, payload.ambiguous ? "ambiguous" : undefined);
        if (!payload.ambiguous && projection.state !== "ambiguous") {
          projection = withRunState(
            projection,
            "needs-attention",
            runGateFacts({ workersDetached: true }),
          );
        }
        if (payload.ambiguityId !== null) {
          projection = appendAmbiguity(projection, {
            ambiguityId: payload.ambiguityId,
            status: "needs-reconcile",
            reason: payload.reason,
            relatedIds: [payload.workerId, payload.attemptId],
            recordedAt: event.occurredAt,
          });
        }
        break;
      }
      case "contract-revision-requested": {
        const payload = parseContractRevisionRequestedPayload(event.payload);
        if (payload.contract.runId !== projection.runId) {
          throw new LedgerIntegrityError("Contract revision run ID mismatch");
        }
        projection = updateWorker(projection, payload.workerId, (worker) => {
          if (worker.state === "blocked") {
            requiredMatchingWait(worker, payload.correlation, true);
          }
          if (worker.contractRevisions.some(
            (revision) => revision.revisionId === payload.revisionId,
          )) {
            throw new LedgerIntegrityError(`Duplicate contract revision ${payload.revisionId}`);
          }
          if (payload.revision !== worker.contractRevisions.length + 1) {
            throw new LedgerIntegrityError("Contract revision sequence is not contiguous");
          }
          return {
            ...worker,
            contractRevisions: [
              ...worker.contractRevisions,
              {
                revisionId: payload.revisionId,
                revision: payload.revision,
                contract: payload.contract,
                status: "pending",
                correlation: payload.correlation,
                requestedAt: event.occurredAt,
                deliveredAt: null,
              },
            ],
          };
        });
        break;
      }
      case "contract-revision-delivered": {
        const payload = parseContractRevisionDeliveredPayload(event.payload);
        projection = updateWorker(projection, payload.workerId, (worker) => {
          const revision = worker.contractRevisions.find(
            (candidate) => candidate.revisionId === payload.revisionId,
          );
          if (!revision) throw new LedgerIntegrityError(`Unknown contract revision ${payload.revisionId}`);
          if (revision.correlation !== payload.correlation) {
            throw new LedgerIntegrityError("Contract revision delivery correlation mismatch");
          }
          const woken = wakeWorkerFromWait(
            worker,
            "revise_contract",
            payload.correlation,
            event.occurredAt,
          );
          return {
            ...woken,
            contractRevisions: worker.contractRevisions.map((revision) =>
              revision.revisionId === payload.revisionId
                ? { ...revision, status: "delivered" as const, deliveredAt: event.occurredAt }
                : revision),
          };
        });
        break;
      }
      case "worker-wait-declared": {
        const payload = parseWorkerWaitDeclaredPayload(event.payload);
        projection = updateWorker(projection, payload.workerId, (worker) => {
          const attempt = requiredCurrentAttempt(worker);
          assertFoldGeneration(
            attempt,
            payload.workerAttemptGeneration,
            payload.supervisorGeneration,
          );
          if (payload.supervisorGeneration !== event.supervisorGeneration) {
            throw new LedgerIntegrityError("Typed-wait supervisor generation mismatch");
          }
          if (attempt.attemptId !== payload.attemptId) {
            throw new LedgerIntegrityError("Typed-wait attempt is not current");
          }
          if (effectiveWorkerContract(worker).channel !== "rpc") {
            throw new LedgerIntegrityError("Typed wait requires an RPC contract");
          }
          if (worker.waits.some((wait) => wait.status === "waiting")) {
            throw new LedgerIntegrityError("Worker already has an active typed wait");
          }
          return {
            ...worker,
            state: gatedWorkerState(
              worker.state,
              "blocked",
              workerGateFacts({ waitDeclared: true }),
            ),
            attempts: worker.attempts.map((candidate) => candidate.attemptId === attempt.attemptId
              ? {
                  ...candidate,
                  state: gatedWorkerState(
                    candidate.state,
                    "blocked",
                    workerGateFacts({ waitDeclared: true }),
                  ),
                }
              : candidate),
            waits: [...worker.waits, {
              attemptId: attempt.attemptId,
              workerAttemptGeneration: attempt.workerAttemptGeneration,
              reason: payload.reason,
              correlation: payload.correlation,
              status: "waiting" as const,
              declaredAt: event.occurredAt,
              satisfiedAt: null,
              satisfiedBy: null,
            }],
          };
        });
        break;
      }
      case "collect-observed": {
        const payload = parseCollectObservedPayload(event.payload);
        projection = updateWorker(projection, payload.workerId, (worker) => {
          const previous = worker.stall.noProgress;
          const identicalRounds = previous.fingerprint === payload.payloadFingerprint
            ? previous.identicalRounds + 1
            : 1;
          const detected = identicalRounds >= payload.consecutiveRounds;
          return {
            ...worker,
            stall: {
              ...worker.stall,
              noProgress: {
                detected,
                fingerprint: payload.payloadFingerprint,
                identicalRounds,
                recordedAt: detected ? event.occurredAt : null,
              },
            },
          };
        });
        break;
      }
      case "silent-stall-recorded": {
        const payload = parseSilentStallPayload(event.payload);
        projection = updateWorker(projection, payload.workerId, (worker) => ({
          ...worker,
          stall: {
            ...worker.stall,
            silent: {
              detected: true,
              expectedCadenceMs: payload.expectedCadenceMs,
              recordedAt: event.occurredAt,
            },
          },
        }));
        break;
      }
      case "gate-evaluated": {
        const payload = parseGateEvaluatedPayload(event.payload);
        projection = updateAttempt(
          projection,
          payload.workerId,
          payload.attemptId,
          (attempt) => {
            if (attempt.state !== payload.fromState) {
              throw new LedgerIntegrityError("Gate source state does not match projection");
            }
            return payload.allowed ? {
              ...attempt,
              state: gatedWorkerState(
                attempt.state,
                "done",
                workerGateFacts({
                  evidenceExists: payload.evidenceExists,
                  reportSchemaDeclared: payload.reportMatches !== null,
                  reportMatches: payload.reportMatches === true,
                  ...(payload.reportSchemaPolicy === null ? {} : { reportSchemaPolicy: payload.reportSchemaPolicy }),
                  acceptanceDeclared: payload.acceptanceCheckExitCode !== null,
                  acceptanceExitCode: payload.acceptanceCheckExitCode,
                  humanRequired: payload.humanApproved !== null,
                  humanApproved: payload.humanApproved === true,
                  budgetExceeded: false,
                }),
              ),
            } : attempt;
          },
          payload.allowed ? "done" : undefined,
        );
        projection = updateWorker(projection, payload.workerId, (worker) => ({
          ...worker,
          gate: {
            evidenceExists: payload.evidenceExists,
            reportMatches: payload.reportMatches,
            reportSchemaPolicy: payload.reportSchemaPolicy,
            acceptanceCheckExitCode: payload.acceptanceCheckExitCode,
            humanApproved: payload.humanApproved,
            blockers: payload.blockers,
          },
        }));
        break;
      }
      case "steer-delivered": {
        const payload = parseSteerDeliveredPayload(event.payload);
        projection = updateWorker(projection, payload.workerId, (worker) =>
          wakeWorkerFromWait(worker, "steer", payload.correlation, event.occurredAt));
        break;
      }
      case "peer-message-requested": {
        const payload = parsePeerMessageRequestedPayload(event.payload);
        const sender = requiredWorker(projection, payload.senderWorkerId);
        const recipient = requiredWorker(projection, payload.recipientWorkerId);
        if (sender.workerId === recipient.workerId) {
          throw new LedgerIntegrityError("Peer message cannot target its sender");
        }
        if (!effectiveWorkerContract(sender).capabilities.includes(payload.grant)) {
          throw new LedgerIntegrityError(`Peer-message sender lacks grant ${payload.grant}`);
        }
        if (effectiveWorkerContract(sender).channel !== "rpc"
          || effectiveWorkerContract(recipient).channel !== "rpc") {
          throw new LedgerIntegrityError("Peer-message endpoints must both use RPC contracts");
        }
        if (recipient.state !== "running" && recipient.state !== "blocked") {
          throw new LedgerIntegrityError("Peer-message recipient is not running or blocked");
        }
        if (recipient.state === "blocked") {
          requiredMatchingWait(recipient, payload.correlation, true);
        }
        const grantedRecipient = resolvePeerGrantRecipient(projection, payload.grant, true);
        if (grantedRecipient.workerId !== recipient.workerId) {
          throw new LedgerIntegrityError(
            `Peer grant ${payload.grant} does not resolve to ${recipient.workerId}`,
          );
        }
        if (sender.peerMessages.some((message) => message.messageId === payload.messageId)
          || recipient.peerMessages.some((message) => message.messageId === payload.messageId)) {
          throw new LedgerIntegrityError(`Duplicate peer message ${payload.messageId}`);
        }
        const message: PeerMessageProjection = {
          messageId: payload.messageId,
          senderWorkerId: payload.senderWorkerId,
          recipientWorkerId: payload.recipientWorkerId,
          grant: payload.grant,
          correlation: payload.correlation,
          messageHash: payload.messageHash,
          status: "pending",
          requestedAt: event.occurredAt,
          deliveredAt: null,
          diagnostic: null,
        };
        projection = updateWorker(projection, sender.workerId, (worker) => ({
          ...worker,
          peerMessages: [...worker.peerMessages, message],
        }));
        projection = updateWorker(projection, recipient.workerId, (worker) => ({
          ...worker,
          lastLedgerAt: event.occurredAt,
          peerMessages: [...worker.peerMessages, message],
        }));
        break;
      }
      case "peer-message-delivery-observed": {
        const payload = parsePeerMessageDeliveryPayload(event.payload);
        const recipient = requiredWorker(projection, payload.recipientWorkerId);
        const attempt = requiredCurrentAttempt(recipient);
        if (attempt.attemptId !== payload.recipientAttemptId
          || attempt.workerAttemptGeneration !== payload.recipientWorkerAttemptGeneration) {
          throw new LedgerIntegrityError("Peer-message delivery recipient generation is stale");
        }
        const sender = requiredWorker(projection, payload.senderWorkerId);
        if (!sender.peerMessages.some((message) => message.messageId === payload.messageId)
          || !recipient.peerMessages.some((message) => message.messageId === payload.messageId)) {
          throw new LedgerIntegrityError(`Peer message ${payload.messageId} is absent from one endpoint`);
        }
        for (const workerId of [payload.senderWorkerId, payload.recipientWorkerId]) {
          projection = updateWorker(projection, workerId, (worker) => ({
            ...worker,
            ...(workerId === payload.recipientWorkerId ? { lastLedgerAt: event.occurredAt } : {}),
            peerMessages: worker.peerMessages.map((message) => message.messageId === payload.messageId
              ? {
                  ...message,
                  status: payload.status,
                  deliveredAt: payload.status === "delivered" ? event.occurredAt : null,
                  diagnostic: payload.diagnostic,
                }
              : message),
          }));
        }
        if (payload.status === "delivered") {
          projection = updateWorker(projection, payload.recipientWorkerId, (worker) =>
            wakeWorkerFromWait(worker, "peer_message", messageCorrelation(worker, payload.messageId), event.occurredAt));
        }
        break;
      }
      case "ambiguity-recorded": {
        const ambiguity = parseAmbiguityPayload(event.payload, event.occurredAt);
        if (projection.unresolvedAmbiguities.some((item) => item.ambiguityId === ambiguity.ambiguityId)) {
          throw new LedgerIntegrityError(`Duplicate ambiguity ID ${ambiguity.ambiguityId}`);
        }
        projection = {
          ...projection,
          state: gatedRunState(projection.state, "ambiguous", runGateFacts()),
          unresolvedAmbiguities: [...projection.unresolvedAmbiguities, ambiguity],
        };
        break;
      }
      case "reconciliation-recorded": {
        const payload = parseReconciliationPayload(event.payload);
        if (!projection.unresolvedAmbiguities.some(
          (item) => item.ambiguityId === payload.ambiguityId,
        )) {
          throw new LedgerIntegrityError(
            `Reconciliation references resolved or unknown ambiguity ${payload.ambiguityId}`,
          );
        }
        const unresolved = projection.unresolvedAmbiguities.filter(
          (item) => item.ambiguityId !== payload.ambiguityId,
        );
        projection = {
          ...projection,
          state: unresolved.length === 0
            ? gatedRunState(
                projection.state,
                "open",
                runGateFacts({ reconciliationRecorded: true, uncertaintyCleared: true }),
              )
            : "ambiguous",
          unresolvedAmbiguities: unresolved,
        };
        if (payload.workerId !== null && payload.attemptId !== null && payload.workerState !== null) {
          const reconciledWorker = requiredWorker(projection, payload.workerId);
          if (payload.workerState === "blocked") {
            const activeWaits = reconciledWorker.waits.filter((wait) =>
              wait.status === "waiting" && wait.attemptId === payload.attemptId);
            if (activeWaits.length !== 1) {
              throw new LedgerIntegrityError(
                "Reconciliation to blocked requires the attempt's existing typed wait",
              );
            }
          }
          projection = updateAttempt(
            projection,
            payload.workerId,
            payload.attemptId,
            (attempt) => {
              if (attempt.state !== "ambiguous") {
                throw new LedgerIntegrityError("Reconciled worker attempt is not ambiguous");
              }
              return {
                ...attempt,
                state: gatedWorkerState(
                  attempt.state,
                  payload.workerState!,
                  workerGateFacts({
                    reconciliation: true,
                    uncertaintyCleared: true,
                    waitDeclared: payload.workerState === "blocked",
                    failureObserved: payload.workerState === "failed",
                    cancellationRequested: payload.workerState === "cancelled",
                    terminationObserved: payload.workerState === "cancelled",
                  }),
                ),
                sideEffects: {
                  ...attempt.sideEffects,
                  status: attempt.sideEffects.policy === "none" ? "none" : "known",
                },
              };
            },
            payload.workerState,
          );
        }
        break;
      }
    }
    const workerId = event.payload["workerId"];
    if (typeof workerId === "string" && projection.workers[workerId]) {
      projection = updateWorker(projection, workerId as WorkerId, (worker) => ({
        ...worker,
        lastLedgerAt: event.occurredAt,
        ...(event.type === "silent-stall-recorded" ? {} : {
          stall: {
            ...worker.stall,
            silent: { detected: false, expectedCadenceMs: null, recordedAt: null },
          },
        }),
      }));
    }
    projection = {
      ...projection,
      budget: withRemainingWallClock(projection.budget, projection.createdAt, event.occurredAt),
      ledger: { throughSeq: event.seq, chainHash: event.checksum },
      updatedAt: event.occurredAt,
    };
  }
  if (!projection) throw new LedgerIntegrityError("Run ledger did not produce a projection");
  return projection;
}

function createEvent(input: Omit<WorkflowLedgerEventV1, "schema" | "checksum">): WorkflowLedgerEventV1 {
  const unsigned = { schema: 1 as const, ...input };
  return { ...unsigned, checksum: fingerprint(unsigned) };
}

function parseEvent(value: unknown): WorkflowLedgerEventV1 {
  const object = strictRecord(value, EVENT_KEYS, "workflow ledger event");
  if (numberValue(object, "schema") !== 1) throw new LedgerIntegrityError("Workflow event schema must be 1");
  const type = enumValue(object, "type", [
    "run-opened",
    "worker-spawn-requested",
    "attempt-spawn-intent",
    "attempt-start-observed",
    "worker-ready-observed",
    "attempt-usage-observed",
    "cancel-requested",
    "cancel-signal-sent",
    "attempt-exit-observed",
    "worker-needs-attention",
    "contract-revision-requested",
    "contract-revision-delivered",
    "collect-observed",
    "silent-stall-recorded",
    "gate-evaluated",
    "worker-wait-declared",
    "steer-delivered",
    "peer-message-requested",
    "peer-message-delivery-observed",
    "ambiguity-recorded",
    "reconciliation-recorded",
  ] as const);
  const event: WorkflowLedgerEventV1 = {
    schema: 1,
    eventId: nonEmptyString(object, "eventId"),
    seq: nonNegativeInteger(object, "seq"),
    runId: prefixedId(object, "runId", "run") as RunId,
    type,
    supervisorGeneration: nonEmptyString(object, "supervisorGeneration"),
    occurredAt: isoTimestamp(object, "occurredAt"),
    previousHash: nullableHash(object, "previousHash"),
    payload: recordValue(object, "payload"),
    checksum: hashValue(object, "checksum"),
  };
  const { checksum, ...unsigned } = event;
  if (fingerprint(unsigned) !== checksum) {
    throw new LedgerIntegrityError(`Workflow event checksum mismatch at seq ${event.seq}`);
  }
  return event;
}

function parseRunOpenedPayload(value: unknown): {
  readonly requestKey: string;
  readonly requestFingerprint: string;
  readonly source: RunSource;
  readonly objective: ObjectiveContract;
} {
  const object = strictRecord(
    value,
    ["requestKey", "requestFingerprint", "source", "objective"],
    "run-opened payload",
  );
  const payload = {
    requestKey: validateRequestKey(nonEmptyString(object, "requestKey")),
    requestFingerprint: hashValue(object, "requestFingerprint"),
    source: parseRunSource(object["source"]),
    objective: parseObjectiveContract(object["objective"]),
  };
  if (fingerprint({ source: payload.source, objective: payload.objective }) !== payload.requestFingerprint) {
    throw new LedgerIntegrityError("Run request fingerprint does not match its immutable contract");
  }
  return payload;
}

function parseWorkerSpawnPayload(value: unknown): {
  readonly stepId: StepId;
  readonly workerId: WorkerId;
  readonly contractFingerprint: string;
  readonly contract: WorkerContract;
  readonly continuation: WorkerContinuationProvenance | null;
  readonly parentWorkerId: WorkerId | null;
  readonly depth: number;
  readonly attempt: WorkerAttemptProjection;
} {
  const object = strictRecord(
    value,
    [
      "stepId", "workerId", "contractFingerprint", "contract", "continuation",
      "parentWorkerId", "depth", "attempt",
    ],
    "worker-spawn payload",
  );
  const workerId = prefixedId(object, "workerId", "worker") as WorkerId;
  const contractObject = recordValue(object, "contract");
  const request = parseWorkerRequest(contractObject, true);
  const contract: WorkerContract = {
    ...request,
    context: request.context ?? "fresh",
    runId: prefixedId(contractObject, "runId", "run") as RunId,
    workerId: prefixedId(contractObject, "workerId", "worker") as WorkerId,
    createdByDriverSessionId: nonEmptyString(contractObject, "createdByDriverSessionId"),
  };
  if (contract.workerId !== workerId) throw new LedgerIntegrityError("Worker contract ID mismatch");
  const attempt = parseWorkerAttempt(object["attempt"]);
  const continuation = object["continuation"] === undefined || object["continuation"] === null
    ? null
    : parseWorkerContinuationProvenance(object["continuation"]);
  const parentWorkerId = nullablePrefixedId(object, "parentWorkerId", "worker") as WorkerId | null;
  const depth = positiveInteger(object, "depth");
  const contractFingerprint = hashValue(object, "contractFingerprint");
  if (fingerprint({
    request,
    createdByDriverSessionId: contract.createdByDriverSessionId,
    sessionPath: attempt.session.path,
    parentWorkerId,
    depth,
    ...(continuation ? { continuation } : {}),
  }) !== contractFingerprint) {
    throw new LedgerIntegrityError("Worker contract fingerprint does not match its immutable input");
  }
  return {
    stepId: prefixedId(object, "stepId", "step") as StepId,
    workerId,
    contractFingerprint,
    contract,
    continuation,
    parentWorkerId,
    depth,
    attempt,
  };
}

function parseWorkerContinuationProvenance(value: unknown): WorkerContinuationProvenance {
  const object = strictRecord(
    value,
    ["priorWorkerId", "priorAttemptId", "priorLeaseGeneration", "sessionFile"],
    "worker continuation provenance",
  );
  return {
    priorWorkerId: prefixedId(object, "priorWorkerId", "worker") as WorkerId,
    priorAttemptId: prefixedId(object, "priorAttemptId", "attempt") as AttemptId,
    priorLeaseGeneration: nonEmptyString(object, "priorLeaseGeneration"),
    sessionFile: absolutePath(object, "sessionFile"),
  };
}

function parseAttemptSpawnIntentPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attempt: WorkerAttemptProjection;
} {
  const object = strictRecord(value, ["workerId", "attempt"], "attempt-spawn-intent payload");
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attempt: parseWorkerAttempt(object["attempt"]),
  };
}

function parseAttemptStartPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly leaseGeneration: string;
  readonly processIdentity: ProcessIdentity;
  readonly launchContract: WorkerLaunchContract;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "leaseGeneration", "processIdentity", "launchContract"],
    "attempt-start payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    leaseGeneration: nonEmptyString(object, "leaseGeneration"),
    processIdentity: parseProcessIdentity(object["processIdentity"]),
    launchContract: parseWorkerLaunchContract(object["launchContract"]),
  };
}

function parseWorkerReadyPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly sessionFile: string;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "sessionFile"],
    "worker-ready payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    sessionFile: absolutePath(object, "sessionFile"),
  };
}

function parseAttemptUsagePayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly usage: WorkerAttemptProjection["usage"];
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "usage"],
    "attempt-usage payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    usage: parseUsage(object["usage"]),
  };
}

function parseCancelRequestedPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly reason: string;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "reason"],
    "cancel-requested payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    reason: nonEmptyString(object, "reason"),
  };
}

function parseCancelSignalPayload(value: unknown): ReturnType<typeof parseCancelRequestedPayload> & {
  readonly signal: string;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "signal"],
    "cancel-signal payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    reason: "signal-sent",
    signal: nonEmptyString(object, "signal"),
  };
}

function parseAttemptExitPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly exitStatus: { readonly code: number | null; readonly signal: string | null };
  readonly candidateOutcome: WorkerAttemptProjection["candidateOutcome"];
  readonly uncertainSideEffects: boolean;
  readonly reason: string;
  readonly ambiguityId: AmbiguityId | null;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "exitStatus", "candidateOutcome", "uncertainSideEffects", "reason", "ambiguityId"],
    "attempt-exit payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    exitStatus: parseExitStatus(object["exitStatus"]),
    candidateOutcome: parseCandidateOutcome(object["candidateOutcome"]),
    uncertainSideEffects: booleanValue(object, "uncertainSideEffects"),
    reason: nonEmptyString(object, "reason"),
    ambiguityId: nullablePrefixedId(object, "ambiguityId", "ambiguity") as AmbiguityId | null,
  };
}

function parseWorkerNeedsAttentionPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly reason: string;
  readonly ambiguous: boolean;
  readonly ambiguityId: AmbiguityId | null;
} {
  const object = strictRecord(
    value,
    ["workerId", "attemptId", "workerAttemptGeneration", "reason", "ambiguous", "ambiguityId"],
    "worker-needs-attention payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    reason: nonEmptyString(object, "reason"),
    ambiguous: booleanValue(object, "ambiguous"),
    ambiguityId: nullablePrefixedId(object, "ambiguityId", "ambiguity") as AmbiguityId | null,
  };
}

function parseContractRevisionRequestedPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly revisionId: ContractRevisionId;
  readonly revision: number;
  readonly contract: WorkerContract;
  readonly correlation: string | null;
} {
  const object = strictRecord(
    value,
    ["workerId", "revisionId", "revision", "contract", "correlation"],
    "contract-revision-requested payload",
  );
  const workerId = prefixedId(object, "workerId", "worker") as WorkerId;
  const contract = parseWorkerContract(object["contract"]);
  if (contract.workerId !== workerId) {
    throw new LedgerIntegrityError("Contract revision worker ID mismatch");
  }
  return {
    workerId,
    revisionId: prefixedId(object, "revisionId", "revision") as ContractRevisionId,
    revision: positiveInteger(object, "revision"),
    contract,
    correlation: nullableBoundedString(object, "correlation", 256),
  };
}

function parseContractRevisionDeliveredPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly revisionId: ContractRevisionId;
  readonly correlation: string | null;
} {
  const object = strictRecord(
    value,
    ["workerId", "revisionId", "boundary", "correlation"],
    "contract-revision-delivered payload",
  );
  if (nonEmptyString(object, "boundary") !== "next-inference") {
    throw new LedgerIntegrityError("Contract revision delivery boundary is invalid");
  }
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    revisionId: prefixedId(object, "revisionId", "revision") as ContractRevisionId,
    correlation: nullableBoundedString(object, "correlation", 256),
  };
}

function parseWorkerWaitDeclaredPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly workerAttemptGeneration: string;
  readonly supervisorGeneration: string;
  readonly reason: string;
  readonly correlation: string;
} {
  const object = strictRecord(
    value,
    [
      "workerId", "attemptId", "workerAttemptGeneration", "supervisorGeneration",
      "reason", "correlation",
    ],
    "worker-wait-declared payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    supervisorGeneration: nonEmptyString(object, "supervisorGeneration"),
    reason: boundedString(object, "reason", 2_048),
    correlation: boundedString(object, "correlation", 256),
  };
}

function parseCollectObservedPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly payloadFingerprint: string;
  readonly consecutiveRounds: number;
} {
  const object = strictRecord(
    value,
    ["workerId", "payloadFingerprint", "consecutiveRounds"],
    "collect-observed payload",
  );
  const consecutiveRounds = positiveInteger(object, "consecutiveRounds");
  if (consecutiveRounds < 2) {
    throw new LedgerIntegrityError("No-progress threshold must be at least 2");
  }
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    payloadFingerprint: hashValue(object, "payloadFingerprint"),
    consecutiveRounds,
  };
}

function parseSilentStallPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly expectedCadenceMs: number;
} {
  const object = strictRecord(
    value,
    ["workerId", "expectedCadenceMs"],
    "silent-stall-recorded payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    expectedCadenceMs: positiveInteger(object, "expectedCadenceMs"),
  };
}

function parseGateEvaluatedPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly attemptId: AttemptId;
  readonly fromState: WorkerAttemptProjection["state"];
  readonly evidenceExists: boolean;
  readonly reportMatches: boolean | null;
  readonly reportSchemaPolicy: ReportSchemaPolicy | null;
  readonly acceptanceCheckExitCode: number | null;
  readonly humanApproved: boolean | null;
  readonly allowed: boolean;
  readonly blockers: readonly string[];
} {
  const object = strictRecord(
    value,
    [
      "workerId", "attemptId", "fromState", "targetState", "evidenceExists",
      "reportMatches", "acceptanceCheckExitCode", "humanApproved", "budgetExceeded",
      "allowed", "blockers", "reportSchemaPolicy",
    ],
    "gate-evaluated payload",
  );
  if (nonEmptyString(object, "targetState") !== "done") {
    throw new LedgerIntegrityError("Gate target state must be done");
  }
  booleanValue(object, "budgetExceeded");
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    fromState: enumValue(
      object,
      "fromState",
      ["pending", "running", "blocked", "ambiguous", "done", "failed", "cancelled"] as const,
    ),
    evidenceExists: booleanValue(object, "evidenceExists"),
    reportMatches: nullableBoolean(object, "reportMatches"),
    reportSchemaPolicy: nullableReportSchemaPolicy(object, "reportSchemaPolicy"),
    acceptanceCheckExitCode: nullableNonNegativeInteger(object, "acceptanceCheckExitCode"),
    humanApproved: nullableBoolean(object, "humanApproved"),
    allowed: booleanValue(object, "allowed"),
    blockers: stringArray(object, "blockers"),
  };
}

function parseSteerDeliveredPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly messageHash: string;
  readonly correlation: string | null;
} {
  const object = strictRecord(
    value,
    ["workerId", "messageHash", "correlation"],
    "steer-delivered payload",
  );
  return {
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    messageHash: hashValue(object, "messageHash"),
    correlation: nullableBoundedString(object, "correlation", 256),
  };
}

function parsePeerMessageRequestedPayload(value: unknown): Omit<PeerMessageProjection, "status" | "requestedAt" | "deliveredAt" | "diagnostic"> & { readonly workerId: WorkerId } {
  const object = strictRecord(
    value,
    ["workerId", "messageId", "senderWorkerId", "recipientWorkerId", "grant", "correlation", "messageHash"],
    "peer-message-requested payload",
  );
  const senderWorkerId = prefixedId(object, "senderWorkerId", "worker") as WorkerId;
  const workerId = prefixedId(object, "workerId", "worker") as WorkerId;
  if (workerId !== senderWorkerId) throw new LedgerIntegrityError("Peer-message worker ID must be the sender");
  return {
    workerId,
    messageId: prefixedId(object, "messageId", "peer") as PeerMessageId,
    senderWorkerId,
    recipientWorkerId: prefixedId(object, "recipientWorkerId", "worker") as WorkerId,
    grant: nonEmptyString(object, "grant"),
    correlation: boundedString(object, "correlation", 256),
    messageHash: hashValue(object, "messageHash"),
  };
}

function parsePeerMessageDeliveryPayload(value: unknown): {
  readonly workerId: WorkerId;
  readonly messageId: PeerMessageId;
  readonly senderWorkerId: WorkerId;
  readonly recipientWorkerId: WorkerId;
  readonly recipientAttemptId: AttemptId;
  readonly recipientWorkerAttemptGeneration: string;
  readonly status: "delivered" | "indeterminate";
  readonly diagnostic: string | null;
} {
  const object = strictRecord(
    value,
    [
      "workerId", "messageId", "senderWorkerId", "recipientWorkerId", "recipientAttemptId",
      "recipientWorkerAttemptGeneration", "status", "diagnostic",
    ],
    "peer-message-delivery payload",
  );
  const senderWorkerId = prefixedId(object, "senderWorkerId", "worker") as WorkerId;
  const workerId = prefixedId(object, "workerId", "worker") as WorkerId;
  if (workerId !== senderWorkerId) throw new LedgerIntegrityError("Peer-message delivery worker ID must be the sender");
  return {
    workerId,
    messageId: prefixedId(object, "messageId", "peer") as PeerMessageId,
    senderWorkerId,
    recipientWorkerId: prefixedId(object, "recipientWorkerId", "worker") as WorkerId,
    recipientAttemptId: prefixedId(object, "recipientAttemptId", "attempt") as AttemptId,
    recipientWorkerAttemptGeneration: nonEmptyString(object, "recipientWorkerAttemptGeneration"),
    status: enumValue(object, "status", ["delivered", "indeterminate"] as const),
    diagnostic: nullableString(object, "diagnostic"),
  };
}

function parseWorkerAttempt(value: unknown): WorkerAttemptProjection {
  const object = strictRecord(
    value,
    [
      "attemptId", "attemptNumber", "workerAttemptGeneration", "supervisorGeneration", "state",
      "session", "launchContract", "pidStartEvidence", "usage", "candidateOutcome", "sideEffects",
      "control", "cancellation", "modelSelection",
    ],
    "worker attempt",
  );
  const session = strictRecord(object["session"], ["path", "file", "leaseGeneration"], "worker session");
  const process = strictRecord(
    object["pidStartEvidence"],
    [
      "pid", "processStartIdentity", "bootIdentity", "executable", "hostname", "spawnedAt",
      "observedExitAt", "exitStatus",
    ],
    "process start evidence",
  );
  const sideEffects = strictRecord(object["sideEffects"], ["policy", "status"], "attempt side effects");
  const control = object["control"] === undefined
    ? undefined
    : strictRecord(object["control"], ["status", "reason", "recordedAt"], "attempt control");
  const cancellation = strictRecord(
    object["cancellation"],
    ["reason", "requestedAt", "sentAt", "observedAt"],
    "attempt cancellation",
  );
  const modelSelection = object["modelSelection"] === undefined
    ? undefined
    : strictRecord(object["modelSelection"], ["model", "fallbackReason"], "attempt model selection");
  return {
    attemptId: prefixedId(object, "attemptId", "attempt") as AttemptId,
    attemptNumber: positiveInteger(object, "attemptNumber"),
    workerAttemptGeneration: nonEmptyString(object, "workerAttemptGeneration"),
    supervisorGeneration: nonEmptyString(object, "supervisorGeneration"),
    state: normalizeWorkerAttemptState(object["state"]),
    session: {
      path: absolutePath(session, "path"),
      file: nullableAbsolutePath(session, "file"),
      leaseGeneration: nullableString(session, "leaseGeneration"),
    },
    launchContract: object["launchContract"] === null
      ? null
      : parseWorkerLaunchContract(object["launchContract"]),
    ...(modelSelection ? {
      modelSelection: {
        model: parseModelSpec(modelSelection["model"]),
        fallbackReason: nullableString(modelSelection, "fallbackReason"),
      },
    } : {}),
    pidStartEvidence: {
      pid: nullablePositiveInteger(process, "pid"),
      processStartIdentity: nullableString(process, "processStartIdentity"),
      bootIdentity: nullableString(process, "bootIdentity"),
      executable: nullableAbsolutePath(process, "executable"),
      hostname: nullableString(process, "hostname"),
      spawnedAt: nullableIsoTimestamp(process, "spawnedAt"),
      observedExitAt: nullableIsoTimestamp(process, "observedExitAt"),
      exitStatus: process["exitStatus"] === null ? null : parseExitStatus(process["exitStatus"]),
    },
    usage: parseUsage(object["usage"]),
    candidateOutcome: parseCandidateOutcome(object["candidateOutcome"]),
    sideEffects: {
      policy: enumValue(
        sideEffects,
        "policy",
        ["none", "idempotent", "reconcilable", "external"] as const,
      ),
      status: enumValue(sideEffects, "status", ["none", "known", "uncertain"] as const),
    },
    control: control === undefined ? {
      status: "attached",
      reason: null,
      recordedAt: null,
    } : {
      status: enumValue(control, "status", ["attached", "detached"] as const),
      reason: nullableString(control, "reason"),
      recordedAt: nullableIsoTimestamp(control, "recordedAt"),
    },
    cancellation: {
      reason: nullableString(cancellation, "reason"),
      requestedAt: nullableIsoTimestamp(cancellation, "requestedAt"),
      sentAt: nullableIsoTimestamp(cancellation, "sentAt"),
      observedAt: nullableIsoTimestamp(cancellation, "observedAt"),
    },
  };
}

function parseProcessIdentity(value: unknown): ProcessIdentity {
  const object = strictRecord(
    value,
    ["pid", "processStartIdentity", "bootIdentity", "executable", "hostname"],
    "process identity",
  );
  return {
    pid: positiveInteger(object, "pid"),
    processStartIdentity: nonEmptyString(object, "processStartIdentity"),
    bootIdentity: nullableString(object, "bootIdentity"),
    executable: absolutePath(object, "executable"),
    hostname: nonEmptyString(object, "hostname"),
  };
}

function parseUsage(value: unknown): WorkerAttemptProjection["usage"] {
  const object = strictRecord(value, ["tokens", "costUsd", "outputBytes"], "attempt usage");
  return {
    tokens: nonNegativeInteger(object, "tokens"),
    costUsd: nonNegativeNumber(object, "costUsd"),
    outputBytes: nonNegativeInteger(object, "outputBytes"),
  };
}

function parseExitStatus(value: unknown): { readonly code: number | null; readonly signal: string | null } {
  const object = strictRecord(value, ["code", "signal"], "process exit status");
  const code = object["code"] === null ? null : nonNegativeInteger(object, "code");
  const signal = nullableString(object, "signal");
  if (code === null && signal === null) throw new TypeError("Process exit requires code or signal");
  return { code, signal };
}

function parseCandidateOutcome(value: unknown): WorkerAttemptProjection["candidateOutcome"] {
  const object = strictRecord(value, ["valueOrRef", "hash"], "candidate outcome");
  const valueOrRef = nullableString(object, "valueOrRef");
  const hash = nullableString(object, "hash");
  return { valueOrRef, hash };
}

function parseAmbiguityPayload(value: unknown, recordedAt: string): AmbiguityProjection {
  const object = strictRecord(
    value,
    ["ambiguityId", "reason", "relatedIds", "status"],
    "ambiguity payload",
  );
  if (nonEmptyString(object, "status") !== "needs-reconcile") {
    throw new LedgerIntegrityError("New ambiguity must be needs-reconcile");
  }
  return {
    ambiguityId: prefixedId(object, "ambiguityId", "ambiguity") as AmbiguityId,
    status: "needs-reconcile",
    reason: nonEmptyString(object, "reason"),
    relatedIds: stringArray(object, "relatedIds"),
    recordedAt,
  };
}

function parseReconciliationPayload(value: unknown): {
  readonly ambiguityId: AmbiguityId;
  readonly decision: string;
  readonly evidenceRefs: readonly string[];
  readonly workerId: WorkerId | null;
  readonly attemptId: AttemptId | null;
  readonly workerState: Exclude<WorkerAttemptProjection["state"], "ambiguous"> | null;
} {
  const object = strictRecord(
    value,
    ["ambiguityId", "decision", "evidenceRefs", "workerId", "attemptId", "workerState"],
    "reconciliation payload",
  );
  const workerState = object["workerState"] === undefined || object["workerState"] === null
    ? null
    : enumValue(
        object,
        "workerState",
        ["pending", "running", "blocked", "done", "failed", "cancelled"] as const,
      );
  return {
    ambiguityId: prefixedId(object, "ambiguityId", "ambiguity") as AmbiguityId,
    decision: nonEmptyString(object, "decision"),
    evidenceRefs: stringArray(object, "evidenceRefs"),
    workerId: object["workerId"] === undefined
      ? null
      : nullablePrefixedId(object, "workerId", "worker") as WorkerId | null,
    attemptId: object["attemptId"] === undefined
      ? null
      : nullablePrefixedId(object, "attemptId", "attempt") as AttemptId | null,
    workerState,
  };
}

function parseRunSource(value: unknown): RunSource {
  const object = strictRecord(value, ["kind", "amqMessageId", "driverSessionId"], "run source");
  const kind = enumValue(object, "kind", ["interactive", "amq", "external-engine"] as const);
  const amqMessageId = nullableString(object, "amqMessageId");
  if (kind === "amq" && amqMessageId === null) throw new TypeError("AMQ run source requires amqMessageId");
  if (kind !== "amq" && amqMessageId !== null) {
    throw new TypeError("Only an AMQ run source may carry amqMessageId");
  }
  return { kind, amqMessageId, driverSessionId: nonEmptyString(object, "driverSessionId") };
}

function parseObjectiveContract(value: unknown): ObjectiveContract {
  const object = strictRecord(
    value,
    ["schema", "goal", "acceptance", "constraints", "permissions", "budget", "evidenceGraph"],
    "objective contract",
  );
  if (numberValue(object, "schema") !== 1) throw new TypeError("Objective schema must be 1");
  const evidenceGraph = object["evidenceGraph"] === undefined
    ? undefined
    : parseEvidenceGraph(object["evidenceGraph"]);
  return {
    schema: 1,
    goal: nonEmptyString(object, "goal"),
    acceptance: nonEmptyStringArray(object, "acceptance"),
    constraints: stringArray(object, "constraints"),
    permissions: stringArray(object, "permissions"),
    budget: parseRunBudget(object["budget"]),
    ...(evidenceGraph ? { evidenceGraph } : {}),
  };
}

function parseWorkerRequest(value: unknown, allowContractFields = false): WorkerRequest {
  const requestKeys = [
    "schema", "requestKey", "difficulty", "role", "proposal", "context", "forkSource", "goal", "acceptance", "constraints", "budget",
    "model", "modelFallbacks", "capabilities", "workspace", "contextRefs", "channel", "sideEffectPolicy",
    "report_schema", "report_schema_policy", "requires_human_approval", "acceptance_check",
  ] as const;
  const object = strictRecord(
    value,
    allowContractFields
      ? [...requestKeys, "runId", "workerId", "createdByDriverSessionId"]
      : requestKeys,
    "worker request",
  );
  if (numberValue(object, "schema") !== 1) throw new TypeError("Worker request schema must be 1");
  const difficulty = object["difficulty"] === undefined
    ? undefined
    : enumValue(object, "difficulty", ["hard", "medium", "easy"] as const);
  const role = object["role"] === undefined ? undefined : parseWorkerRoleContract(object["role"]);
  const proposal = object["proposal"] === undefined
    ? undefined
    : parseWorkerProposalProvenance(object["proposal"]);
  const context = object["context"] === undefined
    ? "fresh"
    : enumValue(object, "context", ["fresh", "fork"] as const);
  const forkSource = object["forkSource"] === undefined
    ? undefined
    : parseWorkerForkSource(object["forkSource"]);
  if ((context === "fork") !== (forkSource !== undefined)) {
    throw new TypeError("Fork worker context requires forkSource, and fresh context forbids it");
  }
  const reportSchema = object["report_schema"] === undefined
    ? undefined
    : parseReportSchema(object["report_schema"]);
  const reportSchemaPolicy = object["report_schema_policy"] === undefined
    ? undefined
    : enumValue(object, "report_schema_policy", ["strict", "preferred", "advisory"] as const);
  const requiresHumanApproval = object["requires_human_approval"] === undefined
    ? undefined
    : booleanValue(object, "requires_human_approval");
  const acceptanceCheck = object["acceptance_check"] === undefined
    ? undefined
    : parseAcceptanceCheck(object["acceptance_check"]);
  const model = parseModelSpec(object["model"]);
  const modelFallbacks = object["modelFallbacks"] === undefined
    ? []
    : parseModelFallbacks(object["modelFallbacks"]);
  const modelKeys = [model, ...modelFallbacks].map((candidate) => `${candidate.provider}/${candidate.id}`);
  if (new Set(modelKeys).size !== modelKeys.length) {
    throw new TypeError("Primary and fallback model chain entries must be unique");
  }
  return {
    schema: 1,
    requestKey: validateRequestKey(nonEmptyString(object, "requestKey")),
    ...(difficulty ? { difficulty } : {}),
    ...(role ? { role } : {}),
    ...(proposal ? { proposal } : {}),
    context,
    ...(forkSource ? { forkSource } : {}),
    goal: nonEmptyString(object, "goal"),
    acceptance: nonEmptyStringArray(object, "acceptance"),
    constraints: stringArray(object, "constraints"),
    budget: parseWorkerBudget(object["budget"]),
    model,
    ...(modelFallbacks.length > 0 ? { modelFallbacks } : {}),
    capabilities: stringArray(object, "capabilities"),
    workspace: parseWorkspaceGrant(object["workspace"]),
    contextRefs: parseContentRefs(object["contextRefs"]),
    channel: enumValue(object, "channel", ["json-result", "rpc"] as const),
    sideEffectPolicy: enumValue(
      object,
      "sideEffectPolicy",
      ["none", "idempotent", "reconcilable", "external"] as const,
    ),
    ...(reportSchema ? { report_schema: reportSchema } : {}),
    ...(reportSchemaPolicy ? { report_schema_policy: reportSchemaPolicy } : {}),
    ...(requiresHumanApproval === undefined
      ? {}
      : { requires_human_approval: requiresHumanApproval }),
    ...(acceptanceCheck ? { acceptance_check: acceptanceCheck } : {}),
  };
}

function parseWorkerProposalProvenance(value: unknown): WorkerProposalProvenance {
  const object = strictRecord(
    value,
    ["source", "confirmSpawns", "disposition", "humanEditedFields"],
    "worker proposal provenance",
  );
  const humanEditedFields = stringArray(object, "humanEditedFields").map((field) => {
    if (!["role", "task", "model", "difficulty", "budget"].includes(field)) {
      throw new TypeError(`Unknown human-edited proposal field ${field}`);
    }
    return field as WorkerProposalProvenance["humanEditedFields"][number];
  });
  if (new Set(humanEditedFields).size !== humanEditedFields.length) {
    throw new TypeError("Human-edited proposal fields must be unique");
  }
  const source = enumValue(object, "source", ["tool", "human-edited"] as const);
  if ((source === "tool") !== (humanEditedFields.length === 0)) {
    throw new TypeError("Proposal source and human-edited fields disagree");
  }
  return {
    source,
    confirmSpawns: enumValue(object, "confirmSpawns", ["on", "multi-worker-only", "off"] as const),
    disposition: enumValue(
      object,
      "disposition",
      ["accepted", "skipped-policy-off", "skipped-single-worker"] as const,
    ),
    humanEditedFields,
  };
}

function parseWorkerForkSource(value: unknown): import("./workflow-types.js").WorkerForkSource {
  const object = strictRecord(
    value,
    ["sourceSessionFile", "branchPointId", "inheritedEntries", "inheritedBytes"],
    "worker fork source",
  );
  const sourceSessionFile = nonEmptyString(object, "sourceSessionFile");
  if (!isAbsolute(sourceSessionFile)) throw new TypeError("Fork source session file must be absolute");
  return {
    sourceSessionFile: normalize(sourceSessionFile),
    branchPointId: nonEmptyString(object, "branchPointId"),
    inheritedEntries: positiveInteger(object, "inheritedEntries"),
    inheritedBytes: nonNegativeInteger(object, "inheritedBytes"),
  };
}

function parseWorkerRoleContract(value: unknown): WorkerRoleContract {
  const object = strictRecord(
    value,
    ["name", "tier", "charter", "presetSource", "presetFingerprint"],
    "worker role contract",
  );
  const presetFingerprint = nonEmptyString(object, "presetFingerprint");
  if (!/^[a-f0-9]{64}$/.test(presetFingerprint)) {
    throw new TypeError("Worker role preset fingerprint must be a lowercase SHA-256 hex digest");
  }
  return {
    name: nonEmptyString(object, "name"),
    tier: nonEmptyString(object, "tier"),
    charter: nonEmptyString(object, "charter"),
    presetSource: enumValue(object, "presetSource", ["builtin", "user", "project"] as const),
    presetFingerprint,
  };
}

function parseWorkerContract(value: unknown): WorkerContract {
  const object = recordValue({ value }, "value");
  const request = parseWorkerRequest(object, true);
  return {
    ...request,
    context: request.context ?? "fresh",
    runId: prefixedId(object, "runId", "run") as RunId,
    workerId: prefixedId(object, "workerId", "worker") as WorkerId,
    createdByDriverSessionId: nonEmptyString(object, "createdByDriverSessionId"),
  };
}

function parseReportSchema(value: unknown): ReportSchemaV1 {
  const object = strictRecord(value, ["schema", "required", "properties"], "report schema");
  if (numberValue(object, "schema") !== 1) throw new TypeError("Report schema must be 1");
  const propertiesValue = object["properties"];
  if (typeof propertiesValue !== "object" || propertiesValue === null || Array.isArray(propertiesValue)) {
    throw new TypeError("Report schema properties must be an object");
  }
  const properties: Record<string, { readonly type: ReportSchemaV1["properties"][string]["type"] }> = {};
  for (const [name, definition] of Object.entries(propertiesValue)) {
    requireNonEmpty(name, "report schema property name");
    const field = strictRecord(definition, ["type"], `report schema property ${name}`);
    properties[name] = {
      type: enumValue(
        field,
        "type",
        ["string", "number", "boolean", "object", "array", "null"] as const,
      ),
    };
  }
  const required = stringArray(object, "required");
  for (const name of required) {
    if (!properties[name]) throw new TypeError(`Required report property ${name} is not declared`);
  }
  return { schema: 1, required, properties };
}

function parseAcceptanceCheck(value: unknown): AcceptanceCheck {
  const object = strictRecord(value, ["command", "args"], "acceptance check");
  return {
    command: nonEmptyString(object, "command"),
    args: stringArray(object, "args"),
  };
}

function parseRunBudget(value: unknown): RunBudget {
  const object = strictRecord(
    value,
    ["maxWorkers", "maxConcurrency", "maxDepth", "maxAttempts", "maxWallClockMs", "maxTokens", "maxCostUsd", "maxOutputBytes"],
    "run budget",
  );
  return {
    maxWorkers: parseBudgetLimit(object["maxWorkers"], "run maxWorkers", { integer: true, allowLegacyNumber: true }),
    maxConcurrency: parseBudgetLimit(object["maxConcurrency"], "run maxConcurrency", { integer: true, allowLegacyNumber: true }),
    maxDepth: parseBudgetLimit(object["maxDepth"], "run maxDepth", { integer: true, allowLegacyNumber: true }),
    maxAttempts: parseBudgetLimit(object["maxAttempts"], "run maxAttempts", { integer: true, allowLegacyNumber: true }),
    maxWallClockMs: parseBudgetLimit(object["maxWallClockMs"], "run maxWallClockMs", { integer: true, allowLegacyNumber: true }),
    maxTokens: parseBudgetLimit(object["maxTokens"], "run maxTokens", { integer: true, allowLegacyNumber: true }),
    maxCostUsd: object["maxCostUsd"] === undefined
      ? UNLIMITED
      : parseBudgetLimit(object["maxCostUsd"], "run maxCostUsd", { integer: false, allowLegacyNumber: true }),
    maxOutputBytes: object["maxOutputBytes"] === undefined
      ? UNLIMITED
      : parseBudgetLimit(object["maxOutputBytes"], "run maxOutputBytes", { integer: true, allowLegacyNumber: true }),
  };
}

function parseWorkerBudget(value: unknown): WorkerBudget {
  const object = strictRecord(
    value,
    ["maxAttempts", "maxWallClockMs", "maxTokens", "maxCostUsd", "maxOutputBytes"],
    "worker budget",
  );
  return {
    maxAttempts: parseBudgetLimit(object["maxAttempts"], "worker maxAttempts", { integer: true, allowLegacyNumber: true }),
    maxWallClockMs: parseBudgetLimit(object["maxWallClockMs"], "worker maxWallClockMs", { integer: true, allowLegacyNumber: true }),
    maxTokens: parseBudgetLimit(object["maxTokens"], "worker maxTokens", { integer: true, allowLegacyNumber: true }),
    maxCostUsd: object["maxCostUsd"] === undefined
      ? UNLIMITED
      : parseBudgetLimit(object["maxCostUsd"], "worker maxCostUsd", { integer: false, allowLegacyNumber: true }),
    maxOutputBytes: object["maxOutputBytes"] === undefined
      ? UNLIMITED
      : parseBudgetLimit(object["maxOutputBytes"], "worker maxOutputBytes", { integer: true, allowLegacyNumber: true }),
  };
}

function parseModelSpec(value: unknown): ModelSpec {
  const object = strictRecord(value, ["provider", "id", "thinkingLevel"], "model spec");
  const thinkingLevel = object["thinkingLevel"] === undefined
    ? undefined
    : nonEmptyString(object, "thinkingLevel");
  return {
    provider: nonEmptyString(object, "provider"),
    id: nonEmptyString(object, "id"),
    ...(thinkingLevel ? { thinkingLevel } : {}),
  };
}

function parseModelFallbacks(value: unknown): readonly ModelSpec[] {
  if (!Array.isArray(value)) throw new TypeError("modelFallbacks must be an array");
  if (value.length > 4) throw new TypeError("modelFallbacks may contain at most four models");
  return value.map(parseModelSpec);
}

function parseWorkspaceGrant(value: unknown): WorkspaceGrant {
  const object = strictRecord(value, ["path", "mode", "worktreePath"], "workspace grant");
  const worktreePath = object["worktreePath"] === undefined
    ? undefined
    : absolutePath(object, "worktreePath");
  return {
    path: absolutePath(object, "path"),
    mode: enumValue(object, "mode", ["read-only", "mutating"] as const),
    ...(worktreePath ? { worktreePath } : {}),
  };
}

function parseContentRefs(value: unknown): readonly ContentRef[] {
  if (!Array.isArray(value)) throw new TypeError("contextRefs must be an array");
  return value.map((item) => {
    const object = strictRecord(item, ["kind", "value", "hash"], "content reference");
    const hash = object["hash"] === undefined ? undefined : nonEmptyString(object, "hash");
    return {
      kind: enumValue(object, "kind", ["file", "session", "artifact"] as const),
      value: nonEmptyString(object, "value"),
      ...(hash ? { hash } : {}),
    };
  });
}

function parseEvidenceGraph(value: unknown): { readonly nodes: readonly EvidenceNode[] } {
  const object = strictRecord(value, ["nodes"], "evidence graph");
  const nodes = object["nodes"];
  if (!Array.isArray(nodes)) throw new TypeError("evidence graph nodes must be an array");
  const parsed = nodes.map((item) => {
      const node = strictRecord(
        item,
        ["id", "kind", "assertion", "verifier", "requires"],
        "evidence node",
      );
      return {
        id: nonEmptyString(node, "id"),
        kind: enumValue(node, "kind", ["claim", "artifact"] as const),
        assertion: nonEmptyString(node, "assertion"),
        verifier: nonEmptyString(node, "verifier"),
        requires: stringArray(node, "requires"),
      };
    });
  const byId = new Map<string, EvidenceNode>();
  for (const node of parsed) {
    if (byId.has(node.id)) throw new TypeError(`Duplicate evidence node ID ${node.id}`);
    byId.set(node.id, node);
  }
  for (const node of parsed) {
    for (const required of node.requires) {
      if (!byId.has(required)) {
        throw new TypeError(`Evidence node ${node.id} requires unknown node ${required}`);
      }
    }
  }
  const visiting = new Set<string>();
  const visited = new Set<string>();
  const visit = (id: string): void => {
    if (visiting.has(id)) throw new TypeError(`Evidence graph contains a cycle at ${id}`);
    if (visited.has(id)) return;
    visiting.add(id);
    for (const required of byId.get(id)!.requires) visit(required);
    visiting.delete(id);
    visited.add(id);
  };
  for (const node of parsed) visit(node.id);
  return { nodes: parsed };
}

function initialAttempt(
  attemptId: AttemptId,
  attemptNumber: number,
  workerAttemptGeneration: string,
  supervisorGeneration: string,
  sessionPath: string,
  sideEffectPolicy: WorkerRequest["sideEffectPolicy"],
  model: ModelSpec,
  fallbackReason: string | null,
): WorkerAttemptProjection {
  return {
    attemptId,
    attemptNumber,
    workerAttemptGeneration: requireNonEmpty(workerAttemptGeneration, "worker attempt generation"),
    supervisorGeneration: requireNonEmpty(supervisorGeneration, "supervisor generation"),
    state: "pending",
    session: { path: sessionPath, file: null, leaseGeneration: null },
    launchContract: null,
    modelSelection: {
      model: parseModelSpec(model),
      fallbackReason: fallbackReason === null ? null : requireNonEmpty(fallbackReason, "fallback reason"),
    },
    pidStartEvidence: {
      pid: null,
      processStartIdentity: null,
      bootIdentity: null,
      executable: null,
      hostname: null,
      spawnedAt: null,
      observedExitAt: null,
      exitStatus: null,
    },
    usage: { tokens: 0, costUsd: 0, outputBytes: 0 },
    candidateOutcome: { valueOrRef: null, hash: null },
    sideEffects: { policy: sideEffectPolicy, status: "none" },
    control: { status: "attached", reason: null, recordedAt: null },
    cancellation: { reason: null, requestedAt: null, sentAt: null, observedAt: null },
  };
}

function initialBudgetProjection(limits: RunBudget): RunBudgetProjection {
  return {
    limits,
    reserved: { workers: 0, attempts: 0, tokens: 0, costUsd: 0, outputBytes: 0 },
    spent: { tokens: 0, costUsd: 0, outputBytes: 0 },
    remaining: {
      workers: limits.maxWorkers,
      attempts: limits.maxAttempts,
      tokens: limits.maxTokens,
      wallClockMs: limits.maxWallClockMs,
      costUsd: limits.maxCostUsd,
      outputBytes: limits.maxOutputBytes,
    },
  };
}

function reserveProjectionBudget(
  projection: RunProjectionV2,
  workerBudget: WorkerBudget,
  newWorker: boolean,
): RunProjectionV2 {
  const reserved = {
    workers: projection.budget.reserved.workers + (newWorker ? 1 : 0),
    attempts: projection.budget.reserved.attempts + 1,
    tokens: reserveBudget(projection.budget.reserved.tokens, workerBudget.maxTokens),
    costUsd: reserveBudget(projection.budget.reserved.costUsd, workerBudget.maxCostUsd),
    outputBytes: reserveBudget(projection.budget.reserved.outputBytes, workerBudget.maxOutputBytes),
  };
  return {
    ...projection,
    budget: {
      ...projection.budget,
      reserved,
      remaining: {
        ...projection.budget.remaining,
        workers: remainingBudget(projection.budget.limits.maxWorkers, reserved.workers),
        attempts: remainingBudget(projection.budget.limits.maxAttempts, reserved.attempts),
        tokens: remainingBudget(projection.budget.limits.maxTokens, reserved.tokens),
        costUsd: remainingBudget(projection.budget.limits.maxCostUsd, reserved.costUsd),
        outputBytes: remainingBudget(projection.budget.limits.maxOutputBytes, reserved.outputBytes),
      },
    },
  };
}

function spendProjectionBudget(
  projection: RunProjectionV2,
  delta: { readonly tokens: number; readonly costUsd: number; readonly outputBytes: number },
): RunProjectionV2 {
  return {
    ...projection,
    budget: {
      ...projection.budget,
      spent: {
        tokens: projection.budget.spent.tokens + delta.tokens,
        costUsd: projection.budget.spent.costUsd + delta.costUsd,
        outputBytes: projection.budget.spent.outputBytes + delta.outputBytes,
      },
    },
  };
}

function withRemainingWallClock(
  budget: RunBudgetProjection,
  createdAt: string,
  observedAt: string,
): RunBudgetProjection {
  const elapsed = Math.max(0, Date.parse(observedAt) - Date.parse(createdAt));
  return {
    ...budget,
    remaining: {
      ...budget.remaining,
      wallClockMs: remainingBudget(budget.limits.maxWallClockMs, elapsed),
    },
  };
}

function assertWorkerCapabilitySubset(current: RunProjectionV2, request: WorkerRequest): void {
  const permissions = new Set(current.objective.permissions);
  const missing = request.capabilities.filter((capability) => {
    if (capability.startsWith("message_worker:")) {
      if (capability.slice("message_worker:".length).trim().length === 0) {
        throw new TypeError("message_worker capability requires a worker ID or role target");
      }
      return !permissions.has("message_worker");
    }
    return !permissions.has(capability);
  });
  if (missing.length > 0) {
    throw new TypeError(`Worker capabilities exceed objective permissions: ${missing.join(", ")}`);
  }
}

function resolvePeerGrantRecipient(
  current: RunProjectionV2,
  grant: string,
  ledgerReplay: boolean,
): WorkerProjection {
  const prefix = "message_worker:";
  if (!grant.startsWith(prefix)) {
    throw peerGrantError(`Invalid peer-message grant ${grant}`, ledgerReplay);
  }
  const target = grant.slice(prefix.length);
  if (target.trim().length === 0 || target !== target.trim()) {
    throw peerGrantError("Peer-message grant requires a normalized target", ledgerReplay);
  }
  const candidates = target.startsWith("worker_")
    ? [current.workers[target as WorkerId]].filter(
        (worker): worker is WorkerProjection => worker !== undefined,
      )
    : Object.values(current.workers).filter(
        (worker) => effectiveWorkerContract(worker).role?.name === target,
      );
  if (candidates.length !== 1) {
    throw peerGrantError(
      `Peer-message grant ${grant} resolves to ${candidates.length} workers`,
      ledgerReplay,
    );
  }
  return candidates[0]!;
}

function peerGrantError(message: string, ledgerReplay: boolean): Error {
  return ledgerReplay ? new LedgerIntegrityError(message) : new TypeError(message);
}

function assertSpawnBudget(
  current: RunProjectionV2,
  workerBudget: WorkerBudget,
  depth: number,
  now: Date,
): void {
  if (budgetExhausted(current.budget.reserved.workers, current.budget.limits.maxWorkers)) {
    throw new BudgetExceededError("Run worker budget is exhausted");
  }
  assertReplacementBudget(current, workerBudget, depth, now);
}

function assertReplacementBudget(
  current: RunProjectionV2,
  workerBudget: WorkerBudget,
  depth: number,
  now: Date,
): void {
  const active = Object.values(current.workers).filter((worker) =>
    isAttemptProcessActive(requiredCurrentAttempt(worker))).length;
  if (budgetExhausted(active, current.budget.limits.maxConcurrency)) {
    throw new BudgetExceededError("Run concurrency budget is exhausted");
  }
  if (budgetExceeded(depth, current.budget.limits.maxDepth)) {
    throw new BudgetExceededError(`Worker depth ${depth} exceeds run maxDepth`);
  }
  if (budgetExhausted(current.budget.reserved.attempts, current.budget.limits.maxAttempts)) {
    throw new BudgetExceededError("Run attempt budget is exhausted");
  }
  if (!budgetRequestFits(current.budget.limits.maxTokens, current.budget.reserved.tokens, workerBudget.maxTokens)) {
    throw new BudgetExceededError(reservationRefusal(
      "token", "max_tokens", workerBudget.maxTokens,
      remainingBudget(current.budget.limits.maxTokens, current.budget.reserved.tokens),
    ));
  }
  if (!budgetRequestFits(current.budget.limits.maxCostUsd, current.budget.reserved.costUsd, workerBudget.maxCostUsd)) {
    throw new BudgetExceededError(reservationRefusal(
      "cost", "max_cost_usd", workerBudget.maxCostUsd,
      remainingBudget(current.budget.limits.maxCostUsd, current.budget.reserved.costUsd),
    ));
  }
  if (!budgetRequestFits(current.budget.limits.maxOutputBytes, current.budget.reserved.outputBytes, workerBudget.maxOutputBytes)) {
    throw new BudgetExceededError(`${reservationRefusal(
      "output-byte", "max_output_bytes", workerBudget.maxOutputBytes,
      remainingBudget(current.budget.limits.maxOutputBytes, current.budget.reserved.outputBytes),
    )}. ${OUTPUT_BYTE_UNIT_NOTE}`);
  }
  const elapsed = Math.max(0, now.getTime() - Date.parse(current.createdAt));
  if (budgetExhausted(elapsed, current.budget.limits.maxWallClockMs)) {
    throw new BudgetExceededError("Run wall-clock budget is exhausted");
  }
  // Compared against the run's whole wall-clock allowance, not the time still left. The
  // remaining figure shrinks between composing a request and validating it, so a request for
  // exactly the remaining time can never pass — including the default budget, which is derived
  // from that same figure. The run deadline still terminates the worker independently, so this
  // guard only has to reject a worker that declares more time than the run could ever have.
  if (!budgetRequestFits(current.budget.limits.maxWallClockMs, 0, workerBudget.maxWallClockMs)) {
    throw new BudgetExceededError(reservationRefusal(
      "wall-clock", "max_wall_clock_ms", workerBudget.maxWallClockMs,
      current.budget.limits.maxWallClockMs,
    ));
  }
}

function budgetRequestFits(
  limit: import("./workflow-types.js").BudgetLimit,
  reserved: number | "unlimited",
  requested: import("./workflow-types.js").BudgetLimit,
): boolean {
  if (limit.kind === "unlimited") return true;
  if (reserved === "unlimited" || requested.kind === "unlimited") return false;
  return reserved + requested.value <= limit.value;
}

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

export function effectiveWorkerContract(worker: WorkerProjection): WorkerContract {
  return worker.contractRevisions.filter((revision) => revision.status === "delivered").at(-1)
    ?.contract ?? worker.contract;
}

export function latestWorkerContract(worker: WorkerProjection): WorkerContract {
  return worker.contractRevisions.at(-1)?.contract ?? worker.contract;
}

function optionalRevisionField(
  key: "report_schema" | "acceptance_check",
  input: {
    readonly report_schema?: ReportSchemaV1 | null;
    readonly report_schema_policy?: ReportSchemaPolicy;
    readonly acceptance_check?: AcceptanceCheck | null;
  },
  previous: WorkerContract,
): Record<string, unknown> {
  if (Object.prototype.hasOwnProperty.call(input, key)) {
    const value = input[key];
    return value === null || value === undefined ? {} : { [key]: value };
  }
  const previousValue = previous[key];
  return previousValue === undefined ? {} : { [key]: previousValue };
}

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

function requiredWorkerAttempt(
  projection: RunProjectionV2,
  workerId: WorkerId,
  attemptId: AttemptId,
): { readonly worker: WorkerProjection; readonly attempt: WorkerAttemptProjection } {
  const worker = requiredWorker(projection, workerId);
  const attempt = worker.attempts.find((candidate) => candidate.attemptId === attemptId);
  if (!attempt) throw new Error(`Unknown attempt ${attemptId} for worker ${workerId}`);
  return { worker, attempt };
}

function updateWorker(
  projection: RunProjectionV2,
  workerId: WorkerId,
  update: (worker: WorkerProjection) => WorkerProjection,
): RunProjectionV2 {
  const worker = requiredWorker(projection, workerId);
  const updated = update(worker);
  return { ...projection, workers: { ...projection.workers, [workerId]: updated } };
}

function workerGateFacts(input: {
  readonly evidenceExists?: boolean;
  readonly reportSchemaDeclared?: boolean;
  readonly reportMatches?: boolean;
  readonly reportSchemaPolicy?: ReportSchemaPolicy;
  readonly acceptanceDeclared?: boolean;
  readonly acceptanceExitCode?: number | null;
  readonly humanRequired?: boolean;
  readonly humanApproved?: boolean;
  readonly budgetExceeded?: boolean;
  readonly runtimeCertain?: boolean;
  readonly effectsUncertain?: boolean;
  readonly supervisorLost?: boolean;
  readonly failureObserved?: boolean;
  readonly terminationObserved?: boolean;
  readonly waitDeclared?: boolean;
  readonly reconciliation?: boolean;
  readonly uncertaintyCleared?: boolean;
  readonly input?: "revise_contract" | "steer" | "peer_message";
  readonly inputCorrelationMatches?: boolean;
  readonly cancellationRequested?: boolean;
} = {}): WorkerGateFacts {
  return {
    evidence: { exists: input.evidenceExists ?? false },
    report: {
      schemaDeclared: input.reportSchemaDeclared ?? false,
      matches: input.reportMatches ?? false,
      ...(input.reportSchemaPolicy === undefined ? {} : { policy: input.reportSchemaPolicy }),
    },
    acceptanceCheck: {
      declared: input.acceptanceDeclared ?? false,
      exitCode: input.acceptanceExitCode ?? null,
    },
    human: {
      approvalRequired: input.humanRequired ?? false,
      approved: input.humanApproved ?? false,
    },
    budget: { exceeded: input.budgetExceeded ?? false },
    runtime: {
      supervisorLost: input.supervisorLost ?? false,
      effectsUncertain: input.effectsUncertain ?? false,
      certain: input.runtimeCertain ?? !(input.supervisorLost || input.effectsUncertain),
    },
    failure: { observed: input.failureObserved ?? false },
    termination: { observed: input.terminationObserved ?? false },
    wait: { declared: input.waitDeclared ?? false },
    reconcile: {
      decisionExists: input.reconciliation ?? false,
      uncertaintyCleared: input.uncertaintyCleared ?? false,
    },
    input: {
      kind: input.input ?? null,
      correlationMatches: input.inputCorrelationMatches ?? false,
    },
    cancel: { verbReceived: input.cancellationRequested ?? false },
  };
}

function requiredMatchingWait(
  worker: WorkerProjection,
  correlation: string | null,
  ledgerReplay: boolean,
): WorkerProjection["waits"][number] {
  const attempt = requiredCurrentAttempt(worker);
  const active = worker.waits.filter((wait) => wait.status === "waiting");
  if (active.length !== 1) {
    throw waitCorrelationError(
      `Blocked worker ${worker.workerId} must have exactly one active typed wait`,
      ledgerReplay,
    );
  }
  const wait = active[0]!;
  if (wait.attemptId !== attempt.attemptId
    || wait.workerAttemptGeneration !== attempt.workerAttemptGeneration) {
    throw waitCorrelationError("Typed wait belongs to a stale worker attempt", ledgerReplay);
  }
  if (correlation === null || wait.correlation !== correlation) {
    throw waitCorrelationError(
      `Input correlation does not match typed wait ${wait.correlation}`,
      ledgerReplay,
    );
  }
  return wait;
}

function wakeWorkerFromWait(
  worker: WorkerProjection,
  input: "revise_contract" | "steer" | "peer_message",
  correlation: string | null,
  occurredAt: string,
): WorkerProjection {
  if (worker.state !== "blocked") return worker;
  const wait = requiredMatchingWait(worker, correlation, true);
  const attempt = requiredCurrentAttempt(worker);
  const facts = workerGateFacts({ input, inputCorrelationMatches: true });
  return {
    ...worker,
    state: gatedWorkerState(worker.state, "running", facts),
    attempts: worker.attempts.map((candidate) => candidate.attemptId === attempt.attemptId
      ? { ...candidate, state: gatedWorkerState(candidate.state, "running", facts) }
      : candidate),
    waits: worker.waits.map((candidate) => candidate === wait
      ? {
          ...candidate,
          status: "satisfied" as const,
          satisfiedAt: occurredAt,
          satisfiedBy: input,
        }
      : candidate),
  };
}

function messageCorrelation(worker: WorkerProjection, messageId: PeerMessageId): string {
  const message = worker.peerMessages.find((candidate) => candidate.messageId === messageId);
  if (!message) throw new LedgerIntegrityError(`Unknown peer message ${messageId}`);
  return message.correlation;
}

function waitCorrelationError(message: string, ledgerReplay: boolean): Error {
  return ledgerReplay ? new LedgerIntegrityError(message) : new RunConflictError(message);
}

function gatedWorkerState(
  from: WorkerAttemptState,
  to: WorkerAttemptState,
  facts: WorkerGateFacts,
): WorkerAttemptState {
  if (from === to) return to;
  const result = evaluateWorkerAttemptTransition(from, to, facts);
  if (!result.allowed) {
    throw new LedgerIntegrityError(
      `Worker Gate refused ${from} -> ${to}: ${result.reason}${
        result.blockers.length > 0 ? ` (${result.blockers.join(", ")})` : ""
      }`,
    );
  }
  return to;
}

function runGateFacts(input: {
  readonly workersDetached?: boolean;
  readonly reconciliationRecorded?: boolean;
  readonly uncertaintyCleared?: boolean;
} = {}): RunGateFacts {
  return {
    objective: {
      evidenceComplete: false,
      verifierPassed: false,
      prerequisitesComplete: false,
    },
    workers: {
      active: false,
      uncertain: false,
      detached: input.workersDetached ?? false,
    },
    cancellation: { requested: false, settled: false },
    input: { delivered: false },
    reconciliation: {
      recorded: input.reconciliationRecorded ?? false,
      uncertaintyCleared: input.uncertaintyCleared ?? false,
    },
    attention: { cleared: false },
  };
}

function gatedRunState(
  from: RunProjectionV2["state"],
  to: RunProjectionV2["state"],
  facts: RunGateFacts,
): RunProjectionV2["state"] {
  if (from === to) return to;
  const result = evaluateRunTransition(from, to, facts);
  if (!result.allowed) {
    throw new LedgerIntegrityError(
      `Run Gate refused ${from} -> ${to}: ${result.reason}${
        result.blockers.length > 0 ? ` (${result.blockers.join(", ")})` : ""
      }`,
    );
  }
  return to;
}

function withRunState(
  projection: RunProjectionV2,
  state: RunProjectionV2["state"],
  facts: RunGateFacts,
): RunProjectionV2 {
  return { ...projection, state: gatedRunState(projection.state, state, facts) };
}

function updateAttempt(
  projection: RunProjectionV2,
  workerId: WorkerId,
  attemptId: AttemptId,
  update: (attempt: WorkerAttemptProjection) => WorkerAttemptProjection,
  workerState?: WorkerProjection["state"],
): RunProjectionV2 {
  return updateWorker(projection, workerId, (worker) => {
    let found = false;
    const attempts = worker.attempts.map((attempt) => {
      if (attempt.attemptId !== attemptId) return attempt;
      found = true;
      return update(attempt);
    });
    if (!found) throw new LedgerIntegrityError(`Attempt ${attemptId} is missing from worker ${workerId}`);
    return { ...worker, ...(workerState ? { state: workerState } : {}), attempts };
  });
}

function appendAmbiguity(
  projection: RunProjectionV2,
  ambiguity: AmbiguityProjection,
): RunProjectionV2 {
  if (projection.unresolvedAmbiguities.some(
    (candidate) => candidate.ambiguityId === ambiguity.ambiguityId,
  )) {
    throw new LedgerIntegrityError(`Duplicate ambiguity ID ${ambiguity.ambiguityId}`);
  }
  return {
    ...projection,
    state: gatedRunState(projection.state, "ambiguous", runGateFacts()),
    unresolvedAmbiguities: [...projection.unresolvedAmbiguities, ambiguity],
  };
}

function assertAttemptGeneration(
  attempt: WorkerAttemptProjection,
  workerAttemptGeneration: string,
  supervisorGeneration: string,
): void {
  if (attempt.workerAttemptGeneration !== workerAttemptGeneration
    || attempt.supervisorGeneration !== supervisorGeneration) {
    throw new RunConflictError(`Attempt ${attempt.attemptId} generation is stale`);
  }
}

function assertFoldGeneration(
  attempt: WorkerAttemptProjection,
  workerAttemptGeneration: string,
  supervisorGeneration: string,
): void {
  if (attempt.workerAttemptGeneration !== workerAttemptGeneration
    || attempt.supervisorGeneration !== supervisorGeneration) {
    throw new LedgerIntegrityError(`Attempt ${attempt.attemptId} event generation is stale`);
  }
}

function isAttemptProcessActive(attempt: WorkerAttemptProjection): boolean {
  return (
    (attempt.state === "pending" || attempt.state === "running" || attempt.state === "blocked") &&
    attempt.pidStartEvidence.observedExitAt === null
  );
}

function assertContinuationSource(
  projection: RunProjectionV2,
  continuation: WorkerContinuationProvenance,
  sessionPath: string,
): void {
  const prior = requiredWorker(projection, continuation.priorWorkerId);
  const attempt = requiredCurrentAttempt(prior);
  if (attempt.attemptId !== continuation.priorAttemptId) {
    throw new RunConflictError("Continuation must link the prior worker's current attempt");
  }
  if (prior.state !== "done" && prior.state !== "failed" && prior.state !== "cancelled") {
    throw new RunConflictError(`Continuation prior worker is not terminal: ${prior.state}`);
  }
  if (attempt.state !== prior.state) {
    throw new LedgerIntegrityError("Continuation prior worker and attempt terminal states disagree");
  }
  if (attempt.pidStartEvidence.observedExitAt === null
    || attempt.pidStartEvidence.exitStatus === null) {
    throw new RunConflictError("Continuation prior process death is not durably observed");
  }
  if (attempt.session.file !== continuation.sessionFile) {
    throw new RunConflictError("Continuation session file does not match the prior attempt");
  }
  if (attempt.session.leaseGeneration !== continuation.priorLeaseGeneration) {
    throw new RunConflictError("Continuation prior lease generation does not match the ledger");
  }
  if (attempt.session.path !== sessionPath) {
    throw new RunConflictError("Continuation session directory does not match the prior attempt");
  }
}

export function expectedWorkerCadenceMs(budget: WorkerBudget): number {
  const wallClock = budgetValue(budget.maxWallClockMs);
  return wallClock === undefined ? 60_000 : Math.max(1_000, Math.min(60_000, Math.floor(wallClock / 4)));
}

function workerBudgetExceeded(attempt: WorkerAttemptProjection, budget: WorkerBudget): boolean {
  return (
    budgetExceeded(attempt.usage.tokens, budget.maxTokens)
    || budgetExceeded(attempt.usage.costUsd, budget.maxCostUsd)
    || budgetExceeded(attempt.usage.outputBytes, budget.maxOutputBytes)
  );
}

function existingRun(existing: RunProjectionV2, fingerprintValue: string): StartRunResult {
  if (existing.requestFingerprint !== fingerprintValue) {
    throw new RunConflictError(
      `Run request key ${existing.requestKey} already has a different immutable contract`,
    );
  }
  return { kind: "existing", projection: existing };
}

function eventPath(runDirectory: string, event: WorkflowLedgerEventV1): string {
  return join(
    runDirectory,
    "events",
    `${formatSequence(event.seq)}.${safePathPart(event.eventId)}.json`,
  );
}

function formatSequence(sequence: number): string {
  return sequence.toString().padStart(20, "0");
}

function fingerprint(value: unknown): string {
  return sha256(canonicalJson(value));
}

function validateRequestKey(value: string): string {
  requireNonEmpty(value, "request key");
  if (value.length > 512) throw new TypeError("Request key exceeds 512 characters");
  return value;
}

function strictRecord(
  value: unknown,
  allowed: readonly string[],
  name: string,
): Record<string, unknown> {
  if (typeof value !== "object" || value === null || Array.isArray(value)) {
    throw new TypeError(`${name} must be an object`);
  }
  const object = value as Record<string, unknown>;
  const unknown = Object.keys(object).filter((key) => !allowed.includes(key));
  if (unknown.length > 0) throw new TypeError(`${name} contains unknown fields: ${unknown.join(", ")}`);
  return object;
}

function recordValue(object: Record<string, unknown>, key: string): Record<string, unknown> {
  return strictRecord(object[key], Object.keys((object[key] ?? {}) as object), key);
}

function nonEmptyString(object: Record<string, unknown>, key: string): string {
  const value = object[key];
  if (typeof value !== "string" || value.length === 0) throw new TypeError(`${key} must be a non-empty string`);
  return value;
}

function nullableString(object: Record<string, unknown>, key: string): string | null {
  const value = object[key];
  if (value === null) return null;
  if (typeof value !== "string" || value.length === 0) throw new TypeError(`${key} must be null or a non-empty string`);
  return value;
}

function boundedString(
  object: Record<string, unknown>,
  key: string,
  maxLength: number,
): string {
  const value = nonEmptyString(object, key);
  if (value.trim().length === 0) throw new TypeError(`${key} must not be blank`);
  if (value.length > maxLength) throw new TypeError(`${key} exceeds ${maxLength} characters`);
  return value;
}

function nullableBoundedString(
  object: Record<string, unknown>,
  key: string,
  maxLength: number,
): string | null {
  if (object[key] === null) return null;
  return boundedString(object, key, maxLength);
}

function stringArray(object: Record<string, unknown>, key: string): readonly string[] {
  const value = object[key];
  if (!Array.isArray(value)) throw new TypeError(`${key} must be an array`);
  return value.map((item) => {
    if (typeof item !== "string" || item.length === 0) throw new TypeError(`${key} entries must be non-empty strings`);
    return item;
  });
}

function nonEmptyStringArray(object: Record<string, unknown>, key: string): readonly string[] {
  const value = stringArray(object, key);
  if (value.length === 0) throw new TypeError(`${key} must not be empty`);
  return value;
}

function numberValue(object: Record<string, unknown>, key: string): number {
  const value = object[key];
  if (typeof value !== "number" || !Number.isFinite(value)) throw new TypeError(`${key} must be a number`);
  return value;
}

function nonNegativeInteger(object: Record<string, unknown>, key: string): number {
  const value = numberValue(object, key);
  if (!Number.isSafeInteger(value) || value < 0) throw new TypeError(`${key} must be a non-negative integer`);
  return value;
}

function nonNegativeNumber(object: Record<string, unknown>, key: string): number {
  const value = numberValue(object, key);
  if (value < 0) throw new TypeError(`${key} must be non-negative`);
  return value;
}

function positiveInteger(object: Record<string, unknown>, key: string): number {
  const value = numberValue(object, key);
  if (!Number.isSafeInteger(value) || value <= 0) throw new TypeError(`${key} must be a positive integer`);
  return value;
}

function positiveIntegerValue(value: number, name: string): number {
  if (!Number.isSafeInteger(value) || value <= 0) {
    throw new TypeError(`${name} must be a positive integer`);
  }
  return value;
}

function nullablePositiveInteger(object: Record<string, unknown>, key: string): number | null {
  if (object[key] === null) return null;
  return positiveInteger(object, key);
}

function booleanValue(object: Record<string, unknown>, key: string): boolean {
  const value = object[key];
  if (typeof value !== "boolean") throw new TypeError(`${key} must be a boolean`);
  return value;
}

function nullableBoolean(object: Record<string, unknown>, key: string): boolean | null {
  return object[key] === null ? null : booleanValue(object, key);
}

function nullableReportSchemaPolicy(object: Record<string, unknown>, key: string): ReportSchemaPolicy | null {
  if (object[key] === undefined || object[key] === null) return null;
  return enumValue(object, key, ["strict", "preferred", "advisory"] as const);
}

function nullableNonNegativeInteger(
  object: Record<string, unknown>,
  key: string,
): number | null {
  return object[key] === null ? null : nonNegativeInteger(object, key);
}

function positiveNumber(object: Record<string, unknown>, key: string): number {
  const value = numberValue(object, key);
  if (value <= 0) throw new TypeError(`${key} must be positive`);
  return value;
}

function enumValue<T extends string>(
  object: Record<string, unknown>,
  key: string,
  allowed: readonly T[],
): T {
  const value = nonEmptyString(object, key);
  if (!allowed.includes(value as T)) throw new TypeError(`${key} is invalid`);
  return value as T;
}

function prefixedId(object: Record<string, unknown>, key: string, prefix: string): string {
  const value = nonEmptyString(object, key);
  if (!value.startsWith(`${prefix}_`)) throw new TypeError(`${key} must start with ${prefix}_`);
  return value;
}

function nullablePrefixedId(
  object: Record<string, unknown>,
  key: string,
  prefix: string,
): string | null {
  if (object[key] === null) return null;
  return prefixedId(object, key, prefix);
}

function hashValue(object: Record<string, unknown>, key: string): string {
  const value = nonEmptyString(object, key);
  if (!/^sha256:[0-9a-f]{64}$/.test(value)) throw new TypeError(`${key} must be a sha256 hash`);
  return value;
}

function requireSha256(value: string, label: string): string {
  const normalized = requireNonEmpty(value, label);
  if (!/^sha256:[a-f0-9]{64}$/.test(normalized)) throw new TypeError(`${label} must be a SHA-256 hash`);
  return normalized;
}

function nullableHash(object: Record<string, unknown>, key: string): string | null {
  if (object[key] === null) return null;
  return hashValue(object, key);
}

function isoTimestamp(object: Record<string, unknown>, key: string): string {
  const value = nonEmptyString(object, key);
  if (!Number.isFinite(Date.parse(value))) throw new TypeError(`${key} must be an ISO timestamp`);
  return value;
}

function nullableIsoTimestamp(object: Record<string, unknown>, key: string): string | null {
  if (object[key] === null) return null;
  return isoTimestamp(object, key);
}

function absolutePath(object: Record<string, unknown>, key: string): string {
  const value = nonEmptyString(object, key);
  if (!isAbsolute(value)) throw new TypeError(`${key} must be absolute`);
  return normalize(value);
}

function nullableAbsolutePath(object: Record<string, unknown>, key: string): string | null {
  if (object[key] === null) return null;
  return absolutePath(object, key);
}

function requireNonEmpty(value: string, name: string): string {
  if (value.length === 0) throw new TypeError(`${name} is required`);
  return value;
}

function requireBoundedText(value: string, name: string, maxLength: number): string {
  if (value.trim().length === 0) throw new TypeError(`${name} is required`);
  if (value.length > maxLength) throw new TypeError(`${name} exceeds ${maxLength} characters`);
  return value;
}

function safePathPart(value: string): string {
  return value.replace(/[^A-Za-z0-9._-]/g, "_");
}

/** Worker output bytes are measured on Pi's own event stream, not on the answer text. */
export const OUTPUT_BYTE_UNIT_NOTE =
  "max_output_bytes counts every byte Pi writes on its event stream, not the size of the "
  + "worker's answer; a trivial worker already emits several kilobytes before replying";

/**
 * A refusal a caller can act on in one step: what was asked for, and what is actually left.
 * Without the remaining figure the only recovery is to guess a smaller number and retry.
 */
function reservationRefusal(
  dimension: string,
  field: string,
  requested: import("./workflow-types.js").BudgetLimit,
  remaining: import("./workflow-types.js").BudgetLimit,
): string {
  const show = (limit: import("./workflow-types.js").BudgetLimit): string =>
    limit.kind === "unlimited" ? "unlimited" : String(limit.value);
  return `Run ${dimension} reservation budget is exhausted: requested ${field}=${show(requested)}, `
    + `remaining=${show(remaining)}`;
}