import { describe, test, expect } from "bun:test";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";

import { type } from "arktype";

import { generateKeyPair } from "@intx/crypto";
import { hexDecode, hexEncode, signalName } from "@intx/types";
import type { InferenceSource } from "@intx/types/runtime";
import type { RepoId, RepoStore } from "@intx/hub-sessions";
import { StaleInboxEnqueueError } from "@intx/hub-sessions";
import type { EnqueueInboxOutcome } from "@intx/hub-sessions";

import {
  createWorkflowSupervisor,
  type DrainTimeoutAccumulator,
  type DrainTimeoutAccumulatorFactory,
  type DrainTimeoutOpts,
  type InboxPrimitives,
  type MailBusBindings,
  type SubprocessSpawner,
  type SubprocessHandle,
  type SignedPayload,
  type WorkflowSupervisorBindings,
} from "./index";
import {
  assembleCredentialsSnapshot,
  defaultStepRepoId,
  hashGrants,
  STEP_GRANTS_PATH,
} from "./credentials";
import { commitCancelRequested } from "./cancel-signing";
import {
  createControlChannelSender,
  createEventChannelSender,
  ControlPayload,
  receiveControlChannel,
  SignedEnvelope,
  generateHmacKey,
  generateChannelId,
  type FrameReader,
  type NdjsonReader,
  type NdjsonWriter,
} from "../ipc/index";

/**
 * Parse the `runId` carried on each `trigger.fire` frame the
 * supervisor wrote to the in-memory child control stream. Validates
 * every signed envelope's payload through the canonical `ControlPayload`
 * narrow so the helper does not need to `as`-cast at the boundary.
 */
function parseTriggerFireRunIds(lines: readonly string[]): string[] {
  const ids: string[] = [];
  for (const line of lines) {
    if (!line.includes("trigger.fire")) continue;
    const raw: unknown = JSON.parse(line);
    const signed = SignedEnvelope(raw);
    if (signed instanceof type.errors) continue;
    const payload = ControlPayload(signed.envelope.payload);
    if (payload instanceof type.errors) continue;
    if (payload.type !== "trigger.fire") continue;
    ids.push(payload.data.runId);
  }
  return ids;
}

/**
 * The frame `type`s the supervisor writes to the child control stream, in
 * write order. Used to assert the per-run grants barrier orders a
 * `grants-updated` push STRICTLY before the run's `trigger.fire`.
 */
function parseControlFrameTypes(lines: readonly string[]): string[] {
  const types: string[] = [];
  for (const line of lines) {
    const raw: unknown = JSON.parse(line);
    const signed = SignedEnvelope(raw);
    if (signed instanceof type.errors) continue;
    const payload = ControlPayload(signed.envelope.payload);
    if (payload instanceof type.errors) continue;
    types.push(payload.type);
  }
  return types;
}

/**
 * Parse every `signal.deliver` frame in the supervisor-to-child stream,
 * returning the `{ signalName, signalId }` for each.
 */
function parseSignalDelivers(
  lines: readonly string[],
): { signalName: string; signalId: string; payload: unknown }[] {
  const out: { signalName: string; signalId: string; payload: unknown }[] = [];
  for (const line of lines) {
    if (!line.includes("signal.deliver")) continue;
    const raw: unknown = JSON.parse(line);
    const signed = SignedEnvelope(raw);
    if (signed instanceof type.errors) continue;
    const payload = ControlPayload(signed.envelope.payload);
    if (payload instanceof type.errors) continue;
    if (payload.type !== "signal.deliver") continue;
    out.push({
      signalName: payload.data.signalName,
      signalId: payload.data.signalId,
      payload: payload.data.payload,
    });
  }
  return out;
}

function createNoopDrainAccumulator(): DrainTimeoutAccumulator {
  return {
    start() {
      // noop
    },
    pause() {
      // noop
    },
    resume() {
      // noop
    },
    stop() {
      // noop
    },
    accumulatedMs() {
      return 0;
    },
    get escalated() {
      return false;
    },
    async disposed() {
      // noop
    },
  };
}

function parseSourcesUpdatedFrames(
  lines: readonly string[],
): { sources: InferenceSource[]; defaultSource: string }[] {
  const out: { sources: InferenceSource[]; defaultSource: string }[] = [];
  for (const line of lines) {
    if (!line.includes("sources-updated")) continue;
    const raw: unknown = JSON.parse(line);
    const signed = SignedEnvelope(raw);
    if (signed instanceof type.errors) continue;
    const payload = ControlPayload(signed.envelope.payload);
    if (payload instanceof type.errors) continue;
    if (payload.type !== "sources-updated") continue;
    out.push({
      sources: payload.data.sources,
      defaultSource: payload.data.defaultSource,
    });
  }
  return out;
}

function parseCredentialsUpdatedFrames(lines: readonly string[]) {
  return lines.flatMap((line) => {
    if (!line.includes("credentials-updated")) return [];
    const raw: unknown = JSON.parse(line);
    const signed = SignedEnvelope(raw);
    if (signed instanceof type.errors) return [];
    const payload = ControlPayload(signed.envelope.payload);
    if (payload instanceof type.errors) return [];
    if (payload.type !== "credentials-updated") return [];
    return [payload.data.delivery];
  });
}

const CancelRequestedBlob = type({
  type: "string",
  seq: "number",
  origin: "string",
  reason: "string",
  signature: {
    principalKind: "string",
    sig: "string",
  },
  "+": "ignore",
});

function readCancelRequestedBlob(
  raw: string,
): typeof CancelRequestedBlob.infer {
  const parsed: unknown = JSON.parse(raw);
  const validated = CancelRequestedBlob(parsed);
  if (validated instanceof type.errors) {
    throw new Error(`unexpected blob shape: ${validated.summary}`);
  }
  return validated;
}

async function makeTempDir(prefix: string): Promise<string> {
  const dir = await fs.mkdtemp(path.join(os.tmpdir(), prefix));
  return dir;
}

function createMemoryNdjsonStream() {
  const buffer: string[] = [];
  let waiter: (() => void) | null = null;
  let done = false;
  function wake() {
    const w = waiter;
    waiter = null;
    if (w) w();
  }
  const reader: NdjsonReader = {
    read(): AsyncIterableIterator<string> {
      return (async function* () {
        while (true) {
          if (buffer.length > 0) {
            const next = buffer.shift();
            if (next === undefined) {
              throw new Error("buffer shift returned undefined");
            }
            yield next;
            continue;
          }
          if (done) return;
          await new Promise<void>((resolve) => {
            waiter = resolve;
          });
        }
      })();
    },
  };
  const writer: NdjsonWriter = {
    write(line: string) {
      buffer.push(line.replace(/\n$/, ""));
      wake();
    },
  };
  return {
    writer,
    reader,
    inject(line: string) {
      buffer.push(line.replace(/\n$/, ""));
      wake();
    },
    flushed(): readonly string[] {
      return buffer.slice();
    },
    close() {
      done = true;
      wake();
    },
  };
}

function createMemoryFrameStream() {
  const buffer: Uint8Array[] = [];
  let waiter: (() => void) | null = null;
  let done = false;
  function wake() {
    const w = waiter;
    waiter = null;
    if (w) w();
  }
  const reader: FrameReader = {
    read(): AsyncIterableIterator<Uint8Array> {
      return (async function* () {
        while (true) {
          if (buffer.length > 0) {
            const next = buffer.shift();
            if (next === undefined) {
              throw new Error("frame buffer shift returned undefined");
            }
            yield next;
            continue;
          }
          if (done) return;
          await new Promise<void>((resolve) => {
            waiter = resolve;
          });
        }
      })();
    },
  };
  return {
    reader,
    inject(bytes: Uint8Array) {
      buffer.push(bytes);
      wake();
    },
    close() {
      done = true;
      wake();
    },
  };
}

function createMockMailBus(): MailBusBindings & {
  registered(): readonly string[];
  deliver(address: string, message: Uint8Array): void;
} {
  const registered: string[] = [];
  const subscribers = new Map<
    string,
    Set<(rawMessage: Uint8Array) => Promise<void>>
  >();
  return {
    registerAddress(address: string) {
      registered.push(address);
    },
    unregisterAddress(address: string) {
      const idx = registered.lastIndexOf(address);
      if (idx >= 0) registered.splice(idx, 1);
      subscribers.delete(address);
    },
    subscribeMailForAddress(
      address: string,
      handler: (rawMessage: Uint8Array) => Promise<void>,
    ) {
      let set = subscribers.get(address);
      if (set === undefined) {
        set = new Set();
        subscribers.set(address, set);
      }
      set.add(handler);
      return () => {
        const current = subscribers.get(address);
        current?.delete(handler);
      };
    },
    sendOutbound() {
      throw new Error("sendOutbound not exercised in this test");
    },
    registered(): readonly string[] {
      return registered.slice();
    },
    deliver(address: string, message: Uint8Array) {
      const set = subscribers.get(address);
      if (set === undefined) return;
      for (const handler of set) void handler(message).catch(() => undefined);
    },
  };
}

// A mail bus that exposes the durable settlement of a delivery via `settle`
// (the subscribed handler's returned promise), so the ack/withhold mapping can
// be asserted directly. It RETAINS the subscribed handler after the disposer
// runs, so a test can drive `onMailMessage`'s own phase gate after teardown --
// the belt that guards the racy "mail arrives while the deployment is stopping"
// window. Structurally a superset of `createMockMailBus`'s shape so it drops
// into the same spawn fixture.
function createSettleableMailBus(): ReturnType<typeof createMockMailBus> & {
  settle(address: string, message: Uint8Array): Promise<void>;
} {
  const registered: string[] = [];
  const handlers = new Map<string, (m: Uint8Array) => Promise<void>>();
  return {
    registerAddress(address: string) {
      registered.push(address);
    },
    unregisterAddress(address: string) {
      const idx = registered.lastIndexOf(address);
      if (idx >= 0) registered.splice(idx, 1);
    },
    subscribeMailForAddress(
      address: string,
      handler: (rawMessage: Uint8Array) => Promise<void>,
    ) {
      handlers.set(address, handler);
      return () => undefined;
    },
    sendOutbound() {
      throw new Error("sendOutbound not exercised in this test");
    },
    registered(): readonly string[] {
      return registered.slice();
    },
    deliver(address: string, message: Uint8Array) {
      const handler = handlers.get(address);
      if (handler === undefined) return;
      void handler(message).catch(() => undefined);
    },
    settle(address: string, message: Uint8Array): Promise<void> {
      const handler = handlers.get(address);
      if (handler === undefined) {
        throw new Error(`no subscriber for ${address}`);
      }
      return handler(message);
    },
  };
}

// Wrap the in-memory inbox primitives, replacing only `enqueueInbox` with a
// programmable stub so a test can drive each enqueue outcome (a fresh enqueue,
// an already-present message, a transient failure, a stale refusal) through
// the supervisor's real mail-arrival path.
function inboxPrimitivesWithEnqueue(
  enqueue: InboxPrimitives["enqueueInbox"],
): MemoryInboxPrimitives {
  return { ...createMemoryInboxPrimitives(), enqueueInbox: enqueue };
}

function enqueuedOutcome(args: {
  address: string;
  messageId: string;
  receivedAt: number;
  mailAuditRef: { store: string; path: string };
}): EnqueueInboxOutcome {
  return {
    outcome: "enqueued",
    commitSha: "memory",
    inboxKey: `${String(args.receivedAt)}-${args.messageId}`,
    envelope: {
      messageId: args.messageId,
      receivedAt: args.receivedAt,
      address: args.address,
      mailAuditRef: args.mailAuditRef,
    },
  };
}

/**
 * Create a stub `RepoStore` that satisfies only the subset of the
 * interface the supervisor reaches into in this commit. The
 * supervisor calls `getRepoDir` (credentials assembly) and
 * `writeTreePreservingPrefix` (cancel signing). All other methods
 * throw so a test that accidentally triggers an untested code path
 * surfaces a precise failure.
 */
function createStubRepoStore(opts: {
  baseDir: string;
  onWrite?: (args: {
    principal: { kind: string };
    repoId: RepoId;
    ref: string;
    preservePrefix: string;
    message: string;
    files: Record<string, string | Uint8Array>;
  }) => void;
  /**
   * Called at the START of `writeTreePreservingPrefix`, before the merge
   * callback runs. May throw to simulate a substrate write failure (e.g. a
   * lock or read fault) reaching the caller unmerged -- the throw fires
   * before any sentinel-skip logic inside the merge, so it models a genuine
   * failure regardless of the prefix's contents.
   */
  beforeWrite?: (args: {
    preservePrefix: string;
    message: string;
  }) => void | Promise<void>;
  /**
   * When true, the stub carries committed files across
   * `writeTreePreservingPrefix` invocations keyed by (repoId.id, ref,
   * preservePrefix), so a sequence of appends sees the prior commits
   * in its merge callback's `existing` map. Off by default to keep
   * tests that assert per-call shape from racing across calls.
   */
  statefulWrites?: boolean;
}): RepoStore {
  const committed = new Map<string, Map<string, Uint8Array>>();
  function keyFor(repoId: RepoId, ref: string, preservePrefix: string): string {
    return `${repoId.kind}/${repoId.id}\x00${ref}\x00${preservePrefix}`;
  }
  const stub: Partial<RepoStore> = {
    getRepoDir(repoId: RepoId): string {
      return path.join(opts.baseDir, repoId.kind, repoId.id);
    },
    async writeTreePreservingPrefix(principal, repoId, ref, args) {
      await opts.beforeWrite?.({
        preservePrefix: args.preservePrefix,
        message: args.message,
      });
      const key = keyFor(repoId, ref, args.preservePrefix);
      const existing =
        opts.statefulWrites === true
          ? (committed.get(key) ?? new Map<string, Uint8Array>())
          : new Map<string, Uint8Array>();
      const files = await args.merge(existing);
      opts.onWrite?.({
        principal,
        repoId,
        ref,
        preservePrefix: args.preservePrefix,
        message: args.message,
        files,
      });
      if (opts.statefulWrites === true) {
        const next = new Map<string, Uint8Array>();
        for (const [path, bytes] of Object.entries(files)) {
          if (!path.startsWith(args.preservePrefix)) continue;
          next.set(
            path,
            typeof bytes === "string" ? new TextEncoder().encode(bytes) : bytes,
          );
        }
        committed.set(key, next);
      }
      return { commitSha: "deadbeefcafef00d", newlyTerminalRuns: [] };
    },
  };
  // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- test stub; only the subset the supervisor invokes is implemented and a missing method throws via the proxy below
  return new Proxy(stub as RepoStore, {
    get(target, prop, receiver) {
      const value = Reflect.get(target, prop, receiver);
      if (value !== undefined) return value;
      return () => {
        throw new Error(
          `stub RepoStore: ${String(prop)} not implemented for this test`,
        );
      };
    },
  });
}

