// backends/pi.ts — the pi-subagents backend: RPC spawn/fleet + the file
// control channel for steering detached async runs. This is the ACTIVE backend
// for the pi extension (the queue tools run in an interactive pi session).
//
// The seam contract lives in backends/types.ts; backends/index.ts is the
// factory. pi-subagents specifics live here: the RPC spawn/fleet seams and the
// control-channel steer. Steer is written directly to the run's control inbox
// (the mechanism pi-subagents' own FleetView uses) because the public
// `api/control-channel` export only exposes stop, and the subagent steer
// action routes workflow-mode detached runs to the foreground path ('no live
// foreground child') — an upstream routing gap.
//
// FLEET-vs-INVENTORY PARITY (AUTOPILOT-6): the RPC `status → fleet` count is
// gated on the caller's CURRENT session id — a workflow-parent run started in
// an earlier session or surviving a runtime restart is invisible to it (the
// observed "FLEET 0/3 while items ran" bug). fleetStatus() therefore unions
// the RPC count with the in-flight async runs on disk (scanAsyncRunInventory
// scans the same async-subagent-runs root asyncDirFor resolves) so no tick or
// zombie sweep can ever see a fake idle fleet from an undercounting RPC. The
// mode-scoped inventory is the AUTHORITATIVE count and the RPC is a fail-closed
// FLOOR (AUTOPILOT-117): it can only raise occupancy, never lower it to a fake
// 0 — dropping it outright made an empty disk read the sole authority for a 0
// and condemned live workers (docs/AUTOPILOT-117-residual-capacity-coupling.md).
//
// RUNS, NOT ENTRIES (AUTOPILOT-95): every worker the harness spawns is a
// one-step WORKFLOW SCRIPT, which pi-subagents materializes as TWO run dirs
// sharing the spawning session — the workflow WRAPPER (status.mode ===
// "workflow") and its nested CHILD (status.mode === "single",
// status.parentWorkflowRunId = the wrapper id) — where the child is the run
// that actually does the work. pi-subagents' RPC fleet counts BOTH as active
// children (its `totalActive` is an ENTRY count; the upstream TUI collapses
// child into parent via `parentKey`, the count does not), and a naive disk
// scan counts both too: one logical run reads as TWO, so a 3-run fleet reads
// as 6 and pins occupancy at the cap on half the real work. scanAsyncRun
// collapses each wrapper into its in-flight child/children for the run-level
// count, while keeping the raw entry count visible alongside it. Two genuinely
// independent children under a wrapper stay TWO runs — the duplicate-worker
// condition is never hidden.
//
// MODE SCOPING (AUTOPILOT-98): every pi mode/profile (`…/.pi/work`,
// `…/.pi/personal`) writes its run records into a SHARED, UID-keyed directory.
// Counting all of them made one mode's runs occupy the other mode's slots —
// live: personal runs pinned the work mode at capacity so its dispatch silently
// refused. A run is therefore attributed to the mode it was SPAWNED under, read
// from the record's own session path: pi-subagents stores the invoking session
// file under `<profileDir>/sessions/…` in status.json's `sessionId` (with
// `sessionDir`/`sessionRoot` sharing the prefix), and THAT profile dir is the
// mode marker. A run under a different profile dir is excluded; a run with NO
// session path (no mode marker) is counted as ours (conservative, fail-closed —
// an unattributable run can only inflate occupancy, never invent free
// capacity). The mode filter is applied INSIDE the union, so AUTOPILOT-6's
// rescue is preserved for THIS mode: a run the RPC cannot see (pre-activation,
// cross-session, post-restart) still counts as long as it was spawned under our
// profile.
//
// THE ROOT MUST BE THE WRITER'S (AUTOPILOT-117): the inventory root is NOT
// profile-scoped — it mirrors where pi-subagents ITSELF writes (its own root
// env, else the UID scope). A profile-scoped default (an AUTOPILOT-117
// regression) diverges from the writer, scans an EMPTY directory, and an empty
// directory reads ZERO — and zero is exactly what zombie reconciliation treats
// as "idle", so live workers were condemned. The root therefore follows the
// writer unless the launcher exported a per-profile root on BOTH sides
// (`PI_SUBAGENTS_TEMP_ROOT`).
//
// Ack protocol — HAND-ROLLED REIMPLEMENTATION (AUTOPILOT-62): pi-subagents does
// NOT export requestAsyncSteer/waitForSteeringAction/etc. as a public npm API,
// and is not even a dependency of this package (no package.json entry, not in
// node_modules — the real install lives entirely outside this repo's module
// graph). Everything below is this repo's own read/write of pi-subagents'
// INTERNAL, UNVERSIONED on-disk wire format. It was traced by inspecting the
// ACTUAL RUNNING pi-subagents build's jiti-compiled module cache on the
// machine ($TMPDIR/jiti/background-control-channel.*.mjs,
// background-steering.*.mjs, foreground-async-steering-action.*.mjs) rather
// than any published source — re-run that same trace after any pi-subagents
// version bump, since this can silently drift. Confirmed against that live
// cache (AUTOPILOT-62 investigation):
//   - request write: <asyncDir>/control/steer-requests/<ts13>-<base64url id>
//     .json — a FLAT directory (steerRequestsDir in control-channel.ts).
//     There is NO per-child "steer-targets/<index>" subdirectory anywhere in
//     the protocol (that was this repo's original AUTOPILOT-62 bug — a
//     readiness gate on a directory the runtime never creates); the target is
//     the `targetIndex`/`targetIndexes` FIELD inside the request JSON.
//   - closed-inbox refusal: <asyncDir>/control/steer-inbox-closed.json —
//     checked before AND after the write (the real requestAsyncSteer
//     re-checks post-write and deletes the request if the inbox closed
//     mid-write; this reimplementation mirrors that race-safety).
//   - delivery outcome: NOT a `steer-acks/<index>/` directory — nothing in
//     the currently-running runtime ever creates one (the env var plumbing
//     for it, PI_SUBAGENT_STEER_ACK_DIR_ENV, is dead: no launch path sets it).
//     The REAL signal is <asyncDir>/status.json's `.steering.recent[]` array
//     (written by the runner via applySteerDelivery → writeStatusPayload) —
//     each entry has `targets: [{index, state, reason?}]` where state ∈
//     scheduled|routed|queued|delivered|late|failed|recovered. This mirrors
//     pi-subagents' own waitForSteeringAction/actionResultFromSteeringStatus
//     (background/steering.ts), degenerate to our single-target (index 0)
//     case: delivered|recovered → ack "delivered"; scheduled|queued → ack
//     "queued"; failed|late → thrown as a failed delivery; routed → keep
//     polling (routed to the child, not yet settled).