/**
 * Per-address claim-check state for the in-memory inbox stub. Mirrors
 * the substrate's three subdirectories (`inbox`, `processing`,
 * `consumed`) so a sequence of `enqueueInbox` / `dequeueToProcessing`
 * / `markConsumed` / `replayProcessingToInbox` calls is observable
 * without standing up a real git repo.
 */
type MemoryInboxEntry = {
  messageId: string;
  receivedAt: number;
  mailAuditRef: { store: string; path: string };
  rawMessage?: string;
  rejection?: { code: string; message: string };
};

export type MemoryInboxState = {
  inbox: Map<string, MemoryInboxEntry>;
  processing: Map<string, MemoryInboxEntry>;
  consumed: Map<string, MemoryInboxEntry>;
};

export type MemoryInboxPrimitives = InboxPrimitives & {
  /** Snapshot the in-memory state for a given address (testing only). */
  snapshot(address: string): MemoryInboxState;
};

function filenameKey(receivedAt: number, messageId: string): string {
  return `${String(receivedAt)}-${messageId}`;
}

function createMemoryInboxPrimitives(): MemoryInboxPrimitives {
  const byAddress = new Map<string, MemoryInboxState>();
  function getOrCreate(address: string): MemoryInboxState {
    let entry = byAddress.get(address);
    if (entry === undefined) {
      entry = {
        inbox: new Map(),
        processing: new Map(),
        consumed: new Map(),
      };
      byAddress.set(address, entry);
    }
    return entry;
  }
  return {
    snapshot(address: string): MemoryInboxState {
      return getOrCreate(address);
    },
    async enqueueInbox(_store, _principal, _repoId, args) {
      const state = getOrCreate(args.address);
      const key = filenameKey(args.receivedAt, args.messageId);
      // Mirror the real `enqueueInbox` contract: an already-present messageId
      // is a returned outcome (ack-worthy), not a throw.
      if (state.consumed.has(args.messageId)) {
        return { outcome: "already-present", reason: "consumed" };
      }
      for (const existingKey of state.inbox.keys()) {
        if (existingKey.endsWith(`-${args.messageId}`)) {
          return { outcome: "already-present", reason: "already_inbox" };
        }
      }
      for (const existingKey of state.processing.keys()) {
        if (existingKey.endsWith(`-${args.messageId}`)) {
          return { outcome: "already-present", reason: "processing" };
        }
      }
      const envelope: MemoryInboxEntry = {
        messageId: args.messageId,
        receivedAt: args.receivedAt,
        mailAuditRef: args.mailAuditRef,
        ...(args.rawMessage !== undefined
          ? { rawMessage: args.rawMessage }
          : {}),
      };
      state.inbox.set(key, envelope);
      return {
        outcome: "enqueued",
        commitSha: "memory-inbox",
        inboxKey: key,
        envelope: {
          messageId: args.messageId,
          receivedAt: args.receivedAt,
          address: args.address,
          mailAuditRef: args.mailAuditRef,
        },
      };
    },
    async dequeueToProcessing(_store, _principal, _repoId, address) {
      const state = getOrCreate(address);
      const entries = [...state.inbox.entries()].sort(([, a], [, b]) => {
        if (a.receivedAt !== b.receivedAt) return a.receivedAt - b.receivedAt;
        if (a.messageId < b.messageId) return -1;
        if (a.messageId > b.messageId) return 1;
        return 0;
      });
      if (entries.length === 0) return null;
      const head = entries[0];
      if (head === undefined) throw new Error("unreachable");
      const [key, envelope] = head;
      state.inbox.delete(key);
      state.processing.set(key, envelope);
      return {
        commitSha: "memory-inbox",
        key,
        envelope: {
          messageId: envelope.messageId,
          receivedAt: envelope.receivedAt,
          address,
          mailAuditRef: envelope.mailAuditRef,
          ...(envelope.rawMessage !== undefined
            ? { rawMessage: envelope.rawMessage }
            : {}),
        },
      };
    },
    async markConsumed(_store, _principal, _repoId, args) {
      const state = getOrCreate(args.address);
      let foundKey: string | null = null;
      let envelope: MemoryInboxEntry | null = null;
      for (const [key, value] of state.processing) {
        if (value.messageId === args.messageId) {
          foundKey = key;
          envelope = value;
          break;
        }
      }
      if (foundKey === null || envelope === null) {
        throw new Error(
          `claim_check_processing_not_found: ${args.address} ${args.messageId}`,
        );
      }
      state.processing.delete(foundKey);
      const consumedEntry: MemoryInboxEntry = {
        ...envelope,
        ...(args.rejection !== undefined ? { rejection: args.rejection } : {}),
      };
      state.consumed.set(args.messageId, consumedEntry);
      return {
        commitSha: "memory-inbox",
        envelope: {
          messageId: envelope.messageId,
          receivedAt: envelope.receivedAt,
          address: args.address,
          runId: args.runId,
          consumedAt: args.consumedAt,
          mailAuditRef: envelope.mailAuditRef,
          ...(args.rejection !== undefined
            ? { rejection: args.rejection }
            : {}),
        },
        watermark: 0,
        prunedMessageIds: [],
      };
    },
    async replayProcessingToInbox(_store, _principal, _repoId, address) {
      const state = getOrCreate(address);
      const replayedKeys: string[] = [];
      for (const [key, value] of state.processing) {
        if (state.inbox.has(key)) {
          throw new Error(
            `claim_check_replay_collision: inbox already has ${key}`,
          );
        }
        state.inbox.set(key, value);
        replayedKeys.push(key);
      }
      state.processing.clear();
      return { commitSha: "memory-inbox", replayedKeys };
    },
  };
}

async function buildBindings(opts: {
  baseDir: string;
  spawner: SubprocessSpawner;
  signSpy: (kind: string, payload: Uint8Array) => SignedPayload;
  mailBus: MailBusBindings;
  onWrite?: (args: {
    principal: { kind: string };
    repoId: RepoId;
    ref: string;
    preservePrefix: string;
    message: string;
    files: Record<string, string | Uint8Array>;
  }) => void;
  beforeWrite?: (args: {
    preservePrefix: string;
    message: string;
  }) => void | Promise<void>;
  statefulWrites?: boolean;
  inboxPrimitives?: InboxPrimitives;
}): Promise<WorkflowSupervisorBindings> {
  const repoStore = createStubRepoStore({
    baseDir: opts.baseDir,
    ...(opts.onWrite !== undefined ? { onWrite: opts.onWrite } : {}),
    ...(opts.beforeWrite !== undefined
      ? { beforeWrite: opts.beforeWrite }
      : {}),
    ...(opts.statefulWrites === true ? { statefulWrites: true } : {}),
  });
  return {
    repoStore,
    signAsPrincipal: async (kind, payload) => opts.signSpy(kind, payload),
    mailBus: opts.mailBus,
    subprocessSpawner: opts.spawner,
    binaryPath: "/fake/bin/workflow-child",
    substrateEnv: { DATA_DIR: opts.baseDir },
    dynamicSpawnEnv: () => ({}),
    workflowRunRepoId: { kind: "workflow-run", id: "deployment-x" },
    workflowRunRef: "refs/heads/main",
    deploymentId: "deployment-x",
    stepCount: 1,
    deploymentMailAddress: "deployment-x@example.com",
    readPrincipal: { kind: "supervisor" },
    deriveStepAddress: ({ deploymentId, stepId }) =>
      `${deploymentId}-${stepId}@example.com`,
    inboxPrimitives: opts.inboxPrimitives ?? createMemoryInboxPrimitives(),
  };
}

async function seedStepGrants(
  baseDir: string,
  repoId: RepoId,
  grants: unknown[],
): Promise<void> {
  const dir = path.join(baseDir, repoId.kind, repoId.id);
  await fs.mkdir(path.join(dir, "state"), { recursive: true });
  await fs.writeFile(
    path.join(dir, STEP_GRANTS_PATH),
    JSON.stringify({ grants }),
  );
}

describe("createWorkflowSupervisor", () => {
  test("factory accepts the documented WorkflowSupervisorBindings shape", async () => {
    const baseDir = await makeTempDir("supervisor-bindings-");
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error("spawner not invoked in this test");
      },
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus: createMockMailBus(),
    });
    const supervisor = createWorkflowSupervisor(bindings);
    expect(typeof supervisor.spawn).toBe("function");
    expect(typeof supervisor.requestCancel).toBe("function");
    expect(typeof supervisor.shutdown).toBe("function");
    expect(typeof supervisor.drain).toBe("function");
    expect(typeof supervisor.recycle).toBe("function");
    expect(supervisor.getCredentialsSnapshot()).toBeNull();
  });

  test("spawn completes the IPC handshake, registers mail, and pushes credentials", async () => {
    const baseDir = await makeTempDir("supervisor-spawn-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    // Deterministic IPC keypairs so the test's "child" side can sign
    // a `ready` frame the supervisor accepts. Two keypairs ride per
    // spawn: the supervisor's (downstream signing) and the child's
    // (upstream signing). The supervisor never sees the child's
    // private key; the child publishes its public half in the
    // `ready` frame's payload.
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();

    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let killed = false;

    let observedEnv: Record<string, string> | undefined;
    let observedBinary: string | undefined;
    const spawner: SubprocessSpawner = ({ binaryPath, env }) => {
      observedBinary = binaryPath;
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 4321,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          killed = true;
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };

    const mailBus = createMockMailBus();
    const inbox = createMemoryInboxPrimitives();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus,
      inboxPrimitives: inbox,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const eventsObserved: { type: string }[] = [];
    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: (event) => {
        eventsObserved.push({ type: event.type });
      },
    });
    // Drive the synthetic child side: wait until the spawner has
    // been invoked (so we have the channelId), then sign a `ready`
    // frame with the controlled IPC private key and inject it into
    // the child-to-supervisor stream.
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    // Wait for the supervisor to register the mail address on the
    // bus so the delivers below land inside the supervisor's
    // subscription handler -- a deliver before subscription is a
    // no-op against the mock bus.
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    // Deliver mail while the supervisor is still in `starting`; the
    // supervisor buffers it and replays it after `ready` lands.
    mailBus.deliver("deployment-x@example.com", new TextEncoder().encode("m1"));
    mailBus.deliver("deployment-x@example.com", new TextEncoder().encode("m2"));
    await childSender.send({
      type: "ready",
      data: {
        childPid: 4321,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });

    const result = await spawnPromise;
    expect(observedBinary).toBe("/fake/bin/workflow-child");
    expect(observedEnv).toMatchObject({
      DATA_DIR: baseDir,
      DEPLOYMENT_ID: "deployment-x",
      DEFINITION_HASH: "def-hash-abc",
      MAILBOX_ADDRESS: "deployment-x@example.com",
    });
    expect(observedEnv.IPC_CHANNEL_ID).toMatch(/^[0-9a-f]{32}$/);
    expect(observedEnv.IPC_HMAC_KEY).toMatch(/^[0-9a-f]{64}$/);
    expect(observedEnv.HOST_PUBKEY).toMatch(/^[0-9a-f]{64}$/);
    expect(observedEnv).not.toHaveProperty("HOST_PRIVATE_KEY");
    expect(result.pid).toBe(4321);
    expect(result.channelId).toBe(channelId);
    expect(result.credentialsSnapshot.steps).toHaveLength(1);
    expect(result.credentialsSnapshot.steps[0]?.address).toBe(
      "deployment-x-step-1@example.com",
    );
    expect(mailBus.registered()).toContain("deployment-x@example.com");
    expect(supervisor.getCredentialsSnapshot()).not.toBeNull();

    // The first buffered mail fires the stable top-level run. The FIFO
    // claim-check pipeline holds the second until that run terminates, then
    // rejects it rather than issuing a second trigger.fire.
    const waitForTriggerFires = async (n: number): Promise<string[]> => {
      const deadline = Date.now() + 500;
      while (Date.now() < deadline) {
        const ids = parseTriggerFireRunIds(supervisorToChild.flushed());
        if (ids.length >= n) return ids;
        await new Promise((r) => setTimeout(r, 1));
      }
      return parseTriggerFireRunIds(supervisorToChild.flushed());
    };
    const firstFired = await waitForTriggerFires(1);
    expect(firstFired.length).toBeGreaterThanOrEqual(1);
    const firstRunId = firstFired[0];
    if (firstRunId === undefined) throw new Error("first runId missing");
    await childSender.send({
      type: "terminal.event",
      data: {
        runId: firstRunId,
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });
    const consumedDeadline = Date.now() + 500;
    while (Date.now() < consumedDeadline) {
      if (inbox.snapshot("deployment-x@example.com").consumed.size >= 2) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(parseTriggerFireRunIds(supervisorToChild.flushed())).toEqual([
      firstRunId,
    ]);

    await supervisor.shutdown();
    expect(killed).toBe(true);
    expect(mailBus.registered()).not.toContain("deployment-x@example.com");
  });

  // Stand up a spawned supervisor against a synthetic child, drive the
  // `ready` handshake, and return the pieces a per-run barrier test needs:
  // the child-side control sender (to inject `ready` / `terminal.event`),
  // the supervisor-to-child stream (to observe the frames the supervisor
  // wrote), the mock mail bus (to deliver inbound mail), and the inbox
  // primitives (to observe claim-check state). `onRunStart` is threaded
  // through the bindings so the dispatch loop runs the per-run barrier.
  async function spawnWithRunStart(opts: {
    baseDir: string;
    onRunStart?: WorkflowSupervisorBindings["onRunStart"];
    credentialDelivery?: WorkflowSupervisorBindings["credentialDelivery"];
    drainTimeoutAccumulatorFactory?: DrainTimeoutAccumulatorFactory;
    onWrite?: (args: {
      principal: { kind: string };
      repoId: RepoId;
      ref: string;
      preservePrefix: string;
      message: string;
      files: Record<string, string | Uint8Array>;
    }) => void;
    beforeWrite?: (args: {
      preservePrefix: string;
      message: string;
    }) => void | Promise<void>;
    inboxPrimitives?: MemoryInboxPrimitives;
    mailBus?: ReturnType<typeof createMockMailBus>;
  }) {
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });

    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      return {
        pid: 7777,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
    };

    const mailBus = opts.mailBus ?? createMockMailBus();
    const inboxPrimitives =
      opts.inboxPrimitives ?? createMemoryInboxPrimitives();
    const baseBindings = await buildBindings({
      baseDir: opts.baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus,
      inboxPrimitives,
      ...(opts.onWrite !== undefined ? { onWrite: opts.onWrite } : {}),
      ...(opts.beforeWrite !== undefined
        ? { beforeWrite: opts.beforeWrite }
        : {}),
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      ...(opts.onRunStart !== undefined ? { onRunStart: opts.onRunStart } : {}),
      ...(opts.credentialDelivery !== undefined
        ? { credentialDelivery: opts.credentialDelivery }
        : {}),
      ...(opts.drainTimeoutAccumulatorFactory !== undefined
        ? {
            drainTimeoutAccumulatorFactory: opts.drainTimeoutAccumulatorFactory,
          }
        : {}),
    };
    const supervisor = createWorkflowSupervisor(bindings);
    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-barrier",
      warmKeep: false,
      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    await childSender.send({
      type: "ready",
      data: {
        childPid: 7777,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;
    return {
      supervisor,
      childSender,
      supervisorToChild,
      mailBus,
      inboxPrimitives,
    };
  }

  test("dispatch pushes a per-run grants-updated before the run's trigger.fire", async () => {
    const baseDir = await makeTempDir("supervisor-barrier-ok-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    // The per-run sink reads the step's grants exactly as `spawn` does, but
    // per run rather than once per spawn. When `onRunStart` is wired the
    // spawn-time push is suppressed, so a `grants-updated` on the child
    // stream can only come from this per-run barrier.
    const runStartCalls: { runId: string; deploymentId: string }[] = [];
    const onRunStart: WorkflowSupervisorBindings["onRunStart"] = async (
      args,
    ) => {
      runStartCalls.push(args);
      return assembleCredentialsSnapshot({
        repoStore: createStubRepoStore({ baseDir }),
        principal: { kind: "supervisor" },
        stepOrder: ["step-1"],
        deploymentId: "deployment-x",
        deriveStepAddress: ({ deploymentId, stepId }) =>
          `${deploymentId}-${stepId}@example.com`,
      });
    };

    const wired = await spawnWithRunStart({ baseDir, onRunStart });

    // No grants-updated is pushed at spawn time when onRunStart is wired.
    expect(
      parseControlFrameTypes(wired.supervisorToChild.flushed()),
    ).not.toContain("grants-updated");

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("barrier-m1"),
    );

    const deadline = Date.now() + 1000;
    let firedRunId: string | undefined;
    while (Date.now() < deadline) {
      const ids = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
      if (ids.length >= 1) {
        firedRunId = ids[0];
        break;
      }
      await new Promise((r) => setTimeout(r, 1));
    }
    if (firedRunId === undefined) {
      throw new Error("no trigger.fire observed within deadline");
    }

    // The barrier is load-bearing on ordering: the run's grants-updated
    // must appear on the child stream STRICTLY before its trigger.fire.
    const frameTypes = parseControlFrameTypes(
      wired.supervisorToChild.flushed(),
    );
    const grantsIdx = frameTypes.indexOf("grants-updated");
    const triggerIdx = frameTypes.indexOf("trigger.fire");
    expect(grantsIdx).toBeGreaterThanOrEqual(0);
    expect(triggerIdx).toBeGreaterThanOrEqual(0);
    expect(grantsIdx).toBeLessThan(triggerIdx);

    // The sink was consulted once for this run with the supervisor's
    // deployment id stamped on.
    expect(runStartCalls).toEqual([
      { runId: firedRunId, deploymentId: "deployment-x" },
    ]);

    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: firedRunId, seq: 0, kind: "RunCompleted", at: "test" },
    });
    await wired.supervisor.shutdown();
  });

  test("the barrier pushes credentials-updated before trigger.fire when the deployment has credentials", async () => {
    const baseDir = await makeTempDir("supervisor-barrier-creds-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const onRunStart: WorkflowSupervisorBindings["onRunStart"] = async () =>
      assembleCredentialsSnapshot({
        repoStore: createStubRepoStore({ baseDir }),
        principal: { kind: "supervisor" },
        stepOrder: ["step-1"],
        deploymentId: "deployment-x",
        deriveStepAddress: ({ deploymentId, stepId }) =>
          `${deploymentId}-${stepId}@example.com`,
      });
    const delivery = {
      bindings: [
        { handle: "gh", credentialId: "cred_a", consumer: "tool:@acme/tools" },
      ],
      materials: [
        {
          credentialId: "cred_a",
          providerKey: "http",
          origin: "https://api.example.test",
          secret: "sk-real",
        },
      ],
    };

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart,
      credentialDelivery: delivery,
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("barrier-creds-m1"),
    );

    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (
        parseTriggerFireRunIds(wired.supervisorToChild.flushed()).length >= 1
      ) {
        break;
      }
      await new Promise((r) => setTimeout(r, 1));
    }

    // The material lands on the child's control stream STRICTLY before the
    // trigger, so a tool that resolves a credential on the first step already
    // has it in its cell.
    const frameTypes = parseControlFrameTypes(
      wired.supervisorToChild.flushed(),
    );
    const credsIdx = frameTypes.indexOf("credentials-updated");
    const triggerIdx = frameTypes.indexOf("trigger.fire");
    expect(credsIdx).toBeGreaterThanOrEqual(0);
    expect(triggerIdx).toBeGreaterThanOrEqual(0);
    expect(credsIdx).toBeLessThan(triggerIdx);

    // And the delivered material is the deployment's, verbatim.
    const deliveries = parseCredentialsUpdatedFrames(
      wired.supervisorToChild.flushed(),
    );
    expect(deliveries).toContainEqual(delivery);

    await wired.supervisor.shutdown();
  });

  test("a throwing onRunStart fails the run and never fires its trigger", async () => {
    const baseDir = await makeTempDir("supervisor-barrier-fail-");
    // No seedStepGrants: the sink throws regardless, standing in for any
    // barrier failure (a broken read, an unauthorized run). The run must
    // fail LOUDLY -- the trigger is never fired against absent grants.
    const onRunStart: WorkflowSupervisorBindings["onRunStart"] = async () => {
      throw new Error("synthetic grants-barrier failure");
    };

    const wired = await spawnWithRunStart({ baseDir, onRunStart });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("barrier-fail-m1"),
    );

    // The failed run is settled through the claim-check pipeline: the
    // message moves to `consumed`. Wait on that observable settle.
    const address = "deployment-x@example.com";
    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    // The barrier suppressed the trigger: NO trigger.fire and NO
    // grants-updated ever reached the child for the failed run.
    const frameTypes = parseControlFrameTypes(
      wired.supervisorToChild.flushed(),
    );
    expect(frameTypes).not.toContain("trigger.fire");
    expect(frameTypes).not.toContain("grants-updated");

    await wired.supervisor.shutdown();
  });

  // Inbound-mail ack/withhold mapping. `onMailMessage` returns a promise the
  // host propagates to the wire: resolve => send the durable-receipt ack,
  // reject => withhold it (the hub redelivers). These assert the mapping of
  // every enqueue disposition onto that boundary.
  test("durable receipt resolves for a fresh enqueue and for an already-present message", async () => {
    const baseDir = await makeTempDir("supervisor-ack-present-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    let mode: "enqueued" | "already-present" = "enqueued";
    const inboxPrimitives = inboxPrimitivesWithEnqueue(
      async (_store, _principal, _repoId, args) =>
        mode === "already-present"
          ? { outcome: "already-present", reason: "consumed" }
          : enqueuedOutcome(args),
    );
    const mailBus = createSettleableMailBus();
    const wired = await spawnWithRunStart({
      baseDir,
      inboxPrimitives,
      mailBus,
    });
    const address = "deployment-x@example.com";

    // A fresh enqueue is durably accepted -> the receipt resolves (ack).
    await expect(
      mailBus.settle(address, new TextEncoder().encode("m-fresh")),
    ).resolves.toBeUndefined();

    // An already-present message is also durably accounted for -> ack, and the
    // hub stops retrying (no infinite retry on a message the sidecar holds).
    mode = "already-present";
    await expect(
      mailBus.settle(address, new TextEncoder().encode("m-dup")),
    ).resolves.toBeUndefined();

    await wired.supervisor.shutdown();
  });

  test("durable receipt rejects for a transient failure and a stale refusal, and self-heals on redelivery", async () => {
    const baseDir = await makeTempDir("supervisor-withhold-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    let onEnqueue: (args: {
      address: string;
      messageId: string;
      receivedAt: number;
      mailAuditRef: { store: string; path: string };
    }) => EnqueueInboxOutcome = () => {
      throw new Error("disk exploded");
    };
    const inboxPrimitives = inboxPrimitivesWithEnqueue(
      async (_store, _principal, _repoId, args) => onEnqueue(args),
    );
    const mailBus = createSettleableMailBus();
    const wired = await spawnWithRunStart({
      baseDir,
      inboxPrimitives,
      mailBus,
    });
    const address = "deployment-x@example.com";

    // (a) A transient failure -> the receipt rejects, so no ack is sent and the
    // hub redelivers.
    await expect(
      mailBus.settle(address, new TextEncoder().encode("m-1")),
    ).rejects.toThrow(/disk exploded/);

    // (b) A stale refusal also withholds, and surfaces as its own loud type
    // rather than blending into generic failure noise.
    onEnqueue = () => {
      throw new StaleInboxEnqueueError("claim_check_stale_enqueue: synthetic");
    };
    let staleCause: unknown;
    try {
      await mailBus.settle(address, new TextEncoder().encode("m-1"));
    } catch (err) {
      staleCause = err;
    }
    expect(staleCause).toBeInstanceOf(StaleInboxEnqueueError);

    // (c) Self-heal: the same message, redelivered once the failure clears,
    // enqueues on a fresh receivedAt and the receipt resolves (ack).
    onEnqueue = (args) => enqueuedOutcome(args);
    await expect(
      mailBus.settle(address, new TextEncoder().encode("m-1")),
    ).resolves.toBeUndefined();

    await wired.supervisor.shutdown();
  });

  test("durable receipt rejects when the supervisor is not accepting mail (phase-drop)", async () => {
    const baseDir = await makeTempDir("supervisor-phase-drop-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    // enqueue would resolve if it were reached; the phase gate must reject
    // first, without touching the inbox.
    const inboxPrimitives = inboxPrimitivesWithEnqueue(
      async (_store, _principal, _repoId, args) => enqueuedOutcome(args),
    );
    const mailBus = createSettleableMailBus();
    const wired = await spawnWithRunStart({
      baseDir,
      inboxPrimitives,
      mailBus,
    });
    await wired.supervisor.shutdown();

    // The bus retained the handler, so this drives onMailMessage's own phase
    // gate: phase is "stopped", so it rejects BEFORE calling enqueue -> the
    // ack is withheld and the hub redelivers into a live generation later.
    await expect(
      mailBus.settle(
        "deployment-x@example.com",
        new TextEncoder().encode("m-late"),
      ),
    ).rejects.toThrow(/not accepted: supervisor phase/);
  });

  // Harness for the ready-timeout tests: an injected FakeTimer registry
  // (deterministic, per greybeard's ruling against real timers) plus a
  // controllable child whose control reader the test can close to model a
  // child that exits before signalling ready. `createdTimers` retains every
  // timer even after it is cleared, so a test can capture the ready deadline
  // and later assert it was cancelled.
  async function makeReadyTimeoutHarness(readyTimeoutMs: number) {
    type FakeTimer = { cb: () => void; ms: number; cancelled: boolean };
    const timers = new Set<FakeTimer>();
    const createdTimers: FakeTimer[] = [];

    const baseDir = await makeTempDir("supervisor-ready-timeout-");
    const supervisorIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    const killSignals: string[] = [];

    const spawner: SubprocessSpawner = ({ env: _env }) => ({
      pid: 5150,
      controlWriter: supervisorToChild.writer,
      controlReader: childToSupervisor.reader,
      eventReader: eventChildToSupervisor.reader,
      kill: (signal) => {
        killSignals.push(
          typeof signal === "string" ? signal : String(signal ?? ""),
        );
        childToSupervisor.close();
        eventChildToSupervisor.close();
        resolveExit?.(0);
      },
      exited,
    });

    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      readyTimeoutMs,
      setTimer: (cb, ms) => {
        const t: FakeTimer = { cb, ms, cancelled: false };
        timers.add(t);
        createdTimers.push(t);
        return t;
      },
      clearTimer: (handle) => {
        if (handle === null || typeof handle !== "object") return;
        for (const t of timers) {
          if (t === handle) {
            t.cancelled = true;
            timers.delete(t);
            return;
          }
        }
      },
    };
    const supervisor = createWorkflowSupervisor(bindings);

    // Resolve once the spawn has armed its ready deadline (which happens
    // after the spawner is invoked, so this also confirms the child spawned).
    async function waitForReadyDeadline(): Promise<FakeTimer> {
      for (;;) {
        const t = createdTimers.find((x) => x.ms === readyTimeoutMs);
        if (t !== undefined) return t;
        await new Promise((r) => setTimeout(r, 1));
      }
    }

    return { supervisor, killSignals, childToSupervisor, waitForReadyDeadline };
  }

  const readyTimeoutSpawnOpts = {
    stepOrder: ["step-1"],
    definitionHash: "def-hash-abc",
    warmKeep: false,

    onInferenceEvent: () => {
      /* unused in the ready-timeout tests */
    },
  };

  test("spawn times out, kills the child, rejects, and clears the ready deadline", async () => {
    const h = await makeReadyTimeoutHarness(7_777);
    // Never send `ready`. Spawn blocks on the handshake until the deadline.
    const spawnPromise = h.supervisor.spawn(readyTimeoutSpawnOpts);
    const readyDeadline = await h.waitForReadyDeadline();
    readyDeadline.cb();

    await expect(spawnPromise).rejects.toThrow(
      /child did not emit ready within 7777ms; killed/,
    );
    expect(h.killSignals).toContain("SIGTERM");
    // The unconditional deadline-timer clear ran on the timeout path.
    expect(readyDeadline.cancelled).toBe(true);
  });

  test("spawn clears the ready deadline when the child exits before ready", async () => {
    const h = await makeReadyTimeoutHarness(8_888);
    const spawnPromise = h.supervisor.spawn(readyTimeoutSpawnOpts);
    const readyDeadline = await h.waitForReadyDeadline();

    // The child exits before signalling ready: closing the control reader
    // ends `waitForReady`, rejecting the ready promise. Because the outcomes
    // are folded to values, the race resolves to the failed outcome rather
    // than rejecting, so the unconditional deadline-timer clear still runs.
    // A race that rejected here would skip the clear and leak an armed
    // deadline that keeps the event loop alive for up to readyTimeoutMs.
    h.childToSupervisor.close();

    await expect(spawnPromise).rejects.toThrow(
      /control channel ended before child emitted ready/,
    );
    expect(readyDeadline.cancelled).toBe(true);
  });

  // A spawn that throws AFTER the OS child is running but BEFORE the
  // supervisor reaches the ready handshake must not orphan the child or
  // leave the mail address registered. `shutdownInternal` owns that
  // teardown once the state record enters "starting"; the spawn body
  // routes every post-seam throw through it.
  async function makePreRegistrationFailureHarness(opts: {
    failSubscribe?: boolean;
    failDeriveStepAddress?: boolean;
  }) {
    const baseDir = await makeTempDir("supervisor-spawn-leak-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    const killSignals: string[] = [];
    const spawner: SubprocessSpawner = () => ({
      pid: 4321,
      controlWriter: supervisorToChild.writer,
      controlReader: childToSupervisor.reader,
      eventReader: eventChildToSupervisor.reader,
      kill: (signal) => {
        killSignals.push(
          typeof signal === "string" ? signal : String(signal ?? ""),
        );
        childToSupervisor.close();
        eventChildToSupervisor.close();
        resolveExit?.(0);
      },
      exited,
    });
    const mailBus = createMockMailBus();
    const bindingsMailBus: MailBusBindings = {
      ...mailBus,
      subscribeMailForAddress:
        opts.failSubscribe === true
          ? () => {
              throw new Error("injected subscribe failure");
            }
          : mailBus.subscribeMailForAddress,
    };
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus: bindingsMailBus,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      ...(opts.failDeriveStepAddress === true
        ? {
            deriveStepAddress: () => {
              throw new Error("injected deriveStepAddress failure");
            },
          }
        : {}),
    };
    return {
      supervisor: createWorkflowSupervisor(bindings),
      killSignals,
      registered: mailBus.registered,
    };
  }

  const preRegistrationSpawnOpts = {
    stepOrder: ["step-1"],
    definitionHash: "def-hash-abc",
    warmKeep: false,

    onInferenceEvent: () => {
      /* unused in the pre-registration failure tests */
    },
  };

  test("a spawn whose mail subscription throws kills the child and releases the address", async () => {
    const h = await makePreRegistrationFailureHarness({ failSubscribe: true });
    await expect(h.supervisor.spawn(preRegistrationSpawnOpts)).rejects.toThrow(
      "injected subscribe failure",
    );
    // The address was registered just before subscribe threw; the
    // teardown must unregister it so no orphaned registration survives.
    expect(h.registered()).toHaveLength(0);
    expect(h.killSignals.length).toBeGreaterThan(0);
  });

  test("a spawn whose credentials assembly throws kills the child", async () => {
    const h = await makePreRegistrationFailureHarness({
      failDeriveStepAddress: true,
    });
    await expect(h.supervisor.spawn(preRegistrationSpawnOpts)).rejects.toThrow(
      "injected deriveStepAddress failure",
    );
    expect(h.registered()).toHaveLength(0);
    expect(h.killSignals.length).toBeGreaterThan(0);
  });

  test("drain() forwards the `drain` control frame and arms a drainTimeout accumulator per in-flight run", async () => {
    const baseDir = await makeTempDir("supervisor-drain-arm-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();

    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 9999,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };

    // Mock accumulator factory the supervisor's `drain()` should
    // consult. Each invocation records the opts and returns a
    // controllable stub whose `start`/`stop` calls are visible to the
    // test. The factory shape matches `createDrainTimeoutAccumulator`
    // exactly so the supervisor binds it through the public
    // `WorkflowSupervisorBindings.drainTimeoutAccumulatorFactory`
    // slot.
    type StubAccumulator = DrainTimeoutAccumulator & {
      __opts: DrainTimeoutOpts;
      __startCount: number;
      __stopCount: number;
    };
    const stubs: StubAccumulator[] = [];
    const factory: DrainTimeoutAccumulatorFactory = (opts) => {
      const stub: StubAccumulator = {
        __opts: opts,
        __startCount: 0,
        __stopCount: 0,
        start() {
          this.__startCount += 1;
        },
        pause() {
          /* unused by the supervisor's arming path */
        },
        resume() {
          /* unused by the supervisor's arming path */
        },
        stop() {
          this.__stopCount += 1;
        },
        accumulatedMs() {
          return 0;
        },
        get escalated() {
          return false;
        },
        disposed() {
          return Promise.resolve();
        },
      };
      stubs.push(stub);
      return stub;
    };

    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      drainTimeoutAccumulatorFactory: factory,
      drainTimeoutMs: 7_500,
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: () => {
        /* unused in this test */
      },
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    // Two pre-ready messages. The supervisor's FIFO inbox queue
    // serializes dispatch: only one run is in-flight at a time. By
    // the time `drain()` is called below, the second message may
    // still be mid-dispatch behind the first's `markConsumed`. The
    // accumulator count reflects whichever in-flight runIds remain.
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("drain-msg-A"),
    );
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("drain-msg-B"),
    );
    await childSender.send({
      type: "ready",
      data: {
        childPid: 9999,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;

    // No accumulators armed yet -- drain has not been called.
    expect(stubs).toHaveLength(0);

    // Wait for the dispatch loop to dequeue the first buffered mail
    // and forward its `trigger.fire`. The H-S1 contract gates the
    // dispatch loop's first iteration on the spawn-time replayDone;
    // without polling for the forwarded frame the test would call
    // `drain()` while `cohortRunIds` is still empty and no
    // accumulator would arm.
    const triggerFireDeadline = Date.now() + 500;
    while (Date.now() < triggerFireDeadline) {
      const ids = parseTriggerFireRunIds(supervisorToChild.flushed());
      if (ids.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(
      parseTriggerFireRunIds(supervisorToChild.flushed()).length,
    ).toBeGreaterThanOrEqual(1);

    await supervisor.drain({ deadlineMs: 7_500 });

    // The supervisor's `drain` control frame landed on the
    // supervisor-to-child stream alongside the buffered-mail
    // `trigger.fire` frames. The FIFO claim-check pipeline keeps the
    // dispatch loop running concurrently with `drain()`, so a fresh
    // `trigger.fire` can land before or after the drain frame; find
    // the drain frame by payload type rather than indexing the tail.
    const forwarded = supervisorToChild.flushed();
    expect(forwarded.length).toBeGreaterThanOrEqual(2);
    const SignedFrame = type({
      envelope: {
        seq: "number",
        channelId: "string",
        payload: {
          type: "string",
          "+": "ignore",
        },
        "+": "ignore",
      },
      "+": "ignore",
    });
    const drainFrame = (() => {
      for (const line of forwarded) {
        const parsed = SignedFrame(JSON.parse(line));
        if (parsed instanceof type.errors) continue;
        if (parsed.envelope.payload.type === "drain") return parsed;
      }
      throw new Error("no drain frame observed on supervisor-to-child stream");
    })();
    expect(drainFrame.envelope.payload).toMatchObject({
      type: "drain",
      data: { deadlineMs: 7_500 },
    });

    // The FIFO inbox queue serializes dispatch: one run is in-flight
    // at a time. The drain() call observes exactly one in-flight run
    // (whichever message was mid-dispatch when the loop awaited
    // `waitForRunTerminal`) and arms one accumulator for it. The
    // second message stays in the inbox until the first completes.
    expect(stubs.length).toBeGreaterThanOrEqual(1);
    for (const stub of stubs) {
      expect(stub.__startCount).toBe(1);
      expect(stub.__stopCount).toBe(0);
      expect(stub.__opts.deploymentId).toBe("deployment-x");
      expect(stub.__opts.repoId).toEqual({
        kind: "workflow-run",
        id: "deployment-x",
      });
      expect(stub.__opts.ref).toBe("refs/heads/main");
      expect(stub.__opts.drainTimeoutMs).toBe(7_500);
      expect(typeof stub.__opts.runId).toBe("string");
      expect(stub.__opts.runId.length).toBeGreaterThan(0);
    }
    const runIds = stubs.map((s) => s.__opts.runId);
    expect(new Set(runIds).size).toBe(stubs.length);

    // Shutdown stops every armed accumulator before tearing the
    // child down.
    await supervisor.shutdown();
    for (const stub of stubs) {
      expect(stub.__stopCount).toBe(1);
    }
  });

  test("drain() escalates via signAsPrincipal when the accumulator's timeout fires", async () => {
    // Production-shaped wiring: bind the real
    // `createDrainTimeoutAccumulator` and observe the
    // `CancelRequested{origin: "supervisor-drain"}` commit landing on
    // the stub RepoStore's write side after the supervisor's fake
    // clock advances past the configured `drainTimeoutMs`. This is
    // the supervisor-equivalent of the in-process round-trip the
    // 13c integration test exercises.
    const baseDir = await makeTempDir("supervisor-drain-escalate-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 8888,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };

    type FakeTimer = { cb: () => void; ms: number; cancelled: boolean };
    const timers = new Set<FakeTimer>();
    let fakeNow = 1_700_000_000_000;
    const observedWrites: {
      principal: { kind: string };
      repoId: RepoId;
      ref: string;
      files: Record<string, string | Uint8Array>;
    }[] = [];

    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus,
      onWrite: (args) => observedWrites.push(args),
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      drainTimeoutMs: 1_000,
      now: () => fakeNow,
      setTimer: (cb, ms) => {
        const t: FakeTimer = { cb, ms, cancelled: false };
        timers.add(t);
        return t;
      },
      clearTimer: (handle) => {
        if (handle === null || typeof handle !== "object") return;
        for (const t of timers) {
          if (t === handle) {
            t.cancelled = true;
            timers.delete(t);
            return;
          }
        }
      },
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: () => {
        /* unused in this test */
      },
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("escalate-msg"),
    );
    await childSender.send({
      type: "ready",
      data: {
        childPid: 8888,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;

    // Wait for the dispatch loop to forward the buffered mail's
    // `trigger.fire` so the run is in `cohortRunIds` when `drain()`
    // arms its accumulator. With the H-S1 replayDone gate the first
    // dispatch is no longer synchronous with `await spawnPromise`.
    const triggerFireDeadline = Date.now() + 500;
    while (Date.now() < triggerFireDeadline) {
      const ids = parseTriggerFireRunIds(supervisorToChild.flushed());
      if (ids.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(
      parseTriggerFireRunIds(supervisorToChild.flushed()).length,
    ).toBeGreaterThanOrEqual(1);

    await supervisor.drain({ deadlineMs: 1_000 });
    expect(timers.size).toBe(1);
    // Advance the fake clock past the timeout and fire the
    // accumulator's pending timer.
    fakeNow += 1_000;
    const due = [...timers];
    for (const t of due) {
      if (t.cancelled) continue;
      timers.delete(t);
      t.cb();
    }
    // Allow the async escalate commit to settle.
    await new Promise<void>((r) => setTimeout(r, 5));

    // The accumulator's escalation committed a CancelRequested event
    // through the supervisor's substrate handle. Filter to the write that
    // carries the event rather than coupling this assertion to other
    // substrate maintenance writes.
    const writesWithEvents = observedWrites.filter((w) =>
      Object.keys(w.files).some((k) => k.includes("/events/")),
    );
    expect(writesWithEvents.length).toBe(1);
    const write = writesWithEvents[0];
    if (write === undefined) {
      throw new Error("no CancelRequested commit captured");
    }
    expect(write.principal.kind).toBe("supervisor");
    expect(write.repoId).toEqual({
      kind: "workflow-run",
      id: "deployment-x",
    });
    const eventEntry = Object.entries(write.files).find(([k]) =>
      k.includes("/events/"),
    );
    if (eventEntry === undefined) {
      throw new Error("no event blob captured in the commit");
    }
    const [, blobBytes] = eventEntry;
    const blobJson =
      typeof blobBytes === "string"
        ? blobBytes
        : new TextDecoder().decode(blobBytes);
    const blob = readCancelRequestedBlob(blobJson);
    expect(blob.type).toBe("CancelRequested");
    expect(blob.origin).toBe("supervisor-drain");
    expect(blob.signature.principalKind).toBe("supervisor");

    await supervisor.shutdown();
  });

  test("requestCancel signs CancelRequested via signAsPrincipal for every origin", async () => {
    const baseDir = await makeTempDir("supervisor-cancel-");
    const signSpyCalls: { kind: string; payload: Uint8Array }[] = [];
    const observedWrites: {
      principal: { kind: string };
      repoId: RepoId;
      ref: string;
      files: Record<string, string | Uint8Array>;
    }[] = [];
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error("spawn not invoked in cancel test");
      },
      signSpy: (kind, payload) => {
        signSpyCalls.push({ kind, payload });
        // Synthetic 64-byte signature with the run id encoded in the
        // first bytes so the test asserts which call produced it.
        const sig = new Uint8Array(64);
        sig[0] = signSpyCalls.length;
        return { sig, principalKind: "supervisor" };
      },
      mailBus: createMockMailBus(),
      onWrite: (args) => observedWrites.push(args),
    });
    const supervisor = createWorkflowSupervisor(bindings);

    const origins = [
      "self",
      "supervisor-drain",
      "supervisor-operator",
      "hub-admin",
    ] as const;
    for (const origin of origins) {
      const result = await supervisor.requestCancel({
        runId: `run-${origin}`,
        origin,
        reason: `reason for ${origin}`,
        at: "2026-01-01T00:00:00.000Z",
      });
      expect(result.commitSha).toBe("deadbeefcafef00d");
    }

    // Every origin flows through the supervisor's signing callback
    // with principal kind `"supervisor"`. The kind-handler-side
    // principal-vs-origin map for hub-admin is enforced when the
    // push is presented; the supervisor's signing path itself does
    // not vary by origin.
    expect(signSpyCalls.length).toBe(origins.length);
    for (const call of signSpyCalls) {
      expect(call.kind).toBe("supervisor");
      expect(call.payload).toBeInstanceOf(Uint8Array);
      const text = new TextDecoder().decode(call.payload);
      expect(text).toContain("CancelRequested");
    }

    expect(observedWrites.length).toBe(origins.length);
    for (const write of observedWrites) {
      expect(write.principal.kind).toBe("supervisor");
      expect(write.repoId).toEqual({
        kind: "workflow-run",
        id: "deployment-x",
      });
    }
    const firstWrite = observedWrites[0];
    if (firstWrite === undefined) {
      throw new Error("no observed writes captured");
    }
    const firstEntry = Object.entries(firstWrite.files)[0];
    if (firstEntry === undefined) {
      throw new Error("first write produced no files");
    }
    const [, firstBytes] = firstEntry;
    const firstJson =
      typeof firstBytes === "string"
        ? firstBytes
        : new TextDecoder().decode(firstBytes);
    const onDisk = readCancelRequestedBlob(firstJson);
    expect(onDisk.type).toBe("CancelRequested");
    expect(onDisk.origin).toBe("self");
    expect(onDisk.signature.principalKind).toBe("supervisor");
    expect(onDisk.signature.sig).toMatch(/^01[0-9a-f]+$/);
  });

  test("drain() threads the per-cohort terminal broadcaster into each accumulator's opts", async () => {
    const baseDir = await makeTempDir("supervisor-drain-terminal-source-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      return {
        pid: 7777,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
    };

    type StubAccumulator = DrainTimeoutAccumulator & {
      __opts: DrainTimeoutOpts;
    };
    const stubs: StubAccumulator[] = [];
    const factory: DrainTimeoutAccumulatorFactory = (opts) => {
      const stub: StubAccumulator = {
        __opts: opts,
        start() {
          /* unused */
        },
        pause() {
          /* unused */
        },
        resume() {
          /* unused */
        },
        stop() {
          /* unused */
        },
        accumulatedMs() {
          return 0;
        },
        get escalated() {
          return false;
        },
        disposed() {
          return Promise.resolve();
        },
      };
      stubs.push(stub);
      return stub;
    };

    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      drainTimeoutAccumulatorFactory: factory,
      drainTimeoutMs: 5_000,
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("term-msg-A"),
    );
    await childSender.send({
      type: "ready",
      data: {
        childPid: 7777,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;
    // Wait for the dispatch loop to forward the buffered mail's
    // `trigger.fire` so the run is in `cohortRunIds` when `drain()`
    // arms its accumulator. The H-S1 replayDone gate moves the first
    // dispatch off the `await spawnPromise` critical path.
    const triggerFireDeadline = Date.now() + 500;
    while (Date.now() < triggerFireDeadline) {
      const ids = parseTriggerFireRunIds(supervisorToChild.flushed());
      if (ids.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(
      parseTriggerFireRunIds(supervisorToChild.flushed()).length,
    ).toBeGreaterThanOrEqual(1);
    await supervisor.drain({ deadlineMs: 5_000 });

    // The supervisor's per-cohort terminal broadcaster always backs
    // the accumulator's terminal-event source; the accumulator factory
    // sees a non-undefined slot and can mint a per-runId iterator
    // through it.
    expect(stubs).toHaveLength(1);
    const stub = stubs[0];
    if (stub === undefined) throw new Error("expected one stub accumulator");
    expect(stub.__opts.terminalEventSource).toBeDefined();
    const factorySource = stub.__opts.terminalEventSource;
    if (factorySource === undefined) {
      throw new Error(
        "expected accumulator opts to carry a terminalEventSource",
      );
    }
    const iterable = factorySource(stub.__opts.runId);
    const iter = iterable[Symbol.asyncIterator]();
    await iter.return?.(undefined);

    await supervisor.shutdown();
  });

  test("drain() arms the broadcaster-backed accumulator source on the active cohort", async () => {
    const baseDir = await makeTempDir("supervisor-drain-no-term-source-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      return {
        pid: 6666,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
    };

    type StubAccumulator = DrainTimeoutAccumulator & {
      __opts: DrainTimeoutOpts;
    };
    const stubs: StubAccumulator[] = [];
    const factory: DrainTimeoutAccumulatorFactory = (opts) => {
      const stub: StubAccumulator = {
        __opts: opts,
        start() {
          /* unused */
        },
        pause() {
          /* unused */
        },
        resume() {
          /* unused */
        },
        stop() {
          /* unused */
        },
        accumulatedMs() {
          return 0;
        },
        get escalated() {
          return false;
        },
        disposed() {
          return Promise.resolve();
        },
      };
      stubs.push(stub);
      return stub;
    };

    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      drainTimeoutAccumulatorFactory: factory,
      drainTimeoutMs: 5_000,
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("no-term-msg"),
    );
    await childSender.send({
      type: "ready",
      data: {
        childPid: 6666,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;
    // Wait for the dispatch loop to forward the buffered mail's
    // `trigger.fire` so the run is in `cohortRunIds` when `drain()`
    // arms its accumulator. The H-S1 replayDone gate moves the first
    // dispatch off the `await spawnPromise` critical path.
    const triggerFireDeadline = Date.now() + 500;
    while (Date.now() < triggerFireDeadline) {
      const ids = parseTriggerFireRunIds(supervisorToChild.flushed());
      if (ids.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(
      parseTriggerFireRunIds(supervisorToChild.flushed()).length,
    ).toBeGreaterThanOrEqual(1);
    await supervisor.drain({ deadlineMs: 5_000 });

    expect(stubs).toHaveLength(1);
    const stub = stubs[0];
    if (stub === undefined) throw new Error("expected one stub accumulator");
    // The supervisor owns the per-cohort terminal broadcaster; the
    // accumulator factory always receives a non-undefined terminal
    // source backed by the active cohort's broadcaster. There is no
    // path through the supervisor today that leaves the accumulator
    // on timer-only settlement -- the broadcaster supplants the
    // pre-binding behaviour wholesale.
    expect(stub.__opts.terminalEventSource).toBeDefined();

    await supervisor.shutdown();
  });

  test("drain() is a no-op when the supervisor is idle (no spawn has run)", async () => {
    // Pins the defensive contract for an inbound drain.deliver frame
    // that lands while the supervisor has no in-flight runs to escalate
    // (e.g. the deployment's only run already reached a terminal
    // state). `drain` returns silently in `idle`/`stopping`/`stopped`,
    // does not throw, does not forward a `drain` control frame to a
    // dead child, and does not arm any accumulators. This is the
    // contract higher-level host shutdown sequences depend on -- they
    // call `drain` unconditionally without sniffing the phase.
    const baseDir = await makeTempDir("supervisor-drain-idle-");
    const accumulatorInvocations: DrainTimeoutOpts[] = [];
    const accumulatorFactory: DrainTimeoutAccumulatorFactory = (opts) => {
      accumulatorInvocations.push(opts);
      const stub: DrainTimeoutAccumulator = {
        start() {
          /* unused */
        },
        pause() {
          /* unused */
        },
        resume() {
          /* unused */
        },
        stop() {
          /* unused */
        },
        accumulatedMs() {
          return 0;
        },
        get escalated() {
          return false;
        },
        disposed() {
          return Promise.resolve();
        },
      };
      return stub;
    };
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error("spawner must not be invoked on the idle drain path");
      },
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus: createMockMailBus(),
    });
    const supervisor = createWorkflowSupervisor({
      ...bindings,
      drainTimeoutAccumulatorFactory: accumulatorFactory,
    });
    await supervisor.drain({ deadlineMs: 5_000 });
    expect(accumulatorInvocations).toHaveLength(0);
  });

  test("deliverSignal() rejects when the supervisor is idle (no spawn has run)", async () => {
    // Pins the defensive contract for an inbound signal.deliver frame
    // landing against a supervisor that is not in `starting`/`running`/
    // `recycling`. The supervisor throws so the router's
    // `tryRoute` rejection propagates up to the hub-link's
    // `handleSignalDeliver`, which logs and drops without crashing the
    // sidecar or contaminating sibling deployments.
    const baseDir = await makeTempDir("supervisor-deliver-signal-idle-");
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error("spawner must not be invoked on the idle signal path");
      },
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus: createMockMailBus(),
    });
    const supervisor = createWorkflowSupervisor(bindings);
    await expect(
      supervisor.deliverSignal({
        runId: "run-stale",
        signalName: "approve",
        signalId: "sig-stale",
        payload: null,
      }),
    ).rejects.toThrow(/deliverSignal called in phase idle/);
  });

  test("deliverSources() rejects when the supervisor is idle (no spawn has run)", async () => {
    // Same phase-guard contract as deliverSignal: a sources rotation
    // landing against a supervisor that is not starting/running throws so
    // the sidecar router's rejection surfaces to the hub-link rather than
    // writing into a dead child's pipe.
    const baseDir = await makeTempDir("supervisor-deliver-sources-idle-");
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error("spawner must not be invoked on the idle sources path");
      },
      signSpy: () => ({
        sig: new Uint8Array(64),
        principalKind: "supervisor",
      }),
      mailBus: createMockMailBus(),
    });
    const supervisor = createWorkflowSupervisor(bindings);
    await expect(
      supervisor.deliverSources({
        sources: [
          {
            id: "primary",
            provider: "anthropic",
            baseURL: "https://api.anthropic.com",
            apiKey: "sk-x",
            model: "claude-test",
          },
        ],
        defaultSource: "primary",
      }),
    ).rejects.toThrow(/deliverSources called in phase idle/);
  });

  test("deliverSources() sends a sources-updated frame when running", async () => {
    const baseDir = await makeTempDir("supervisor-deliver-sources-running-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });

    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 4321,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };

    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus: createMockMailBus(),
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: true,

      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    await childSender.send({
      type: "ready",
      data: {
        childPid: 4321,
        childPublicKey: Buffer.from(childIpcKeyPair.publicKey).toString("hex"),
      },
    });
    await spawnPromise;

    const sources: InferenceSource[] = [
      {
        id: "primary",
        provider: "anthropic",
        baseURL: "https://api.anthropic.com",
        apiKey: "sk-primary",
        model: "claude-test",
      },
    ];
    await supervisor.deliverSources({ sources, defaultSource: "primary" });

    const frames = parseSourcesUpdatedFrames(supervisorToChild.flushed());
    expect(frames).toHaveLength(1);
    expect(frames[0]?.sources).toEqual(sources);
    expect(frames[0]?.defaultSource).toBe("primary");

    await supervisor.shutdown();
  });

  test("deliverCredentials() rejects when the supervisor is idle (no spawn has run)", async () => {
    // Same phase-guard contract as deliverSources: a credential push landing
    // against a supervisor that is not starting/running throws so the caller
    // surfaces the race rather than writing into a dead child's pipe.
    const baseDir = await makeTempDir("supervisor-deliver-credentials-idle-");
    const bindings = await buildBindings({
      baseDir,
      spawner: () => {
        throw new Error(
          "spawner must not be invoked on the idle credentials path",
        );
      },
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus: createMockMailBus(),
    });
    const supervisor = createWorkflowSupervisor(bindings);
    await expect(
      supervisor.deliverCredentials({
        delivery: { bindings: [], materials: [] },
      }),
    ).rejects.toThrow(/deliverCredentials called in phase idle/);
  });

  test("deliverCredentials() sends a credentials-updated frame when running", async () => {
    const baseDir = await makeTempDir(
      "supervisor-deliver-credentials-running-",
    );
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });

    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 4321,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };

    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus: createMockMailBus(),
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
    };
    const supervisor = createWorkflowSupervisor(bindings);

    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: true,
      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    await childSender.send({
      type: "ready",
      data: {
        childPid: 4321,
        childPublicKey: Buffer.from(childIpcKeyPair.publicKey).toString("hex"),
      },
    });
    await spawnPromise;

    const delivery = {
      bindings: [
        {
          handle: "gh",
          credentialId: "cred_a",
          consumer: "tool:@intx/tools-example",
        },
      ],
      materials: [
        {
          credentialId: "cred_a",
          providerKey: "http",
          origin: "https://api.example.test",
          secret: "sk-real",
        },
      ],
    };
    await supervisor.deliverCredentials({ delivery });

    const frames = parseCredentialsUpdatedFrames(supervisorToChild.flushed());
    expect(frames).toHaveLength(1);
    expect(frames[0]).toEqual(delivery);

    await supervisor.shutdown();
  });
  // ------------------------------------------------------------------
  // Long-lived dispatch path
  // ------------------------------------------------------------------

  test("long-lived: first message fires trigger.fire with stable runId", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-first-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // In the unified-dispatch path markConsumed waits for the child to
    // reach terminal or park before consuming the message.  Drive the
    // mock child to terminal so the dispatch loop can proceed.
    await new Promise((r) => setTimeout(r, 100));
    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: new Date().toISOString(),
      },
    });

    const address = "deployment-x@example.com";
    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    const runIds = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
    expect(runIds).toEqual(["deployment-x@example.com"]);
    await wired.supervisor.shutdown();
  });

  test("long-lived: a trigger.fire run parking on approval releases the dispatch wait", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-approval-park-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // Let the dispatch loop forward trigger.fire and enter
    // waitForRunTerminalOrPark before the run parks.
    await new Promise((r) => setTimeout(r, 10));

    // The run's first step parks on an APPROVAL gate -- not an input park and
    // not a terminal. An approval park must release the dispatch wait the same
    // as an input park does; without that, this hangs to the terminal-or-park
    // backstop and the mail is never consumed.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: "deployment-x@example.com",
        correlationId: "corr-approval-1",
        parkKind: "approval",
      },
    });

    const address = "deployment-x@example.com";
    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    // markConsumed only runs once the dispatch wait returns; a wait still
    // hanging on the approval park would leave this at 0 until the backstop.
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    const runIds = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
    expect(runIds).toEqual(["deployment-x@example.com"]);
    await wired.supervisor.shutdown();
  });

  test("long-lived: subsequent messages fire signal.deliver after park.notify", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-signal-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // Wait for trigger.fire to land before parking, then send park.notify
    // so the unified-dispatch path can complete markConsumed.
    await new Promise((r) => setTimeout(r, 10));

    const address = "deployment-x@example.com";

    // Child parks on input signal.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: "deployment-x@example.com",
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    let deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-2"),
    );

    // msg-2 lands on the parked run as signal.deliver. Wait for it to be
    // sent (which arms the durable-consume watcher), then complete the
    // resumed run: markConsumed for a signal now holds until the child has
    // durably taken it up (re-parks or terminates), mirroring trigger.fire.
    deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (parseSignalDelivers(wired.supervisorToChild.flushed()).length >= 1) {
        break;
      }
      await new Promise((r) => setTimeout(r, 1));
    }
    // The signal is sent but the run has not taken it up: markConsumed must
    // still be held, so only msg-1 is consumed. A brief settle window would
    // let a premature consume land if the contract regressed.
    await new Promise((r) => setTimeout(r, 25));
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });

    deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 2) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(2);

    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBeGreaterThanOrEqual(1);
    const firstSignal = signals[0];
    if (firstSignal === undefined) throw new Error("unreachable");
    expect(firstSignal.signalName).toBe(signalName("corr-input-1"));
    expect(firstSignal.signalId).toBeTruthy();
    await wired.supervisor.shutdown();
  });

  test("long-lived: messages before park.notify are queued and flushed", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-queue-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // In the unified-dispatch path markConsumed waits for the child to
    // park or reach terminal.  Nothing is consumed yet.
    await new Promise((r) => setTimeout(r, 50));

    const address = "deployment-x@example.com";
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(0);

    // Deliver second message BEFORE child parks.
    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-2"),
    );

    // Wait a short beat so the dispatch loop has a chance to process it.
    await new Promise((r) => setTimeout(r, 50));

    // Still no signal.deliver and still nothing consumed because the
    // channel is unknown and the first run has not parked.
    let signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(0);
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(0);

    // Now child parks.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: "deployment-x@example.com",
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    // Wait for the queued message to be flushed.
    let deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      signals = parseSignalDelivers(wired.supervisorToChild.flushed());
      if (signals.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(signals.length).toBeGreaterThanOrEqual(1);
    const firstSignal = signals[0];
    if (firstSignal === undefined) throw new Error("unreachable");
    expect(firstSignal.signalName).toBe(signalName("corr-input-1"));

    // msg-1 is consumed off its park, but msg-2's signal is not yet taken up
    // by the run, so its markConsumed is still held.
    await new Promise((r) => setTimeout(r, 25));
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    // The signal.deliver above is observed, so its durable-consume watcher
    // is armed; complete the resumed run so markConsumed for msg-2 releases.
    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });

    // Both messages are consumed.
    deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 2) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(2);
    await wired.supervisor.shutdown();
  });

  test("long-lived: drain() does not arm accumulators", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-drain-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const armedStubs: { runId: string }[] = [];
    const factory: DrainTimeoutAccumulatorFactory = (opts) => {
      armedStubs.push({ runId: opts.runId });
      return createNoopDrainAccumulator();
    };

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
      drainTimeoutAccumulatorFactory: factory,
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // Park the run so markConsumed can proceed and the run enters the
    // runtime-determined parked state that drain should skip.
    await new Promise((r) => setTimeout(r, 10));
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: "deployment-x@example.com",
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    const address = "deployment-x@example.com";
    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    await wired.supervisor.drain({ deadlineMs: 5_000 });

    // Drain skips accumulators for runs that have parked (runtime-
    // determined long-lived state).
    expect(armedStubs.length).toBe(0);
    await wired.supervisor.shutdown();
  });

  test("long-lived: grants barrier failure consumes message without firing trigger", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-barrier-");
    const onRunStart: WorkflowSupervisorBindings["onRunStart"] = async () => {
      throw new Error("synthetic grants-barrier failure");
    };

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart,
    });

    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // In the unified-dispatch path markConsumed waits for the child to
    // reach terminal or park before consuming the message.  Drive the
    // mock child to terminal so the dispatch loop can proceed.
    await new Promise((r) => setTimeout(r, 100));
    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: new Date().toISOString(),
      },
    });

    const address = "deployment-x@example.com";
    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    const runIds = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
    expect(runIds).toEqual([]);
    await wired.supervisor.shutdown();
  });

  test("long-lived: mail after terminal is rejected without another trigger.fire", async () => {
    const baseDir = await makeTempDir("supervisor-long-lived-terminal-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,

      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    // First message triggers the run.
    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    // Drive the run to terminal so markConsumed can proceed.
    await new Promise((r) => setTimeout(r, 10));
    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });

    const address = "deployment-x@example.com";
    let deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    // Second message arrives after the deployment's one top-level run
    // terminated. It must be durably rejected, not treated as a new run.
    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-2"),
    );

    deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 2) break;
      await new Promise((r) => setTimeout(r, 1));
    }

    const runIds = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
    expect(runIds).toEqual(["deployment-x@example.com"]);
    expect(
      [...wired.inboxPrimitives.snapshot(address).consumed.values()].some(
        (entry) => entry.rejection?.code === "workflow_run_terminal",
      ),
    ).toBe(true);

    // No signal.deliver because the run never parked.
    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(0);

    await wired.supervisor.shutdown();
  });

  test("a recovery-window mail is rejected when the live run terminates before parking", async () => {
    const baseDir = await makeTempDir("supervisor-recovery-terminal-race-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () =>
        assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        }),
    });
    const address = "deployment-x@example.com";

    // Model restart recovery: the child owns the durable live run, but has not
    // yet re-emitted an input park/correlation for it.
    await wired.childSender.send({
      type: "resumed.runs",
      data: { runIds: [address] },
    });
    wired.mailBus.deliver(address, new TextEncoder().encode("waiting mail"));
    await new Promise((resolve) => setTimeout(resolve, 25));
    expect(parseTriggerFireRunIds(wired.supervisorToChild.flushed())).toEqual(
      [],
    );
    expect(parseSignalDelivers(wired.supervisorToChild.flushed())).toEqual([]);

    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: address,
        seq: 1,
        kind: "RunCompleted",
        at: "test",
      },
    });

    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((resolve) => setTimeout(resolve, 1));
    }
    expect(parseTriggerFireRunIds(wired.supervisorToChild.flushed())).toEqual(
      [],
    );
    expect(
      [...wired.inboxPrimitives.snapshot(address).consumed.values()][0]
        ?.rejection?.code,
    ).toBe("workflow_run_terminal");

    await wired.supervisor.shutdown();
  });

  test("a clean deployment with grants but no events fires its top-level run", async () => {
    const baseDir = await makeTempDir("supervisor-clean-run-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    // `grants.json` may be staged before delivery, but without an event log
    // this is still the deployment's one allowed first fire.
    wired.mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("msg-1"),
    );

    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      const ids = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
      if (ids.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }

    const runIds = parseTriggerFireRunIds(wired.supervisorToChild.flushed());
    expect(runIds.length).toBe(1);
    expect(runIds[0]).toBe("deployment-x@example.com");
    await wired.supervisor.shutdown();
  });

  test("a terminal durable log is never cleared or fired after supervisor restart", async () => {
    const baseDir = await makeTempDir("supervisor-terminal-restart-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const runEventsDir = path.join(
      baseDir,
      "workflow-run",
      "deployment-x",
      "runs",
      "deployment-x@example.com",
      "events",
    );
    await fs.mkdir(runEventsDir, { recursive: true });
    await fs.writeFile(
      path.join(runEventsDir, "0.json"),
      JSON.stringify({ type: "RunStarted", seq: 0 }),
    );
    await fs.writeFile(
      path.join(runEventsDir, "1.json"),
      JSON.stringify({ type: "RunCompleted", seq: 1 }),
    );

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    const address = "deployment-x@example.com";
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));

    const deadline = Date.now() + 1000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(parseTriggerFireRunIds(wired.supervisorToChild.flushed())).toEqual(
      [],
    );
    expect(
      [...wired.inboxPrimitives.snapshot(address).consumed.values()].some(
        (entry) => entry.rejection?.code === "workflow_run_terminal",
      ),
    ).toBe(true);

    await wired.supervisor.shutdown();
  });

  test("an instant park during the pre-wait window still advances the dispatch loop", async () => {
    const baseDir = await makeTempDir("supervisor-instant-park-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    const address = "deployment-x@example.com";
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));

    // Park immediately; the generation check must handle either side of the
    // dispatch loop arming its waiter.
    await new Promise((r) => setTimeout(r, 20));
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    // The loop must proceed and consume msg-1 despite the lost park wake.
    const deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 2));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    await wired.supervisor.shutdown();
  });

  test("two mails to a parked run each deliver on the run's fresh correlation", async () => {
    const baseDir = await makeTempDir("supervisor-fresh-corr-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";
    const waitConsumed = async (n: number) => {
      const deadline = Date.now() + 2000;
      while (Date.now() < deadline) {
        if (wired.inboxPrimitives.snapshot(address).consumed.size >= n) break;
        await new Promise((r) => setTimeout(r, 2));
      }
    };
    const waitSignals = async (n: number) => {
      const deadline = Date.now() + 2000;
      while (Date.now() < deadline) {
        const s = parseSignalDelivers(wired.supervisorToChild.flushed());
        if (s.length >= n) break;
        await new Promise((r) => setTimeout(r, 2));
      }
    };

    // Trigger msg-1 and park the run on corr-1.
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));
    await new Promise((r) => setTimeout(r, 10));
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });
    await waitConsumed(1);
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    // Two mails arrive at the parked run back-to-back. Both must deliver as
    // signals, each on the correlation the run is CURRENTLY parked on: msg-2
    // on corr-1, then -- after the run re-parks on corr-2 -- msg-3 on corr-2,
    // never the stale corr-1 the cache would hold without invalidation.
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-2"));
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-3"));

    await waitSignals(1);
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-2",
        parkKind: "input",
      },
    });
    await waitConsumed(2);

    await waitSignals(2);
    await wired.childSender.send({
      type: "terminal.event",
      data: {
        runId: address,
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });
    await waitConsumed(3);

    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(3);
    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(2);
    expect(signals[0]?.signalName).toBe(signalName("corr-input-1"));
    expect(signals[1]?.signalName).toBe(signalName("corr-input-2"));

    await wired.supervisor.shutdown();
  });

  test("a park-registered run keeps its channel: a mail routes as signal, not a fresh trigger", async () => {
    const baseDir = await makeTempDir("supervisor-resumed-order-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );

    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";

    // Register the run's input channel via park.notify with NO prior local
    // trigger.fire -- the shape the reconnect/resumed path produces, where
    // cohort membership and the channel both come from park discovery, not a
    // fire. The handler must add cohortRunIds BEFORE the channel; otherwise the
    // dispatch loop's routing hygiene sees a channel-without-cohort entry,
    // deletes it as stale, and the next mail wrongly starts a FRESH run.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));

    const deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (parseSignalDelivers(wired.supervisorToChild.flushed()).length >= 1) {
        break;
      }
      await new Promise((r) => setTimeout(r, 2));
    }
    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(1);
    expect(signals[0]?.signalName).toBe(signalName("corr-input-1"));
    expect(
      parseTriggerFireRunIds(wired.supervisorToChild.flushed()).length,
    ).toBe(0);

    // Release the durable-consume wait so shutdown is clean.
    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });
    await wired.supervisor.shutdown();
  });

  test("a mail resumes a parked run with the conversation text, not raw MIME", async () => {
    const baseDir = await makeTempDir("supervisor-signal-text-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";

    // Park the run so the next mail routes as signal.deliver.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    // Deliver a real inbound MIME message whose body is "hello turn two".
    const mail = new TextEncoder().encode(
      "Content-Type: text/plain\r\n\r\nhello turn two",
    );
    wired.mailBus.deliver(address, mail);

    const deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (parseSignalDelivers(wired.supervisorToChild.flushed()).length >= 1) {
        break;
      }
      await new Promise((r) => setTimeout(r, 2));
    }
    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(1);
    // The frame carries the EXTRACTED conversation text, resolved at the
    // dispatch site, not the raw base64 MIME envelope.
    expect(signals[0]?.payload).toBe("hello turn two");

    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });
    await wired.supervisor.shutdown();
  });

  test("deliverSignal ships its structured payload through unchanged", async () => {
    const baseDir = await makeTempDir("supervisor-deliversignal-passthrough-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    // A hub-originated signal (e.g. awaitSignal) carries a STRUCTURED payload
    // that must reach the child verbatim -- the mail-input extraction is the
    // dispatch loop's concern only, and this contract split is what the
    // signal.deliver frame's uniform "final-form payload" contract guarantees.
    await wired.supervisor.deliverSignal({
      runId: "deployment-x@example.com",
      signalName: "go",
      signalId: "sig-1",
      payload: { resumed: true, n: 7 },
    });

    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(1);
    expect(signals[0]?.payload).toEqual({ resumed: true, n: 7 });

    await wired.supervisor.shutdown();
  });

  test("a malformed turn-2 mail is dropped and consumed, not poison-looped", async () => {
    const baseDir = await makeTempDir("supervisor-poison-mail-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";

    // Park the run so a mail routes as signal.deliver.
    await wired.childSender.send({
      type: "park.notify",
      data: {
        runId: address,
        correlationId: "corr-input-1",
        parkKind: "input",
      },
    });

    // A malformed multipart mail whose part 1 cannot be parsed: extraction
    // throws. It must be dropped and CONSUMED, not thrown-and-replayed forever.
    const bad = new TextEncoder().encode(
      "Content-Type: multipart/mixed; boundary=zzz\r\n\r\nno parts here",
    );
    wired.mailBus.deliver(address, bad);

    let deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 2));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);
    expect(wired.inboxPrimitives.snapshot(address).processing.size).toBe(0);
    // No signal was delivered for the poison mail.
    expect(parseSignalDelivers(wired.supervisorToChild.flushed()).length).toBe(
      0,
    );

    // The run survived on its correlation: a subsequent VALID mail resumes it.
    const good = new TextEncoder().encode(
      "Content-Type: text/plain\r\n\r\nhello",
    );
    wired.mailBus.deliver(address, good);
    deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (parseSignalDelivers(wired.supervisorToChild.flushed()).length >= 1) {
        break;
      }
      await new Promise((r) => setTimeout(r, 2));
    }
    const signals = parseSignalDelivers(wired.supervisorToChild.flushed());
    expect(signals.length).toBe(1);
    expect(signals[0]?.signalName).toBe(signalName("corr-input-1"));
    expect(signals[0]?.payload).toBe("hello");

    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });
    await wired.supervisor.shutdown();
  });

  test("a markConsumed failure leaves the mail reclaimable and the loop alive", async () => {
    const baseDir = await makeTempDir("supervisor-markconsumed-fatal-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const memoryInbox = createMemoryInboxPrimitives();
    let failMarkConsumed = true;
    const failingInbox: MemoryInboxPrimitives = {
      ...memoryInbox,
      markConsumed: async (...args) => {
        if (failMarkConsumed) {
          throw new Error("injected markConsumed failure");
        }
        return memoryInbox.markConsumed(...args);
      },
    };
    const wired = await spawnWithRunStart({
      baseDir,
      inboxPrimitives: failingInbox,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";

    // Drive a run to terminal so dispatch reaches markConsumed, which throws.
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));
    await new Promise((r) => setTimeout(r, 10));
    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });

    // The failure propagates into the dispatch fault handler rather than being
    // swallowed: the mail is NOT recorded consumed -- it stays in processing/,
    // reclaimable -- and the dispatch loop survives the throw.
    await new Promise((r) => setTimeout(r, 100));
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(0);
    expect(wired.inboxPrimitives.snapshot(address).processing.size).toBe(1);

    // Loop is alive: a second mail, once markConsumed recovers, is consumed.
    failMarkConsumed = false;
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-2"));
    await new Promise((r) => setTimeout(r, 10));
    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 1, kind: "RunCompleted", at: "test" },
    });
    const deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 2));
    }
    expect(
      wired.inboxPrimitives.snapshot(address).consumed.size,
    ).toBeGreaterThanOrEqual(1);

    await wired.supervisor.shutdown();
  });

  test("a mail enqueued during a dispatch iteration is picked up, not stranded", async () => {
    const baseDir = await makeTempDir("supervisor-lost-wake-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const address = "deployment-x@example.com";
    const mailBus = createMockMailBus();
    const memoryInbox = createMemoryInboxPrimitives();
    let armed = true;
    const racingInbox: MemoryInboxPrimitives = {
      ...memoryInbox,
      dequeueToProcessing: async (...args) => {
        if (armed) {
          armed = false;
          // Model a mail landing DURING this dispatch iteration: deliver it
          // (which fires wakeDispatch and swaps the wake promise) and let that
          // settle, then report the inbox empty so dispatchOne returns false.
          // With the capture-after bug the fired wake is lost and this mail
          // sleeps forever; capture-before catches it on the next loop.
          mailBus.deliver(address, new TextEncoder().encode("msg-1"));
          await new Promise((r) => setTimeout(r, 50));
          return null;
        }
        return memoryInbox.dequeueToProcessing(...args);
      },
    };
    const wired = await spawnWithRunStart({
      baseDir,
      mailBus,
      inboxPrimitives: racingInbox,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });

    // Let the racing dequeue fire the wake and the loop re-dequeue msg-1's run,
    // then drive it to terminal so it can be consumed.
    await new Promise((r) => setTimeout(r, 120));
    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });
    const deadline = Date.now() + 2000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 2));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    await wired.supervisor.shutdown();
  });

  test("a terminal that lands during the trigger's forward window releases the wait", async () => {
    const baseDir = await makeTempDir("supervisor-subscribe-before-fire-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const wired = await spawnWithRunStart({
      baseDir,
      onRunStart: async () => {
        return assembleCredentialsSnapshot({
          repoStore: createStubRepoStore({ baseDir }),
          principal: { kind: "supervisor" },
          stepOrder: ["step-1"],
          deploymentId: "deployment-x",
          deriveStepAddress: ({ deploymentId, stepId }) =>
            `${deploymentId}-${stepId}@example.com`,
        });
      },
    });
    const address = "deployment-x@example.com";
    wired.mailBus.deliver(address, new TextEncoder().encode("msg-1"));

    const triggerDeadline = Date.now() + 1000;
    while (Date.now() < triggerDeadline) {
      if (
        parseTriggerFireRunIds(wired.supervisorToChild.flushed()).length > 0
      ) {
        break;
      }
      await new Promise((r) => setTimeout(r, 1));
    }
    // The watcher is subscribed before trigger.fire, so an immediate terminal
    // frame cannot be lost between forwarding the trigger and entering wait.
    await wired.childSender.send({
      type: "terminal.event",
      data: { runId: address, seq: 0, kind: "RunCompleted", at: "test" },
    });

    // The wait releases and the mail is consumed WITHOUT the (minutes-long)
    // backstop firing.
    const deadline = Date.now() + 3000;
    while (Date.now() < deadline) {
      if (wired.inboxPrimitives.snapshot(address).consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 2));
    }
    expect(wired.inboxPrimitives.snapshot(address).consumed.size).toBe(1);

    await wired.supervisor.shutdown();
  });
});