import { randomUUID } from "node:crypto";
import { tmpdir } from "node:os";
import { join, resolve as resolvePath } from "node:path";
import { existsSync, mkdirSync, readFileSync, writeFileSync, renameSync, readdirSync, rmSync } from "node:fs";
import type { SubagentBackend, PiLike } from "./types.ts";
import type { CompletionEvent } from "../types.ts";
import { taskItemKey } from "../task-key.ts";

const RPC_REQUEST = "subagents:rpc:v1:request";
const RPC_REPLY = (id: string) => `subagents:rpc:v1:reply:${id}`;

/** Test seam for the pi backend's async-run inventory root (the same injection
 *  the opencode backend's runsDir offers): fleetStatus() reads the in-flight
 *  async runs found on disk (the mode-scoped authoritative occupancy,
 *  AUTOPILOT-117), so tests point the scan at a temp dir instead of the LIVE
 *  pi-subagents state. */
export interface PiBackendDeps {
  asyncDirRoot?: string;
  /** The pi config/profile dir that identifies THIS harness's own mode: the
   *  directory the spawning sessions' records live under
   *  (`<modeRoot>/sessions/<…>.jsonl`). Defaults to `$PI_CODING_AGENT_DIR`
   *  (the profile pi was launched in — e.g. `…/.pi/work` vs `…/.pi/personal`;
   *  set by the launcher, not the user's shell config). Test seam only;
   *  production reads the env. Empty/absent → the mode filter is DISABLED
   *  (every in-flight run counts), the fail-closed default; AUTOPILOT-117 makes
   *  that state LOUD (a warning + `modeFilterDisabled` on the fleet result)
   *  rather than silent, since it makes fixes 1-2 inert. */
  modeRoot?: string;
}

/** The pi config/profile dir a run record was spawned under — the substring
 *  before `/sessions/` in its recorded session path. pi-subagents writes the
 *  invoking session's FILE path into every run's status.json (`sessionId`),
 *  with `sessionDir`/`sessionRoot` carrying the same prefix; that prefix IS
 *  the mode marker (two modes run two pi profiles under two config dirs).
 *  Returns null when the record names no session path — i.e. it carries NO
 *  mode marker and cannot be attributed. */
export function sessionProfileDir(status: { sessionId?: unknown; sessionDir?: unknown; sessionRoot?: unknown }): string | null {
  const raw = [status.sessionId, status.sessionDir, status.sessionRoot].find((v): v is string => typeof v === "string" && v.length > 0);
  if (!raw) return null;
  const idx = raw.indexOf("/sessions/");
  if (idx <= 0) return null; // no `/sessions/` segment — not a pi session path
  return raw.slice(0, idx);
}

/** Does an in-flight run record belong to THIS harness's mode?
 *  - `"own"`     — its session path resolves under `modeRoot`: counts.
 *  - `"foreign"` — a DIFFERENT profile dir: another mode's run, never counted.
 *  - `"unknown"` — no mode marker in the record, OR no `modeRoot` to compare
 *                  against: counts. CONSERVATIVE, fail-closed default: an
 *                  unattributable run can only inflate occupancy, never invent
 *                  free capacity (AUTOPILOT-98). */