describe("assembleCredentialsSnapshot", () => {
  test("enumerates each step's agent-state repo and pins per-step grants by hash", async () => {
    const baseDir = await makeTempDir("supervisor-creds-");
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "d1", stepId: "alpha" }),
      [{ resource: "alpha-thing", action: "read" }],
    );
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "d1", stepId: "beta" }),
      [
        { resource: "beta-thing", action: "read" },
        { resource: "beta-thing", action: "write" },
      ],
    );
    const repoStore = createStubRepoStore({ baseDir });
    const snapshot = await assembleCredentialsSnapshot({
      repoStore,
      principal: { kind: "supervisor" },
      stepOrder: ["alpha", "beta"],
      deploymentId: "d1",
      deriveStepAddress: ({ deploymentId, stepId }) =>
        `${deploymentId}-${stepId}@example.com`,
    });
    expect(snapshot.steps).toHaveLength(2);
    expect(snapshot.steps[0]?.stepId).toBe("alpha");
    expect(snapshot.steps[0]?.address).toBe("d1-alpha@example.com");
    expect(snapshot.steps[0]?.grants).toEqual([
      { resource: "alpha-thing", action: "read" },
    ]);
    expect(snapshot.steps[0]?.contentHash).toBe(
      await hashGrants([{ resource: "alpha-thing", action: "read" }]),
    );
    expect(snapshot.steps[1]?.stepId).toBe("beta");
    expect(snapshot.steps[1]?.grants).toHaveLength(2);
    expect(snapshot.steps[0]?.contentHash).not.toBe(
      snapshot.steps[1]?.contentHash,
    );
  });

  test("treats a missing per-step grants file as an empty grant array", async () => {
    const baseDir = await makeTempDir("supervisor-creds-empty-");
    const repoStore = createStubRepoStore({ baseDir });
    const snapshot = await assembleCredentialsSnapshot({
      repoStore,
      principal: { kind: "supervisor" },
      stepOrder: ["solo"],
      deploymentId: "d2",
      deriveStepAddress: ({ deploymentId }) => `${deploymentId}@example.com`,
    });
    expect(snapshot.steps).toHaveLength(1);
    expect(snapshot.steps[0]?.grants).toEqual([]);
    expect(snapshot.steps[0]?.contentHash).toBe(await hashGrants([]));
  });

  test("a malformed grants file fails loudly rather than silently treating it as empty", async () => {
    const baseDir = await makeTempDir("supervisor-creds-bad-");
    const repoId = defaultStepRepoId({ deploymentId: "d3", stepId: "s" });
    const dir = path.join(baseDir, repoId.kind, repoId.id);
    await fs.mkdir(path.join(dir, "state"), { recursive: true });
    await fs.writeFile(path.join(dir, STEP_GRANTS_PATH), "not json");
    const repoStore = createStubRepoStore({ baseDir });
    await expect(
      assembleCredentialsSnapshot({
        repoStore,
        principal: { kind: "supervisor" },
        stepOrder: ["s"],
        deploymentId: "d3",
        deriveStepAddress: () => "d3-s@example.com",
      }),
    ).rejects.toThrow(/is not valid JSON/);
  });
});

describe("commitCancelRequested (low-level)", () => {
  test("attaches the signed payload to the on-disk CancelRequested blob", async () => {
    const baseDir = await makeTempDir("cancel-signing-");
    let observedFiles: Record<string, string | Uint8Array> | undefined;
    const repoStore = createStubRepoStore({
      baseDir,
      onWrite: ({ files }) => {
        observedFiles = files;
      },
    });
    const signed = await commitCancelRequested({
      substrate: repoStore,
      repoId: { kind: "workflow-run", id: "deploy" },
      ref: "refs/heads/main",
      deploymentId: "deploy",
      runId: "r1",
      origin: "self",
      reason: "tests pass",
      at: "2026-01-01T00:00:00.000Z",
      signAsPrincipal: async (kind, payload) => {
        expect(kind).toBe("supervisor");
        const sig = new Uint8Array(64);
        // Embed the payload length so we can verify it was signed.
        sig[0] = payload.length & 0xff;
        return { sig, principalKind: "supervisor" };
      },
    });
    expect(signed.commitSha).toBe("deadbeefcafef00d");
    expect(signed.seq).toBe(0);
    if (observedFiles === undefined) {
      throw new Error("writeTreePreservingPrefix was not invoked");
    }
    const entry = Object.entries(observedFiles).find(([k]) =>
      k.endsWith("/events/0.json"),
    );
    if (entry === undefined) {
      throw new Error("no events/0.json entry observed in commit");
    }
    const [, blobBytes] = entry;
    const blobJson =
      typeof blobBytes === "string"
        ? blobBytes
        : new TextDecoder().decode(blobBytes);
    const blob = readCancelRequestedBlob(blobJson);
    expect(blob.type).toBe("CancelRequested");
    expect(blob.origin).toBe("self");
    expect(blob.reason).toBe("tests pass");
    expect(blob.signature.principalKind).toBe("supervisor");
    expect(blob.signature.sig.length).toBe(128);
  });
});