export function runMode(status: { sessionId?: unknown; sessionDir?: unknown; sessionRoot?: unknown }, modeRoot: string): "own" | "foreign" | "unknown" {
  if (!modeRoot) return "unknown";
  const dir = sessionProfileDir(status);
  if (dir === null) return "unknown";
  // resolve() normalizes trailing slashes / `..` / duplicate slashes so a
  // cosmetic difference in the recorded path is never mistaken for another
  // mode (which would under-count our own occupancy and over-spawn).
  return resolvePath(dir) === resolvePath(modeRoot) ? "own" : "foreign";
}

/** Collapse workflow wrappers into their in-flight children WITHIN one
 *  attribution bucket: a CONTAINER (a dir that parents another dir in the same
 *  map) is replaced by its children, so wrapper+child is ONE run while two
 *  children stay TWO (AUTOPILOT-95, restricted to one bucket so a foreign
 *  wrapper can never collapse one of our runs). */
function collapseRuns(inFlight: Map<string, string | null>): number {
  const containers = new Set<string>();
  for (const parent of inFlight.values()) if (parent && inFlight.has(parent)) containers.add(parent);
  return inFlight.size - containers.size;
}

/** The async-run root pi-subagents ITSELF writes to, given the current env —
 *  the directory this backend must scan to read REAL occupancy.
 *
 *  pi-subagents resolves its temp scope from `PI_SUBAGENTS_TEMP_ROOT` when set
 *  (`<that>/async-subagent-runs`), else the UID scope
 *  (`<tmpdir>/pi-subagents-uid-<uid>/async-subagent-runs`, its `resolveTempScopeId`
 *  → `uid-<uid>`). There is NO profile segment in the writer's path. Verified
 *  live: the UID directory holds the run records; a code-side profile-scoped
 *  default reads an empty directory, which reads zero, and zero means "idle".
 *
 *  This is the ONE authority for both sides. `defaultAsyncRunRoot()` returns it
 *  whenever the launcher has not exported an override, so the disk half can be
 *  trusted exactly when the writer's scope is what we resolved (AUTOPILOT-117). */
export function writerAsyncRunRoot(): string {
  const runtimeRoot = process.env.PI_SUBAGENTS_TEMP_ROOT?.trim();
  if (runtimeRoot) return join(resolvePath(runtimeRoot), "async-subagent-runs");
  return join(tmpdir(), `pi-subagents-uid-${process.getuid?.() ?? ""}`, "async-subagent-runs");
}

/** The async-run inventory root this backend scans (fleetStatus/asyncDirFor/
 *  steer). Resolution order:
 *
 *  1. `AUTOPILOT_PI_ASYNC_ROOT` — this repo's documented explicit override
 *     (tests point the inventory scan at an isolated temp root so a LIVE
 *     pi-subagents session never merges into mocked-host assertions).
 *  2. `writerAsyncRunRoot()` — the runtime's OWN resolution (its root env, else
 *     the UID scope). Chosen on EVIDENCE (`PI_SUBAGENTS_TEMP_ROOT` present =
 *     the launcher exported a per-profile root on both sides), never on
 *     `PI_CODING_AGENT_DIR`, which is set in the live environment while the
 *     writer's root is still UID-scoped (AUTOPILOT-117).
 *
 *  NO profile-scoped default: a root that diverges from the writer is worse
 *  than a shared one, because it reads empty and empty reads zero ("idle"). */
export function defaultAsyncRunRoot(): string {
  if (process.env.AUTOPILOT_PI_ASYNC_ROOT) return process.env.AUTOPILOT_PI_ASYNC_ROOT;
  return writerAsyncRunRoot();
}

/** The pi child session's LAST assistant text — the run's final deliverable
 *  (the reviewer's verdict line + the review body). Line-based read of the
 *  session.jsonl (assistant text parts only). */
export function readLastAssistantText(sessionFile: string): string | null {
  try {
    let last = "";
    for (const line of readFileSync(sessionFile, "utf8").split(/\r?\n/)) {
      if (!line.trim()) continue;
      let msg: { message?: { role?: string; content?: Array<{ type?: string; text?: string }> } } | null = null;
      try {
        msg = JSON.parse(line);
      } catch {
        continue;
      }
      const m = msg?.message;
      if (m?.role !== "assistant" || !Array.isArray(m.content)) continue;
      const text = (m.content ?? [])
        .filter((x) => x?.type === "text" && typeof x.text === "string")
        .map((x) => x.text ?? "")
        .join("");
      if (text.trim()) last = text;
    }
    return last || null;
  } catch {
    return null;
  }
}

/** The pi child session's FIRST user text — the TASK the harness dispatched
 *  (every task builder in framework/auto-dispatch.ts starts it with a
 *  `KEY: <key>` line). AUTOPILOT-104: this is the run's OWN statement of which
 *  item it belongs to, used to attribute a reviewer completion whose
 *  dispatch-time `reviewerRunId` link is missing (see task-key.ts). Same
 *  line-based read as readLastAssistantText. */