describe("IPC integration smoke", () => {
  test("a sender/receiver round-trip on the synthetic streams used by the supervisor tests", async () => {
    // Sanity check that the in-memory stream helpers do not regress
    // the IPC contract -- the supervisor tests rely on these same
    // helpers shaped against the same primitives the production IPC
    // module exposes.
    const upstream = createMemoryNdjsonStream();
    const downstream = createMemoryNdjsonStream();
    const keyPair = await generateKeyPair();
    const channelId = generateChannelId();
    const sender = createControlChannelSender({
      privateKeySeed: keyPair.privateKey,
      channelId,
      writer: upstream.writer,
    });
    await sender.send({
      type: "ready",
      data: {
        childPid: 1,
        childPublicKey: hexEncode(keyPair.publicKey),
      },
    });
    expect(upstream.flushed()).toHaveLength(1);

    const eventStream = createMemoryFrameStream();
    const hmacKey = generateHmacKey();
    const eventSender = createEventChannelSender({
      hmacKey,
      channelId,
      writer: {
        write(bytes: Uint8Array) {
          eventStream.inject(bytes);
        },
      },
    });
    await eventSender.send({
      type: "message.run.started",
      seq: 1,
      data: {
        messageId: "m",
        messageRunId: "r",
        receivedAt: 1,
      },
    });
    eventStream.close();

    // Verify the receiver pipeline picks up the framed bytes.
    const crashes: string[] = [];
    const recvIter = receiveControlChannel({
      publicKey: keyPair.publicKey,
      channelId,
      reader: {
        read(): AsyncIterableIterator<string> {
          return upstream.reader.read();
        },
      },
      onCrash: (reason) => crashes.push(reason),
    });
    upstream.close();
    let firstPayload: { type: string } | undefined;
    for await (const payload of recvIter) {
      firstPayload = { type: payload.type };
      break;
    }
    expect(firstPayload?.type).toBe("ready");
    expect(crashes).toHaveLength(0);
    void downstream;
    void hexDecode;
  });
});