export function readFirstUserText(sessionFile: string): string | null {
  try {
    for (const line of readFileSync(sessionFile, "utf8").split(/\r?\n/)) {
      if (!line.trim()) continue;
      let msg: { message?: { role?: string; content?: Array<{ type?: string; text?: string }> } } | null = null;
      try {
        msg = JSON.parse(line);
      } catch {
        continue;
      }
      const m = msg?.message;
      if (m?.role !== "user" || !Array.isArray(m.content)) continue;
      const text = (m.content ?? [])
        .filter((x) => x?.type === "text" && typeof x.text === "string")
        .map((x) => x.text ?? "")
        .join("");
      if (text.trim()) return text;
    }
    return null;
  } catch {
    return null;
  }
}

export function createPiBackend(pi: PiLike, deps: PiBackendDeps = {}): SubagentBackend {
  // The async-run inventory root — the directory asyncDirFor/steerRouteDir
  // already read (pi-subagents writes every detached run's status.json here).
  // fleetStatus() scans it for the FLEET-vs-INVENTORY parity union.
  const asyncRoot = deps.asyncDirRoot ?? defaultAsyncRunRoot();
  // AUTOPILOT-117 fix 3 (ROOT half): the degraded state that actually condemned
  // live workers is an inventory root that does NOT MATCH where pi-subagents
  // writes — the scan then reads a directory the runtime never fills, occupancy
  // reads 0, and 0 means "idle" to zombie reconciliation. Fix 1 makes the
  // default agree by construction; this comparison keeps any divergence VISIBLE
  // (surfaced as `inventoryRootMatchesWriter` on the fleet result, warned once
  // per backend). `deps.asyncDirRoot` is the explicit test seam, so an injected
  // root is exempt from the warning (the caller chose it deliberately).
  const writerRoot = writerAsyncRunRoot();
  const inventoryRootMatchesWriter = resolvePath(asyncRoot) === resolvePath(writerRoot);
  let warnedInventoryRootMismatch = false;
  function warnInventoryRootMismatchOnce(): void {
    if (warnedInventoryRootMismatch || deps.asyncDirRoot !== undefined) return;
    warnedInventoryRootMismatch = true;
    console.warn(
      `[orchestrator-autopilot] pi backend: the async-run inventory root (${asyncRoot}) does NOT match where pi-subagents writes (${writerRoot}) — the disk scan can read a directory the runtime never fills, so occupancy can read 0 while runs are live (AUTOPILOT-117 fix 3). Unset AUTOPILOT_PI_ASYNC_ROOT, or export the runtime root via PI_SUBAGENTS_TEMP_ROOT so BOTH sides agree.`,
    );
  }
  // The profile dir naming THIS harness's mode (see PiBackendDeps.modeRoot).
  // Empty → the mode filter is DISABLED (every in-flight run counts), the
  // fail-closed default when no mode identity is available.
  const modeRoot = deps.modeRoot ?? process.env.PI_CODING_AGENT_DIR ?? "";
  // AUTOPILOT-117 fix 3: a MISSING modeRoot silently disables the whole mode
  // filter, which makes fixes 1-2 INERT while every count still looks correct.
  // Observe it where it is MATERIAL — an in-flight run that COULD have been
  // attributed (its record carries a session path) — and say so LOUDLY, once
  // per backend. The `unknown ⇒ counts` policy itself is unchanged (fail-closed
  // is deliberate); only its silence is removed. The same fact rides on the
  // fleetStatus result as `modeFilterDisabled` for machine readers/telemetry.
  let warnedModeFilterDisabled = false;
  function warnModeFilterDisabledOnce(): void {
    if (warnedModeFilterDisabled) return;
    warnedModeFilterDisabled = true;
    console.warn(
      "[orchestrator-autopilot] pi backend: PI_CODING_AGENT_DIR is empty, so the fleet mode filter is DISABLED — every profile's in-flight runs count toward THIS mode's capacity (AUTOPILOT-117 fix 3). Set PI_CODING_AGENT_DIR (done by the launcher/wrapper per profile) or pass PiBackendDeps.modeRoot.",
    );
  }

  /** Extract the async workflow id from a spawn RPC reply (structured or text). */
  function extractRunId(reply: unknown): string {
    const d = (reply as { data?: Record<string, unknown> } | null)?.data;
    if (d) {
      const details = d.details as Record<string, unknown> | undefined;
      for (const c of [d.runId, d.id, (d.run as { id?: string } | undefined)?.id, details?.asyncId, details?.runId]) {
        if (typeof c === "string" && c) return c;
      }
      const text = typeof d.text === "string" ? d.text : "";
      const m = text.match(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/i);
      if (m) return m[0];
      const m2 = text.match(/Async workflow \[([0-9a-f]+)\]/i);
      if (m2) return m2[1];
    }
    return "";
  }

  /**
   * One request/reply round-trip on the pi event bus.
   *
   * TIMEOUT SEMANTICS (load-bearing): a timeout RESOLVES `{success:false,
   * error}` — it never rejects. Callers depend on that shape: fleetStatus()
   * turns it into `null`, and the framework runner's degraded-episode logic
   * (AUTOPILOT-24) tells a null answer from a throw.
   *
   * TIMER SAFETY (AUTOPILOT-31): `unsub` is DECLARED AND INITIALIZED before the
   * timer is armed, and a synchronous subscribe/emit failure clears the timer.
   * Previously the timer was armed first and its callback closed over a
   * not-yet-initialized `const unsub`: when `pi.events.on` threw, the orphaned
   * timer fired timeoutMs later and hit `unsub` in its temporal dead zone —
   * a ReferenceError thrown FROM A TIMER CALLBACK, outside this promise chain,
   * so no caller `.catch` could contain it and the HOST PROCESS died instead of
   * degrading. Invariants: no timer callback may reference a possibly
   * uninitialized binding, and a failed subscribe never leaves an armed timer.
   */
  function rpc(method: string, params: Record<string, unknown>, timeoutMs: number): Promise<unknown> {
    return new Promise((resolve, reject) => {
      const requestId = randomUUID();
      let unsub: () => void = () => {}; // no-op until the subscription replaces it
      const timer = setTimeout(() => {
        unsub();
        resolve({ success: false, error: { message: `${method} RPC timed out` } });
      }, timeoutMs);
      try {
        const off = pi.events.on(RPC_REPLY(requestId), (reply: unknown) => {
          clearTimeout(timer);
          unsub();
          resolve(reply);
        });
        // a bus whose on() returns no unsubscribe would be the same
        // host-killing throw-from-a-timer, one frame later
        if (typeof off === "function") unsub = off;
        pi.events.emit(RPC_REQUEST, { version: 1, requestId, method, params });
      } catch (e) {
        // A broken bus rejects HERE (the same failure the executor's throw
        // already surfaced to callers) — but with the timer disarmed, so
        // nothing fires into a dead request afterwards.
        clearTimeout(timer);
        try {
          unsub();
        } catch {
          // tearing down a half-built subscription is best-effort
        }
        reject(e instanceof Error ? e : new Error(String(e)));
      }
    });
  }

  /**
   * Resolve the child's steer-route control inbox: the FLAT <asyncDir>/control
   * dir, and only that. AUTOPILOT-62 traced the live running pi-subagents
   * build (see the ack-protocol note above) and found NO
   * `control/workflow-foreground/<workflowId>/control` nesting anywhere in
   * the real protocol — workflow-mode foreground children are steered through
   * an IN-PROCESS live-child object private to the pi-subagents extension's
   * own state (unreachable from this separate extension); background/detached
   * runs (including workflow parents and their steps) are always steered via
   * the FLAT <asyncDir>/control dir. The former nested-dir probe was dead code
   * — its marker files are produced by no writer in this tree, so it could
   * only ever fall through to the flat dir — and has been removed; do not
   * build new logic on the assumption the nested path is real. Returns null
   * when no route exists (run gone).
   */
  function steerRouteDir(asyncDir: string): string | null {
    const flat = join(asyncDir, "control");
    return existsSync(flat) ? flat : null;
  }

  /**
   * Read one steer request's target outcome from the run's status.json — the
   * REAL ack signal (see the hand-rolled-reimplementation note above), not a
   * steer-acks directory. Returns undefined until the runner has recorded a
   * state for this request/target (or if status.json/the field is absent).
   */
  function readSteeringTarget(statusPath: string, requestId: string, targetIndex: number): { state?: string; reason?: string } | undefined {
    try {
      const status = JSON.parse(readFileSync(statusPath, "utf8")) as {
        steering?: { recent?: Array<{ id?: string; targets?: Array<{ index?: number; state?: string; reason?: string }> }> };
      };
      const request = status.steering?.recent?.find((r) => r.id === requestId);
      return request?.targets?.find((t) => t.index === targetIndex);
    } catch {
      return undefined;
    }
  }

  /**
   * The in-flight async-run inventory under asyncRoot, counted two ways
   * (AUTOPILOT-95): `entries` counts run DIRS (the `subagent status fleet`
   * rows — a workflow wrapper and its nested child are two), `runs` counts
   * LOGICAL runs (the wrapper collapses into its in-flight child/children).
   *
   * pi-subagents gates its RPC fleet on the CALLER's CURRENT session id, so a
   * workflow-parent run that started in an earlier session (engine restart,
   * cross-session deferral) is invisible to the RPC while its run dir still
   * reports queued|running. FLEET-vs-INVENTORY PARITY: fleetStatus() unions the
   * RPC count with this scan — the RPC/runs number is the FLOOR, never a fake
   * 0. Terminal dirs (state moved to complete/failed/...) and foreign entries
   * never count; unreadable entries are skipped, never inflated. Mirrors the
   * run-dir identification asyncDirFor uses: status.json whose runId (when
   * present) matches the dir name.
   *
   * RUN-LEVEL RULE (the distinction that must not be lost): a non-child
   * in-flight dir that OWNS at least one in-flight child is a CONTAINER — its
   * own row is replaced by its children's rows, so one wrapper + one child is
   * ONE run while two children under it are TWO (a genuine duplicate-worker
   * stays visible, never collapsed away). A child whose wrapper is already
   * terminal (parent absent from the inventory) counts on its own, so an
   * orphaned live worker is never hidden either.
   */
  function scanAsyncRunInventory(): { entries: number; runs: number; foreignRuns: number; unattributedRuns: number } {
    let names: string[];
    try {
      names = readdirSync(asyncRoot).filter((n) => !n.startsWith(".")); // skip index dirs (.active-runs …)
    } catch {
      return { entries: 0, runs: 0, foreignRuns: 0, unattributedRuns: 0 }; // inventory absent/unreadable — the RPC count stands alone
    }
    // Each bucket maps a run dir id → its parent workflow run id (null when it
    // is a top-level run). MODE SCOPING (AUTOPILOT-98): every mode sharing this
    // async-run root writes here, so a run is only counted for the mode it was
    // spawned under — `counted` = this mode's own dirs PLUS unattributable
    // dirs (conservative), `foreign` = a DIFFERENT profile's dirs (excluded).
    const counted = new Map<string, string | null>();
    const unattributed = new Map<string, string | null>();
    const foreign = new Map<string, string | null>();
    for (const name of names) {
      try {
        const status = JSON.parse(readFileSync(join(asyncRoot, name, "status.json"), "utf8")) as { runId?: string; state?: string; parentWorkflowRunId?: string; sessionId?: unknown; sessionDir?: unknown; sessionRoot?: unknown } | null;
        if (!status) continue;
        if (typeof status.runId === "string" && status.runId !== name) continue; // not this run's dir
        if (status.state !== "queued" && status.state !== "running") continue;
        const parent = typeof status.parentWorkflowRunId === "string" && status.parentWorkflowRunId ? status.parentWorkflowRunId : null;
        const mode = runMode(status, modeRoot);
        // Fix 3: with NO modeRoot, a record that CARRIES a mode marker is
        // exactly the run the filter would have attributed — the material case.
        if (!modeRoot && sessionProfileDir(status) !== null) warnModeFilterDisabledOnce();
        if (mode === "foreign") {
          foreign.set(name, parent); // another mode's run — it must not occupy THIS mode's slot
        } else {
          counted.set(name, parent);
          if (mode === "unknown") unattributed.set(name, parent);
        }
      } catch {
        continue; // junk/foreign entry — an unreadable dir never counts
      }
    }
    return {
      entries: counted.size,
      runs: collapseRuns(counted),
      foreignRuns: collapseRuns(foreign),
      unattributedRuns: collapseRuns(unattributed),
    };
  }

  return {
    async spawn(task, opts) {
      const script = `return runs.run("main", ${JSON.stringify({
        agent: opts.agent ?? "worker",
        task,
        context: "fresh",
        worktree: opts.worktree ?? true,
        ...(opts.cwd ? { cwd: opts.cwd } : {}),
        // THE ENFORCED DEADLINE (KEY: AUTOPILOT-129): this CHILD-level
        // `timeoutMs` is the deadline that actually fires. The workflow script
        // around it is launched with NO timeout of its own, so the async run is
        // unbounded and cannot cut the child off early — the runtime reports
        // "Subagent timed out after <n>ms" with exactly the number passed here
        // (observed on two AUTOPILOT-126 runs: 3600000 → killed at 3614s and
        // 5400000 → killed at its budget; the wrapper's own status.json carries
        // no timeoutMs). The number is whatever the CALLER passes as
        // `opts.timeoutMs`: on the ITEM-DISPATCH lanes that IS
        // `resolveDispatchBudget(...).timeoutMs`, so the enforced deadline IS
        // the budget the item records — but this seam is NOT universal, so do
        // not read the claim as "every spawn site". The spec-pass recon lane
        // (auto-dispatch.ts `dispatchPendingSpecPass`) passes its own fixed
        // `SPEC_PASS_TIMEOUT_MS` (15 min) and deliberately does not record it
        // on the item. Read the quantifier as "the item-dispatch lanes".
        ...(opts.timeoutMs ? { timeoutMs: opts.timeoutMs } : {}),
        // Only when set: an absent override must leave the script byte-identical
        // (no "model" key at all), so the child inherits the session model.
        ...(opts.model ? { model: opts.model } : {}),
      })})`;
      const reply = await rpc("spawn", { workflowScript: script, context: "fresh" }, 30_000);
      const r = reply as { success?: boolean; error?: { message?: string } };
      if (r?.success !== true) throw new Error(r?.error?.message ?? "spawn rejected");
      return extractRunId(reply);
    },

    async fleetStatus() {
      const reply = await rpc("status", {}, 2500);
      const r = reply as { success?: boolean; data?: { fleet?: { totalActive?: number } } };
      const f = r?.data?.fleet;
      if (r?.success !== true || !f) return null;
      // OCCUPANCY IS THE MODE-SCOPED INVENTORY, FLOORED BY THE RPC
      // (AUTOPILOT-117 fix 2, corrected). The RPC `status → fleet` count is
      // session-gated to the caller's CURRENT session id and its rows carry NO
      // session path (only agent/startedAt/tokens), so it CANNOT be
      // mode-attributed — it must not be the authoritative occupancy. But
      // DROPPING it outright was unsafe: that removed the last thing standing
      // between a stale/invisible count and a flipped worker, because an empty
      // disk read then became the sole authority for a 0 (fix 1 is what makes
      // the disk read trustworthy; the floor keeps it safe until/unless the
      // root provably agrees).
      //
      // So: the mode-scoped inventory (`inv.runs`) is authoritative and
      // lower-bounds the count, and the RPC is a FAIL-CLOSED FLOOR — it can
      // only ever RAISE occupancy (over-count is safe; under-count flips LIVE
      // items). The wrapper/child duplicate entries the RPC counts are
      // subtracted first (RUNS, not entries — AUTOPILOT-95). AUTOPILOT-6's
      // rescue is preserved: the DISK inventory is what sees the cross-session
      // / pre-activation / post-restart run the RPC cannot.
      //
      // MODE SCOPING (AUTOPILOT-98): `inv` excludes another mode's runs from
      // `runs`/`entries`; `foreignRuns` carries how many were excluded. The raw
      // RPC count stays visible as `fleetActive`.
      if (!inventoryRootMatchesWriter) warnInventoryRootMismatchOnce();
      const inv = scanAsyncRunInventory();
      const rpcEntries = f.totalActive ?? 0;
      const duplicateChildEntries = inv.entries - inv.runs;
      const rpcRuns = Math.max(0, rpcEntries - duplicateChildEntries);
      const runs = Math.max(rpcRuns, inv.runs);
      return {
        totalActive: runs,
        runs,
        entries: Math.max(rpcEntries, inv.entries),
        foreignRuns: inv.foreignRuns,
        unattributedRuns: inv.unattributedRuns,
        fleetActive: rpcEntries,
        modeFilterDisabled: modeRoot === "",
        inventoryRootMatchesWriter,
      };
    },

    async steer(runId, message, mode, ackTimeoutMs = 4000) {
      const asyncDir = this.asyncDirFor(runId);
      if (!asyncDir) throw new Error("run dir not found — it likely completed or died; stop + re-dispatch instead");
      const routeDir = steerRouteDir(asyncDir);
      if (!routeDir) throw new Error("run has no steer route — it likely completed or died; stop + re-dispatch instead");
      // 1. capability pre-check: SOFTENED (AUTOPILOT-62) — traced the live
      //    running pi-subagents build and found ZERO evidence any
      //    "steer-capabilities" directory or supported:false publishing exists
      //    in the current runtime (checked the background runner, foreground
      //    executor, prompt runtime, and steering-action modules — none of
      //    them ever set steerCapabilityPath when launching a child). This
      //    check is therefore DORMANT in practice: capDir never exists, so it
      //    silently no-ops and falls through to the write below. Left in
      //    place defensively (harmless when absent) in case a future
      //    pi-subagents version reintroduces per-child capability publishing —
      //    but its absence is not evidence steering works, and nothing else
      //    should assume it is currently populated.
      const capDir = join(routeDir, "steer-capabilities");
      if (existsSync(capDir)) {
        for (const f of readdirSync(capDir).filter((n) => n.endsWith(".json"))) {
          let supported: boolean | undefined;
          try {
            supported = (JSON.parse(readFileSync(join(capDir, f), "utf8")) as { supported?: boolean }).supported;
          } catch {
            continue; // unreadable capability file — fall through to the write
          }
          if (supported === false) {
            throw new Error("the child Pi session does not support steering (capability supported:false) — steer will NOT be delivered; stop + re-dispatch instead");
          }
        }
      }
      // 2. closed-inbox pre-check (AUTOPILOT-62: this repo did not check this
      //    before — the runtime's real requestAsyncSteer does; a closed inbox
      //    means the run stopped accepting steering, e.g. it is wrapping up).
      const closedPath = join(routeDir, "steer-inbox-closed.json");
      if (existsSync(closedPath)) {
        throw new Error("run no longer accepts steering (steer inbox closed) — stop + re-dispatch instead");
      }
      // 3. write the request into the FLAT steer-requests/ dir (AUTOPILOT-62
      //    fix — see the hand-rolled-reimplementation note above; this used to
      //    gate on a "steer-targets/<index>" readiness directory that the
      //    runtime never creates, so it failed on 100% of calls).
      const index = 0;
      const requestsDir = join(routeDir, "steer-requests");
      mkdirSync(requestsDir, { recursive: true });
      const id = randomUUID();
      const request: Record<string, unknown> = {
        type: "steer",
        id,
        ts: Date.now(),
        message,
        targetIndex: index,
        source: "queue_steer",
      };
      if (mode && mode !== "steer") request.mode = mode;
      // match pi-subagents' steerRequestFileName: <ts13>-<base64url id>.json
      const name = `${String(request.ts as number).padStart(13, "0")}-${Buffer.from(id).toString("base64url")}.json`;
      const requestPath = join(requestsDir, name);
      // atomic write (temp + rename), mirroring the real writeAtomicJson
      const tmpPath = `${requestPath}.${process.pid}.${Date.now()}.tmp`;
      writeFileSync(tmpPath, JSON.stringify(request), "utf8");
      renameSync(tmpPath, requestPath);
      if (existsSync(closedPath)) {
        // raced with the inbox closing mid-write — mirror the real
        // requestAsyncSteer's post-write re-check: roll the request back
        // rather than leave one the runner will never consume.
        try {
          rmSync(requestPath, { force: true });
        } catch {
          // best-effort rollback
        }
        throw new Error("run stopped accepting steering before the request was committed — stop + re-dispatch instead");
      }
      // 4. verify delivery via status.json's `.steering` field — the REAL
      //    native ack signal (see the note above; NOT a steer-acks directory).
      const statusPath = join(asyncDir, "status.json");
      const deadline = Date.now() + ackTimeoutMs;
      while (Date.now() < deadline) {
        const target = readSteeringTarget(statusPath, id, index);
        if (target?.state) {
          if (target.state === "failed" || target.state === "late") {
            throw new Error(`steer FAILED delivery: ${target.reason ?? "child rejected it"}`);
          }
          if (target.state === "delivered" || target.state === "recovered") return { id, ack: "delivered" };
          if (target.state === "scheduled" || target.state === "queued") return { id, ack: "queued" };
          // "routed" — routed to the live child but not yet settled; keep polling
        }
        await new Promise((r) => setTimeout(r, 100));
      }
      throw new Error(
        `no steering acknowledgment within ${ackTimeoutMs}ms — the request was written but its delivery has not settled in status.json (the child may be slow, paused, or gone). Stop + re-dispatch if it is gone.`,
      );
    },

    /** Normalize the RAW async-complete payload into the CompletionEvent the
     *  shared machinery expects. The pi runtime's payload is flat
     *  ({id, success, state, asyncDir, sessionId}) — NO per-child agent or
     *  output — so the reviewer attribution (isReviewerRun) NEVER fired and
     *  queue_review wedged on the first reviewer run. Enrich from the run
     *  record: status.json's step agent + the child session's last assistant
     *  text. Same shape the opencode backend builds. */
    buildCompletionEvent(raw: unknown) {
      const base = (raw ?? {}) as Record<string, unknown>;
      const p = base as { id?: string; runId?: string; success?: boolean; state?: string; asyncDir?: string };
      const runId = p.id ?? p.runId ?? "";
      const success = p.success !== false && p.state !== "failed";
      // Start from the raw payload (its own results survive when the run
      // record is missing — e.g. tests, or a cleaned-up run dir); the run
      // record enrichment below overrides agent/results when found.
      const event: Record<string, unknown> = { ...base, runId, id: runId, agent: "workflow", success, status: success ? "completed" : "failed" };
      try {
        const dir = p.asyncDir && existsSync(p.asyncDir) ? p.asyncDir : this.asyncDirFor(runId);
        if (dir && existsSync(join(dir, "status.json"))) {
          const status = JSON.parse(readFileSync(join(dir, "status.json"), "utf8")) as { steps?: Array<{ agent?: string; sessionFile?: string; sessionName?: string }> };
          const step = (status.steps ?? [])[0];
          const agent = step?.agent ?? "";
          let output: string | null = null;
          if (step?.sessionFile && existsSync(step.sessionFile)) output = readLastAssistantText(step.sessionFile);
          // KEY: AUTOPILOT-104 — the item the run's own task names, so a review
          // verdict can be attributed when the dispatch-time ref is missing.
          // The child session carries the FULL task; the step's sessionName
          // (a `<agent>: <task excerpt>` display name) is the fallback when the
          // session file is unavailable.
          const taskText = step?.sessionFile && existsSync(step.sessionFile) ? readFirstUserText(step.sessionFile) : null;
          const itemKey = taskItemKey(taskText) ?? taskItemKey(step?.sessionName);
          if (itemKey) event.itemKey = itemKey;
          if (agent) {
            event.agent = agent;
            event.results = output ? [{ agent, output }] : [{ agent }];
          }
        }
      } catch {
        // best-effort — the raw event still routes worker flips by run id
      }
      return event as never;
    },

    asyncDirFor(runId) {
      try {
        const dir = join(asyncRoot, runId);
        if (existsSync(join(dir, "status.json"))) {
          // sanity: the status must reference this run
          try {
            const status = JSON.parse(readFileSync(join(dir, "status.json"), "utf8")) as { runId?: string; mode?: string };
            if (status.runId && status.runId !== runId) return null;
          } catch {
            // unreadable status — trust the dir name
          }
          return dir;
        }
        return null;
      } catch {
        return null;
      }
    },
  };
}