describe("supervisor inbox FIFO dispatch loop", () => {
  async function buildFifoTestFixture(opts: {
    label: string;
    inbox: InboxPrimitives;
    deriveMailAuditRef?: (
      messageId: string,
      rawMessage: Uint8Array,
    ) => { store: string; path: string };
  }) {
    const baseDir = await makeTempDir(opts.label);
    await seedStepGrants(
      baseDir,
      defaultStepRepoId({ deploymentId: "deployment-x", stepId: "step-1" }),
      [{ resource: "thing", action: "read" }],
    );
    const supervisorIpcKeyPair = await generateKeyPair();
    const childIpcKeyPair = await generateKeyPair();
    const supervisorToChild = createMemoryNdjsonStream();
    const childToSupervisor = createMemoryNdjsonStream();
    const eventChildToSupervisor = createMemoryFrameStream();
    let resolveExit: ((code: number) => void) | undefined;
    const exited = new Promise<number>((resolve) => {
      resolveExit = resolve;
    });
    let observedEnv: Record<string, string> | undefined;
    const spawner: SubprocessSpawner = ({ env }) => {
      observedEnv = env;
      const handle: SubprocessHandle = {
        pid: 11111,
        controlWriter: supervisorToChild.writer,
        controlReader: childToSupervisor.reader,
        eventReader: eventChildToSupervisor.reader,
        kill: () => {
          childToSupervisor.close();
          eventChildToSupervisor.close();
          resolveExit?.(0);
        },
        exited,
      };
      return handle;
    };
    const mailBus = createMockMailBus();
    const baseBindings = await buildBindings({
      baseDir,
      spawner,
      signSpy: () => ({ sig: new Uint8Array(64), principalKind: "supervisor" }),
      mailBus,
      inboxPrimitives: opts.inbox,
    });
    const bindings: WorkflowSupervisorBindings = {
      ...baseBindings,
      ipcKeyPairFactory: () => Promise.resolve(supervisorIpcKeyPair),
      ...(opts.deriveMailAuditRef !== undefined
        ? { deriveMailAuditRef: opts.deriveMailAuditRef }
        : {}),
    };
    const supervisor = createWorkflowSupervisor(bindings);
    const spawnPromise = supervisor.spawn({
      stepOrder: ["step-1"],
      definitionHash: "def-hash-abc",
      warmKeep: false,

      onInferenceEvent: () => undefined,
    });
    while (observedEnv === undefined) {
      await new Promise((r) => setTimeout(r, 1));
    }
    const channelId = observedEnv.IPC_CHANNEL_ID;
    if (channelId === undefined) {
      throw new Error("IPC_CHANNEL_ID not set in spawn-time env");
    }
    const childSender = createControlChannelSender({
      privateKeySeed: childIpcKeyPair.privateKey,
      channelId,
      writer: {
        write(line: string) {
          childToSupervisor.inject(line);
        },
      },
    });
    while (!mailBus.registered().includes("deployment-x@example.com")) {
      await new Promise((r) => setTimeout(r, 1));
    }
    await childSender.send({
      type: "ready",
      data: {
        childPid: 11111,
        childPublicKey: hexEncode(childIpcKeyPair.publicKey),
      },
    });
    await spawnPromise;
    return {
      supervisor,
      mailBus,
      supervisorToChild,
      childSender,
    };
  }

  test("default deriveMailAuditRef stamps `in-process` store on enqueued envelopes", async () => {
    const inbox = createMemoryInboxPrimitives();
    const { supervisor, mailBus } = await buildFifoTestFixture({
      label: "fifo-default-audit-",
      inbox,
    });
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("audit-default-1"),
    );
    // Wait for the enqueue to land in the in-memory inbox. The
    // dispatch loop may pull the entry into `processing` before the
    // assertion fires (the loop dequeues immediately once the
    // supervisor's spawn handshake completes), so the check covers
    // every claim-check substate.
    const deadline = Date.now() + 500;
    while (Date.now() < deadline) {
      const snap = inbox.snapshot("deployment-x@example.com");
      if (
        snap.inbox.size > 0 ||
        snap.processing.size > 0 ||
        snap.consumed.size > 0
      ) {
        break;
      }
      await new Promise((r) => setTimeout(r, 1));
    }
    const snapshot = inbox.snapshot("deployment-x@example.com");
    const all = [
      ...snapshot.consumed.values(),
      ...snapshot.processing.values(),
      ...snapshot.inbox.values(),
    ];
    expect(all.length).toBeGreaterThanOrEqual(1);
    const first = all[0];
    if (first === undefined) throw new Error("unreachable");
    expect(first.mailAuditRef.store).toBe("in-process");
    expect(first.mailAuditRef.path.length).toBeGreaterThan(0);
    await supervisor.shutdown();
  });

  test("deriveMailAuditRef override is invoked with messageId and stamps the envelope", async () => {
    const inbox = createMemoryInboxPrimitives();
    const observed: { messageId: string; len: number }[] = [];
    const { supervisor, mailBus } = await buildFifoTestFixture({
      label: "fifo-override-audit-",
      inbox,
      deriveMailAuditRef: (messageId, rawMessage) => {
        observed.push({ messageId, len: rawMessage.byteLength });
        return {
          store: "test-audit",
          path: `deployment-x/${messageId}`,
        };
      },
    });
    const payload = new TextEncoder().encode("audit-override-1");
    mailBus.deliver("deployment-x@example.com", payload);
    const deadline = Date.now() + 500;
    while (Date.now() < deadline && observed.length === 0) {
      await new Promise((r) => setTimeout(r, 1));
    }
    expect(observed.length).toBe(1);
    const observedEntry = observed[0];
    if (observedEntry === undefined) throw new Error("unreachable");
    expect(observedEntry.len).toBe(payload.byteLength);
    expect(observedEntry.messageId.length).toBeGreaterThan(0);
    const overrideSnapshot = inbox.snapshot("deployment-x@example.com");
    const allEntries = [
      ...overrideSnapshot.inbox.values(),
      ...overrideSnapshot.processing.values(),
      ...overrideSnapshot.consumed.values(),
    ];
    expect(allEntries.length).toBeGreaterThanOrEqual(1);
    const first = allEntries[0];
    if (first === undefined) throw new Error("unreachable");
    expect(first.mailAuditRef.store).toBe("test-audit");
    expect(first.mailAuditRef.path).toBe(
      `deployment-x/${observedEntry.messageId}`,
    );
    await supervisor.shutdown();
  });

  test("two queued messages fire once, then reject the post-terminal message", async () => {
    const inbox = createMemoryInboxPrimitives();
    // The supervisor's per-cohort terminal broadcaster gates each
    // dispatch on a `terminal.event` upstream control frame the test
    // mints through the child IPC sender. Until the test sends the
    // frame, the dispatch loop sits on `waitForRunTerminal` for the
    // forwarded run.
    const { supervisor, mailBus, supervisorToChild, childSender } =
      await buildFifoTestFixture({
        label: "fifo-serial-",
        inbox,
      });
    // Two messages. The first fires the run; the second waits behind its
    // terminal gate and is then rejected.
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("serial-msg-A"),
    );
    mailBus.deliver(
      "deployment-x@example.com",
      new TextEncoder().encode("serial-msg-B"),
    );
    // Helper that pulls the runId carried on the first/next
    // trigger.fire frame the supervisor wrote to the child stream.
    function triggerRunIds(): string[] {
      return parseTriggerFireRunIds(supervisorToChild.flushed());
    }
    // Wait for the first trigger.fire to land on the child stream.
    const deadlineOne = Date.now() + 500;
    while (Date.now() < deadlineOne) {
      if (triggerRunIds().length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    let firedIds = triggerRunIds();
    expect(firedIds.length).toBe(1);
    // Release the first run's terminal event. The dispatch loop
    // proceeds to markConsumed and pulls the second message.
    const firstRunId = firedIds[0];
    if (firstRunId === undefined) throw new Error("first run not minted");
    await childSender.send({
      type: "terminal.event",
      data: {
        runId: firstRunId,
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });
    const deadlineTwo = Date.now() + 500;
    while (Date.now() < deadlineTwo) {
      if (inbox.snapshot("deployment-x@example.com").consumed.size >= 2) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    firedIds = triggerRunIds();
    expect(firedIds).toEqual([firstRunId]);
    const consumed = inbox.snapshot("deployment-x@example.com").consumed;
    expect(consumed.size).toBe(2);
    expect(
      [...consumed.values()].some(
        (entry) => entry.rejection?.code === "workflow_run_terminal",
      ),
    ).toBe(true);
    await supervisor.shutdown();
  });

  test("spawn-time replayProcessingToInbox moves orphaned processing entries back to inbox", async () => {
    const inbox = createMemoryInboxPrimitives();
    // Seed a `processing/` entry before the supervisor spawns. The
    // entry should be moved back to `inbox/` during `spawn()`.
    const state = inbox.snapshot("deployment-x@example.com");
    state.processing.set("1000-msg-orphan", {
      messageId: "msg-orphan",
      receivedAt: 1000,
      mailAuditRef: { store: "test", path: "test/orphan" },
    });
    // In the unified-dispatch path markConsumed waits for the child to
    // reach terminal or park before consuming the message.
    const { supervisor, supervisorToChild, childSender } =
      await buildFifoTestFixture({
        label: "fifo-replay-spawn-",
        inbox,
      });
    // Wait for the dispatch loop to pull the recovered inbox entry
    // and send the trigger.fire.
    const deadline = Date.now() + 500;
    while (Date.now() < deadline) {
      const flushed = supervisorToChild.flushed();
      const triggerFires = flushed.filter((f) => f.includes("trigger.fire"));
      if (triggerFires.length >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    const triggerFires = supervisorToChild
      .flushed()
      .filter((f) => f.includes("trigger.fire"));
    expect(triggerFires.length).toBeGreaterThanOrEqual(1);

    // Drive the run to terminal so markConsumed can proceed.
    await new Promise((r) => setTimeout(r, 10));
    await childSender.send({
      type: "terminal.event",
      data: {
        runId: "deployment-x@example.com",
        seq: 0,
        kind: "RunCompleted",
        at: "test",
      },
    });

    // The processing entry was moved back to inbox during spawn,
    // then dequeued by the dispatch loop and forwarded as trigger.fire.
    // After the child reaches terminal, markConsumed moves it to consumed.
    const consumedDeadline = Date.now() + 1000;
    while (Date.now() < consumedDeadline) {
      const snapshot = inbox.snapshot("deployment-x@example.com");
      if (snapshot.consumed.size >= 1) break;
      await new Promise((r) => setTimeout(r, 1));
    }
    const snapshot = inbox.snapshot("deployment-x@example.com");
    expect(snapshot.consumed.size).toBe(1);
    const consumedEntry = [...snapshot.consumed.values()][0];
    if (consumedEntry === undefined) throw new Error("unreachable");
    expect(consumedEntry.messageId).toBe("msg-orphan");
    await supervisor.shutdown();
  });
});
