// orchestrator-autopilot.ts — pi adapter for the orchestrator autopilot.
//
// The framework core lives in THIS project (src/).
// (imported at runtime via jiti, same pattern as corrections.ts).
//
// The extension owns the PROGRAMMATIC QUEUE (queue.json) and the FLEET ledger
// (event-derived). The orchestrator interacts with the queue via tools:
//   queue_list     — read path (filter by status / last-change, compact view)
//   queue_add      — new proposal/approved item
//   queue_update   — status/flags/notes (validated transitions)
//   queue_dispatch — spawn the worker (same executor as subagent) AND record
//                    approved→active + runId atomically (worktree isolation
//                    enforced: dirty-tree pre-flight + blocked direct spawns)
//   queue_review   — spawn the reviewer for an `ai-review` item
//   queue_steer    — steer a running worker/reviewer via the control channel
// Completion (active→ai-review/failed) is event-inferred from
// subagent:async-complete. state.md is retired: it exists only as a one-time
// migration input (imported into queue.json on first activation).
//
// All subagent-runtime specifics (RPC spawn/fleet, control-channel steer, run
// dir resolution) live behind the backend seam
// (src/backends/) so the queue tools depend
// on an interface, not on pi-subagents internals.
//
// Behavior gated per-session (autopilot.sessions.json), interactive TUI only:
//   - /autopilot on|off|status|capacity <n>
//   - ticks via custom-role sendMessage (customType orchestrator-autopilot)
//   - children (async runners, --mode json -p) NEVER inject or tick.
//
// Design rules: NEVER throw (a broken autopilot must not affect pi); fail-safe
// to action; trigger-only (the orchestrator keeps all judgment).

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { Type } from "typebox";
import { join } from "node:path";

// The framework lib is THIS project — the extension lives at src/hosts/ and
// requires the lib relative to itself (no dotfiles-layout assumption). A
// published/installed layout can override via AUTOPILOT_LIB_DIR.
const LIB_DIR = (() => {
  if (process.env.AUTOPILOT_LIB_DIR) return process.env.AUTOPILOT_LIB_DIR;
  return join(__dirname, "..");
})();
// jiti resolves .ts imports at runtime; createRequire keeps this ESM-safe.
const { Autopilot } = require(`${LIB_DIR}/core.ts`) as typeof import("../core.ts");
const { loadAutopilotConfig, isAutopilotOn, appendTelemetry, autopilotCommand, autopilotModeMessage, resolveStateDir, writeSessionAutopilotState, probeStateDir, logStateDirProbe, readHostStateDir, autopilotHostConfigPath } = require(`${LIB_DIR}/config.ts`) as typeof import("../config.ts");
const { loadStoreOrNew, ensureMigrated, queueLengths } = require(`${LIB_DIR}/queue-store.ts`) as typeof import("../queue-store.ts");
const { createSubagentBackend, defaultRunsDir } = require(`${LIB_DIR}/backends/index.ts`) as typeof import("../backends/index.ts");
const { queueList, queueAdd, queueUpdate, queueDispatch, queueReview, queueSteer, repoCheck } = require(`${LIB_DIR}/tools/queue-ops.ts`) as typeof import("../tools/queue-ops.ts");
const { CONTRACTS } = require(`${LIB_DIR}/tools/contracts.ts`) as typeof import("../tools/contracts.ts");
const { createFrameworkRunner } = require(`${LIB_DIR}/framework/runner.ts`) as typeof import("../framework/runner.ts");
const { isUnisolatedWorkerSpawn } = require(`${LIB_DIR}/framework/auto-dispatch.ts`) as typeof import("../framework/auto-dispatch.ts");
const { flagForReview } = require(`${LIB_DIR}/framework/flag-review.ts`) as typeof import("../framework/flag-review.ts");
const { createScheduleManager, scheduledOffDue } = require(`${LIB_DIR}/framework/scheduled-off.ts`) as typeof import("../framework/scheduled-off.ts");
const { orchestrateProcedure, detectOrchestrateConflict, isPackagePath } = require(`${LIB_DIR}/framework/orchestrate-command.ts`) as typeof import("../framework/orchestrate-command.ts");
const { installPiReviewer } = require(`${LIB_DIR}/agents/install.ts`) as typeof import("../agents/install.ts");
const { registerDecisionPanel } = require(`${LIB_DIR}/hosts/pi-panel.ts`) as typeof import("./pi-panel.ts");
const { logTickDelivery } = require(`${LIB_DIR}/framework/tick-telemetry.ts`) as typeof import("../framework/tick-telemetry.ts");


// Install the framework-owned reviewer agent (idempotent, version-stamped).
// The framework DEPENDS on the reviewer (queue_review spawns it, the lifecycle
// parses its verdict) — shipping it with the extension is what makes the
// review step work on any machine. Never throw: a broken install must not
// affect pi (a missing reviewer surfaces as a generic spawn error at review
// time — re-install happens on the next activation).
try {
  installPiReviewer();
} catch {
  // silent — fail-closed check in queue_review reports if the agent is absent
}

const TICK_TYPE = "orchestrator-autopilot";

/** KEY: AUTOPILOT-122 — how long a submitted user prompt keeps the tick surface
 *  closed after its `input`/`before_agent_start` event. pi core reports the
 *  agent IDLE throughout the PRE-RUN awaits of `AgentSession.prompt` (the
 *  input/before-agent-start handlers, the auth check, the compaction check)
 *  BEFORE a run exists; a tick's triggered turn landing there claims the agent
 *  first, and the user's just-submitted prompt is then refused and recorded
 *  nowhere (pi-agent-core `agent.js:226-228`). The window normally closes when
 *  the run is claimed (`agent_start`) or settles (`agent_settled`); a prompt
 *  that FAILS in the pre-run (no model, bad auth) reaches neither, so this age
 *  bound is the liveness valve — past it the window is treated as closed, so a
 *  failed prompt cannot starve ticks forever. */
const USER_PROMPT_PENDING_MS = 120_000;

/** This host's config dir — the profile whose autopilot.host.json carries the
 *  state-dir fact (AUTOPILOT-32 D3). Shared by resolveStateDir + the state-dir
 *  probe (host-parity input). */
function hostConfigDir(): string | undefined {
  return process.env.PI_CODING_AGENT_DIR || undefined;
}

let stateDir = resolveStateDir(hostConfigDir());

export default function (pi: ExtensionAPI) {
  // Re-resolve per load (the test harness re-imports + mutates env between
  // cases; the real runtime loads the extension once). Production: called once.
  stateDir = resolveStateDir(hostConfigDir());

  let autopilot: InstanceType<typeof Autopilot> | null = null;
  let agentBusy = false;          // orchestrator (main agent) mid-turn
  let compacting = false;         // auto-compaction in progress — never inject into it
  let userPromptPendingAt = 0;    // a user prompt is inside pi core's PRE-RUN window (AUTOPILOT-122 part 2)
  let interactive = false;        // true only in the interactive TUI session
  let sessionId = "";             // current pi session id (per-session autopilot scope)
  let orchestratorLoaded = false; // the packaged procedure was injected into this session
  let conflictWarnedSession = ""; // D4: ONE orchestrate-conflict warning per session
  let hostCtx: { ui?: { notify(message: string, kind?: string): void } } | null = null; // last-seen command/session ctx — the scheduled-off fire needs ui.notify outside a handler


  function ensureAutopilot() {
    if (autopilot) return autopilot;
    const cfg = loadAutopilotConfig(stateDir);
    // KEY: AUTOPILOT-116 — pass the WHOLE resolved host config (`cfg`), not a
    // hand-picked subset. The named-key form let zombieGraceMinutes and
    // intakeSuppressionHours run the engine's built-in defaults while the
    // file's values were accepted and ignored; spreading the resolved config
    // ends the class (every engine key loadAutopilotConfig resolves reaches
    // the engine) instead of fixing a third key by name. `log` is a HOST
    // seam, not a config key, so it stays explicit after the spread.
    autopilot = new Autopilot({
      stateDir,
      ...cfg,
      log: (line) => {
        if (isAutopilotOn(stateDir, sessionId)) appendTelemetry(stateDir, line);
      },
    });
    return autopilot;
  }

  function storeOrNew() {
    return loadStoreOrNew(stateDir);
  }

  /** True while a user prompt is inside pi core's pre-run window — core reports
   *  the agent idle there, but the user's message is not yet recorded, so any
   *  triggered turn we inject would STEAL it (AUTOPILOT-122 part 2). See the
   *  USER_PROMPT_PENDING_MS note for the age bound. */
  function userPromptInFlight(): boolean {
    return userPromptPendingAt !== 0 && Date.now() - userPromptPendingAt < USER_PROMPT_PENDING_MS;
  }

  /** Subagent-runtime seam — see src/backends/. The backend is DEDUCED from
   *  the runtime, never env: this extension IS pi, so workers spawn on the
   *  pi-subagents backend (the opencode plugin wires its own opencode
   *  backend + completion signal to the same queue tools). */
  const backend = createSubagentBackend({ kind: "pi", pi: pi as never });

  function maybeInjectOrchestrate() {
    // Never inject into child/headless processes (async runners are
    // --mode json -p): their session_start must not load orchestrator mode or
    // collide with the runner's task prompt. Also per-session: only the session
    // that turned autopilot on gets orchestrator mode injected.
    if (!interactive || !sessionId) return;
    if (!isAutopilotOn(stateDir, sessionId)) return;
    if (orchestratorLoaded) return;
    // The PACKAGED procedure, injected verbatim (AUTOPILOT-32 D2) — no
    // "/orchestrate" indirection through a host-resolved projection any more.
    const procedure = orchestrateProcedure();
    if (!procedure) return; // broken package: the /orchestrate command reports it
    orchestratorLoaded = true; // assume success; reset on failure or session_start
    // deferAs: when the agent is busy but NOT streaming, the runtime's
    // followUp is ignored and sendUserMessage THROWS — defer to the settle.
    if (agentBusy) {
      runner.deferUserMessage(procedure); // shared deferral — flush at the settle
      return;
    }
    // deliverAs: "followUp" is REQUIRED — sendUserMessage THROWS "Agent is
    // already processing..." while the agent is streaming (e.g. mid-command),
    // and the rejection is async so a sync try/catch cannot contain it.
    // followUp queues the injection until the current turn finishes.
    try {
      const sent: unknown = pi.sendUserMessage(procedure, { deliverAs: "followUp" });
      if (sent && typeof (sent as Promise<void>).catch === "function") {
        void (sent as Promise<void>).catch(() => {
          orchestratorLoaded = false;
        });
      }
    } catch {
      // A synchronous throw (busy-not-streaming edge cases) must NOT take
      // down the /autopilot command — the shared deferral flushes at the settle.
      runner.deferUserMessage(procedure);
    }
  }

  /** ONE conflict warning per session (AUTOPILOT-32 D4): a leftover
   *  orchestrate.md projection alongside the plugin's command is how a
   *  superseded procedure keeps running unnoticed. The live enumeration comes
   *  from pi.getCommands() — the runtime's OWN list of what /orchestrate
   *  resolves to, not a guess. Never throws. */
  function warnOnOrchestrateConflict(ctx?: { ui?: { notify(message: string, kind?: string): void } }) {
    if (!interactive || !sessionId || conflictWarnedSession === sessionId) return;
    conflictWarnedSession = sessionId;
    try {
      const live: Array<{ path: string; owner: string; liveInHost: boolean }> = [];
      const commands = (pi as unknown as { getCommands?: () => Array<{ name: string; source?: string; sourceInfo?: { path?: string } }> }).getCommands?.() ?? [];
      for (const c of commands) {
        // `orchestrate:1` / `orchestrate:2` are the names pi mints when two
        // EXTENSIONS define the same command — the one competitor whose
        // precedence nobody documents, so it must stay in the report.
        if (!/^orchestrate(:\d+)?$/.test(c.name)) continue;
        const p = c.sourceInfo?.path;
        if (!p) continue; // unnameable: a report entry with no path helps nobody
        if (isPackagePath(p)) continue; // that IS this plugin's own command
        live.push({ path: p, owner: `pi ${c.source ?? "command"} (live: /${c.name})`, liveInHost: true });
      }
      const dir = hostConfigDir();
      const conflict = detectOrchestrateConflict({
        host: "pi",
        hostConfigDir: dir,
        live,
        // Same warning, one more fact: WHICH store this session writes to, and
        // whether anything declared it (an undeclared fallback can be another
        // profile's live queue).
        store: {
          resolved: stateDir,
          declaredBy: process.env.AUTOPILOT_STATE_DIR
            ? "AUTOPILOT_STATE_DIR"
            : dir && readHostStateDir(dir)
              ? autopilotHostConfigPath(dir)
              : null,
        },
      });
      if (!conflict) return;
      appendTelemetry(stateDir, JSON.stringify({
        t: new Date().toISOString(),
        type: "orchestrate-command-conflict",
        host: "pi",
        winner: conflict.winner,
        competing: conflict.competing.map((c: { path: string }) => c.path),
      }));
      ctx?.ui?.notify?.(conflict.message, "error");
    } catch {
      // a conflict warning must never affect the session
    }
  }

  // The shared tick machinery (gate + cooldown + delivery) lives in
  // src/framework/runner.ts — identical in both hosts. The pi host supplies
  // its gate state + the sendMessage delivery; runSweep/onCompletion etc.
  // wire the triggers.
  // The custom-role tick channel — ONE sendMessage implementation shared by
  // the runner's ticks AND the decision panel's decision ticks (same
  // customType orchestrator-autopilot, delivered NON-TRIGGERING — KEY:
  // AUTOPILOT-125): the orchestrator sees panel decisions exactly like
  // capacity ticks, no second mechanism.
  // pi.sendMessage is declared `: void` (not a Promise) — the installed pi
  // runtime writes its own `.catch` inside the extension binding and reports a
  // refused triggered turn as `Extension "<runtime>" error:` (it never throws,
  // and the extension never sees the promise). We still WATCH the return value:
  // a thenable is re-held by the runner on rejection, so no delivery race is
  // silently dropped.
  //
  // NO-TRIGGER CONTRACT (KEY: AUTOPILOT-125 — supersedes 122's triggered send;
  // single source of truth: docs/AUTOPILOT-51-tick-audit-gap.md, "AUTOPILOT-125").
  // This surface calls `pi.sendMessage` WITHOUT `triggerTurn`, so a tick can
  // never CLAIM the agent. When core is not streaming that is the
  // `_appendCustomMessage` branch (`agent-session.js:1130-1131`): the message is
  // RECORDED in the transcript immediately and reaches the session's next turn
  // — but it does NOT start one. A message that cannot claim the agent can
  // neither be refused (`pi-agent-core agent.js:227-228`) nor make the user's
  // just-submitted prompt the refused one; that is AUTOPILOT-122's residual
  // window (a tick landing before core's `input` event, between an interactive
  // submit and `prompt()` reaching `emitInput`), now UNREACHABLE rather than
  // merely handled. `deliverAs: "followUp"` is kept so the one racing case core
  // calls streaming queues rather than steering; both branches are
  // non-triggering.
  // The busy/compacting/user-prompt PRE-CHECK is kept as cheap insurance (it
  // holds a tick out of a mid-turn transcript), not as the defence: with
  // `triggerTurn` gone there is nothing left to refuse.
  // SILENT-DEFERRAL (AUTOPILOT-122, still in force on the pre-check): a tick
  // that cannot be injected RIGHT NOW is a HANDLED deferral, not an alarm —
  // `runner.deferTick` re-delivers it at the next settle, NEVER dropped, and a
  // rejected promise is returned (the re-hold the runner implements; the panel
  // callers only best-effort catch it).
  const deliverTick = (message: string): void | Promise<void> => {
    if (agentBusy || compacting || userPromptInFlight()) {
      runner.deferTick(message); // held + re-delivered at the settle — never dropped
      return Promise.reject(new Error("tick deferred: orchestrator busy or a user prompt is mid-submission"));
    }
    let sent: unknown;
    try {
      sent = pi.sendMessage(
        { customType: TICK_TYPE, content: message, display: true },
        { deliverAs: "followUp" }, // AUTOPILOT-125: NON-triggering — recorded, never claims the agent
      );
    } catch (e) {
      // Defensive only (the installed runtime does not throw): a host that
      // refuses synchronously gets the runner's re-hold path, not an escape
      // into pi core as an extension exception.
      return Promise.reject(e);
    }
    // TICK AUDIT TRAIL (AUTOPILOT-51): this is the ONE choke point both the
    // runner (createFrameworkRunner's `deliver`) AND the decision panel
    // (registerDecisionPanel's `deliver`, below) funnel every tick through —
    // logging here captures every tick actually HANDED OFF, from every
    // source, including the INFORMATIONAL ticks delivered while OFF
    // (AUTOPILOT-120: autopilotOn: false is their normal shape; see
    // src/framework/tick-class.ts and docs/AUTOPILOT-120-off-tick-scope.md).
    // A deferral is logged when it is finally delivered, not when it is held.
    logTickDelivery({ stateDir, message, autopilotOn: isAutopilotOn(stateDir, sessionId), sessionId });
    return sent && typeof (sent as Promise<void>).catch === "function" ? (sent as Promise<void>) : undefined;
  };

  const runner = createFrameworkRunner({
    stateDir,
    autopilot: ensureAutopilot(),
    backend,
    host: {
      interactive: () => interactive,
      loaded: () => orchestratorLoaded,
      busy: () => agentBusy || userPromptInFlight(), // a run mid-turn OR a user prompt mid-submission (AUTOPILOT-122)
      compacting: () => compacting,
    },
    deliver: deliverTick,
    deliverUserMessage: (message, options) => {
      try {
        const sent: unknown = pi.sendUserMessage(message, { deliverAs: "followUp", ...(options ?? {}) });
        if (sent && typeof (sent as Promise<void>).catch === "function") {
          void (sent as Promise<void>).catch(() => {});
        }
      } catch {
        throw new Error("deferred"); // the runner re-holds until the next flush
      }
    },
    emit: (e) => emitDomain(e as never),
    enabled: () => {
      // Scheduled-off BACKSTOP: a persisted deadline in the past flips the
      // session OFF here before any sweep runs. Covers the process-restart
      // case (the armed timer died with the old process) and any missed fire.
      fireScheduledOffIfDue(sessionId);
      return isAutopilotOn(stateDir, sessionId);
    },
    sweepIntervalMs: loadAutopilotConfig(stateDir).sweepIntervalMs,
  });

  // Scheduled shutdown ("off in <dur>") — parsing, persistence and the due
  // check are the framework's (config.autopilotCommand + scheduledOffDue);
  // this manager only owns the in-process timer whose firing replays today's
  // immediate-"off" side effects. Nothing else is host-specific.
  const schedules = createScheduleManager({
    onFire: (sid) => fireScheduledOffIfDue(sid),
  });
  /** The ONE fire path (timer AND enable-gate backstop): the framework's
   *  due-check flips OFF + clears the persisted deadline first, so the side
   *  effects below replay EXACTLY ONCE per schedule (a second caller — e.g.
   *  the backstop racing an earlier fire — sees no deadline and returns). */
  function fireScheduledOffIfDue(sid: string) {
    if (!sid || !scheduledOffDue(stateDir, sid)) return;
    informOrchestrator("off"); // best-effort internally (defers while busy)
    try {
      hostCtx?.ui?.notify(`Autopilot OFF (session ${sid.slice(0, 8)}) — scheduled shutdown fired; proactive auto-actions stopped (informational ticks still arrive)`, "info");
    } catch {
      // best-effort — the OFF state itself is already persisted
    }
  }
  /** Inform the ORCHESTRATOR (not the user) when the autopilot toggle changes —
   *  the harness's behavior flips entirely between the two modes and the agent
   *  must know which one it is running under. Best-effort: a busy-agent sync
   *  throw must not fail the command (followUp + guard, same as injection). */
  function informOrchestrator(mode: "on" | "off", opts?: { redundant?: boolean }) {
    // The message is the framework's (autopilotModeMessage) — this host only
    // DELIVERS it (followUp so a busy agent cannot fail the command). The ON
    // message carries the workspace facts pointer (state dir → config).
    // `redundant` (IDEMPOTENCY, threaded from autopilotCommand's `changed`)
    // — a repeated on/off must never re-inject the full activation payload
    // (the run-your-loop-now directive) into an orchestrator already in that
    // mode; the short no-op notice goes out instead.
    const msg = autopilotModeMessage(mode, { stateDir }, opts);
    if (agentBusy) {
      runner.deferUserMessage(msg); // the busy-not-streaming window throws — shared deferral
      return;
    }
    try {
      const sent: unknown = pi.sendUserMessage(msg, { deliverAs: "followUp" });
      if (sent && typeof (sent as Promise<void>).catch === "function") {
        void (sent as Promise<void>).catch(() => {});
      }
    } catch {
      // best-effort — the command itself already succeeded
    }
  }

  /** Publish structured domain events on the in-process bus (orch:*). */
  function emitDomain(events: Array<{ name: string; data: Record<string, unknown> }>) {
    for (const e of events) {
      try {
        pi.events.emit(e.name, { ...e.data, t: Date.now() });
      } catch {
        // silent
      }
    }
  }

  /** B26: block direct worker subagent spawns without worktree isolation.
   *  The RULE is the framework's (isUnisolatedWorkerSpawn — testable, shared);
   *  this host only wires its tool_call interception event to it. */
  pi.on("tool_call", (event) => {
    if (!interactive || !isAutopilotOn(stateDir, sessionId)) return;
    if (event.toolName !== "subagent") return;
    const args = (event.input ?? {}) as { agent?: string; worktree?: boolean; action?: string };
    if (!isUnisolatedWorkerSpawn(args, loadAutopilotConfig(stateDir).workerAgents)) return;
    return {
      block: true,
      reason:
        "Worker dispatch without worktree isolation is BLOCKED by the autopilot — use queue_dispatch instead " +
        "(it enforces worktree:true). A direct subagent worker spawn runs in the main checkout, dirties it, and breaks " +
        "worktree isolation for all parallel workers (B26: B20's strays blocked B22/B23 at launch).",
    };
  });

  // -- lifecycle -------------------------------------------------------------

  // AUTOPILOT-122 part 2 — THE USER-INPUT-LOSS WINDOW. pi core reports the
  // agent idle (`isStreaming` false) throughout the PRE-RUN awaits of
  // `AgentSession.prompt` (`agent-session.js:843` emitInput → `:915`
  // emitBeforeAgentStart → `:949` _runAgentPrompt), BEFORE the run is claimed
  // (`:773`). `agentBusy` (set from the run's `agent_start`) and `compacting`
  // both read false in that window, so a tick's old `triggerTurn` send used to
  // claim the agent first and the user's just-submitted prompt was refused,
  // recorded nowhere (pi-agent-core `agent.js:226-228`), with only
  // `interactive-mode.js:836-842`'s showError as its trace. The tick surface no
  // longer triggers a turn at all (AUTOPILOT-125), so the window can no longer
  // steal the prompt; these two events still mark it so the pre-check HOLDS the
  // tick out of the mid-run transcript and `runner.deferTick` re-delivers it at
  // the settle — HELD, never dropped.
  pi.on("input", () => { userPromptPendingAt = Date.now(); });
  pi.on("before_agent_start", () => { userPromptPendingAt = Date.now(); });

  pi.on("agent_start", () => {
    agentBusy = true;
    userPromptPendingAt = 0; // the run is claimed — the pre-run window is over
    compacting = false; // failsafe: a run beginning means compaction finished or was cancelled
  });
  pi.on("agent_settled", () => {
    agentBusy = false;
    userPromptPendingAt = 0; // a settle closes any pre-run window too
    runner.onSettled(); // the runner flushes the shared deferral + harness queue at the settle (the agent is idle)
  });
  // A tick delivered DURING auto-compaction aborts it ('Turn prefix
  // summarization failed: This operation was aborted') — the runtime can't
  // interleave a triggered turn with the summarization LLM call. agentBusy is
  // false in the compaction window (it sits between agent_end and the next
  // run), so gate on the compaction events explicitly.
  pi.on("session_before_compact", () => { compacting = true; });
  pi.on("session_compact", () => { compacting = false; });
  pi.on("session_start", (event, ctx) => {
    compacting = false;
    interactive = ctx?.mode === "tui";
    sessionId = (ctx?.sessionManager?.getSessionId?.() as string | undefined) ?? "";
    orchestratorLoaded = false;
    hostCtx = ctx ?? null;
    // A session ALWAYS starts with autopilot OFF — no auto-restore from a
    // persisted on (resumed sessions). The user explicitly runs /autopilot on
    // to engage the harness; that is the only activation path (it injects
    // /orchestrate + the run-your-loop directive).
    if (sessionId) writeSessionAutopilotState(stateDir, sessionId, "off");
    conflictWarnedSession = ""; // a new session gets its own single warning
    warnOnOrchestrateConflict(ctx ?? undefined);
    // STATE-DIR PROBE (AUTOPILOT-3) at session start: catch silent state-dir /
    // config split-brain BEFORE the session operates on a phantom/empty store
    // (a stale orchestrate.md projection, a missing config, a host-parity
    // divergence — all observed). Interactive TUI sessions only (children are
    // headless json runners with no UI) and only when THIS host has an
    // orchestrate.md projection to check against (an env-only harness session
    // has no orchestrator reading any state dir yet — its first activation
    // runs the full probe). ONE telemetry log + ONE notify, NEVER a throw or
    // a block. NOTE: this probe ships in the extension — a live session
    // predating the update may need a restart to pick it up.
    if (interactive && hostConfigDir()) {
      const probe = probeStateDir({ stateDir, hostConfigDir: hostConfigDir(), expectWorkspaceConfig: true });
      if (!probe.ok) {
        logStateDirProbe(stateDir, probe, { hook: "session_start" });
        try {
          ctx?.ui?.notify?.(`State-dir probe: ${probe.findings.join(" | ")}`, "error");
        } catch {
          // best-effort — the probe must never affect the session
        }
      }
    }
  });

  // -- periodic capacity sweep (deterministic safety net) --------------------
  // The timer is owned by the shared runner (identical machinery in both
  // hosts); pi gates it via enabled() = autopilot-on for this session.
  runner.start();
  pi.on("session_shutdown", () => {
    runner.stop();
    schedules.dispose(); // no leaked timers across sessions
  });

  // -- subagent lifecycle (in-process bus, emitted by pi-subagents) ----------

  pi.events.on("subagent:async-started", (payload: unknown) => {
    if (!interactive || !isAutopilotOn(stateDir, sessionId)) return;
    // The started event carries the workflow/async id under `id` (not `runId`).
    const p = payload as { id?: string; runId?: string; agent?: string } | null;
    const runId = p?.runId ?? p?.id;
    if (!runId) return;
    ensureAutopilot().handleAsyncStarted(runId, p?.agent);
  });

  pi.events.on("subagent:async-complete", (payload: unknown) => {
    if (!interactive) return; // the RUNNER gates the autopilot toggle (shared)
    maybeInjectOrchestrate();
    // The pi runtime's raw payload is flat (no per-child agent/output) — the
    // backend normalizes it from the run record (same shape the opencode
    // backend builds), so the shared attribution + verdict routing work
    // identically on both hosts. Without this, reviewer completions never
    // route (queue_review wedged on its first reviewer run).
    const ev = backend.buildCompletionEvent ? backend.buildCompletionEvent(payload) : ((payload ?? {}) as never);
    runner.onCompletion(ev);
  });

  // -- queue tools -----------------------------------------------------------

  // Shared queue ops — the six tools are host-agnostic implementations in
  // lib/orchestrator-autopilot/src/tools/queue-ops.ts (the opencode plugin
  // registers the SAME handlers). This ctx binds them to the pi host.
  const opsFor = (ctx?: { cwd?: string }): import("../tools/queue-ops.ts").QueueOpsCtx => ({
    stateDir,
    backend,
    storeOrNew,
    autopilot: () => ensureAutopilot(),
    cfg: () => loadAutopilotConfig(stateDir),
    emit: (e) => emitDomain(e as never),
    repoCheck,
    sessionCwd: ctx?.cwd,
  });

  pi.registerTool({
    name: "queue_list",
    label: "Queue list",
    description: CONTRACTS.queue_list,
    parameters: Type.Object({
      status: Type.Optional(Type.Union([Type.String({ description: "status filter (proposal|approved|active|ai-review|human-review|failed|done)" }), Type.Array(Type.String())])),
      since: Type.Optional(Type.String({ description: "ISO timestamp — only items with updatedAt >= since" })),
      sort: Type.Optional(Type.String({ description: "updatedAt|createdAt|key (default updatedAt desc)" })),
      limit: Type.Optional(Type.Number({ description: "max items (default 50)" })),
      includeNotes: Type.Optional(Type.Boolean({ description: "include scope/evidence/value/urgency/risk/notes" })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueList(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "queue_add",
    label: "Queue add",
    description: CONTRACTS.queue_add,
    parameters: Type.Object({
      key: Type.Optional(Type.String({ description: "unique key; OMIT to auto-allocate the next sequential id (with `series`, e.g. \"B\" → B-<max+1>; default series Q)" })),
      series: Type.Optional(Type.String({ description: "id series for auto-allocation when key is omitted (e.g. \"B\" → B-<n>)" })),
      status: Type.Optional(Type.String({ description: "proposal (default) | approved" })),
      title: Type.String(),
      scope: Type.Optional(Type.String({ description: "draft worker scope (free-form); " + CONTRACTS.approval_gate_param })),
      cwd: Type.Optional(Type.String({ description: "repo the worker runs in; " + CONTRACTS.approval_gate_param })),
      timeoutMs: Type.Optional(Type.Number({ description: "requested wall-clock budget (ms) — recorded on the item and passed to EVERY spawned run (worker dispatch, re-dispatch, review) instead of the runtime default" })),
      model: Type.Optional(Type.String({ description: CONTRACTS.model_param })),
      tier: Type.Optional(Type.String({ description: CONTRACTS.tier_param })),
      after: Type.Optional(Type.Union([Type.Array(Type.String()), Type.String()], { description: CONTRACTS.after_param })),
      evidence: Type.Optional(Type.String()),
      value: Type.Optional(Type.String()),
      urgency: Type.Optional(Type.String()),
      risk: Type.Optional(Type.String()),
      notes: Type.Optional(Type.String({ description: "free-form notes/description" })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueAdd(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "queue_update",
    label: "Queue update",
    description: CONTRACTS.queue_update,
    parameters: Type.Object({
      key: Type.String(),
      status: Type.Optional(Type.String({ description: "target status (see description for valid transitions); " + CONTRACTS.approval_gate_param })),
      blocker: Type.Optional(Type.String()),
      title: Type.Optional(Type.String()),
      scope: Type.Optional(Type.String()),
      cwd: Type.Optional(Type.String()),
      completedRun: Type.Optional(Type.String({ description: "COMPLETED-deliverable evidence for blocked → ai-review: the run id that finished this item's work (recorded as the item's runId). The transition is REFUSED without a completed run (runId/lastRunId) or a recorded branch/sha (runWorktree/landedEvidence/shippedAt) — a parked item must not become reviewable with no run that produced anything" })),
      timeoutMs: Type.Optional(Type.Number({ description: "requested wall-clock budget (ms) recorded on the item; every dispatch/review passes it to the spawned run" })),
      model: Type.Optional(Type.String({ description: CONTRACTS.model_param })),
      tier: Type.Optional(Type.String({ description: CONTRACTS.tier_param })),
      dispatchClass: Type.Optional(Type.String({ description: "worker (default) | finisher — finisher declares the run writes OUTSIDE its worktree, into the declared cwd's checkout (merge finisher)" })),
      finisherSource: Type.Optional(Type.String({ description: "for a finisher item: WHAT it lands (branch/tag/sha in the cwd). Required for landed evidence — without it a finisher's runtime failure verdict stands as usual" })),
      overrideReason: Type.Optional(Type.String({ description: "RECORD an override of the run's failure verdict on the item (append-only overrides[]) instead of hand-writing it into notes — say WHY the failure is not believed" })),
      after: Type.Optional(Type.Union([Type.Array(Type.String()), Type.String()], { description: CONTRACTS.after_param })),
      evidence: Type.Optional(Type.String()),
      value: Type.Optional(Type.String()),
      urgency: Type.Optional(Type.String()),
      risk: Type.Optional(Type.String()),
      notes: Type.Optional(Type.String()),
      findings: Type.Optional(Type.String({ description: "REQUIRED to re-dispatch a human-review item (status:'active'): what needs fixing. Sets the structured handoff the harness's human re-dispatch sweep reads to spawn the redo — do not also call queue_dispatch" })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueUpdate(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "queue_dispatch",
    label: "Queue dispatch",
    description: CONTRACTS.queue_dispatch,
    parameters: Type.Object({
      key: Type.String({ description: "queue key of an approved item" }),
      task: Type.String({ description: "the scoped worker prompt (self-contained; KEY: <key> as first line is recommended)" }),
      cwd: Type.Optional(Type.String({ description: "repo the worker operates on (worktree isolation runs THERE). REQUIRED when the session cwd is not the target repo (e.g. dispatching from a parent dir into the repo the worker must touch)." })),
      timeoutMs: Type.Optional(Type.Number()),
      model: Type.Optional(Type.String({ description: CONTRACTS.model_param })),
      tier: Type.Optional(Type.String({ description: CONTRACTS.tier_param })),
      dispatchClass: Type.Optional(Type.String({ description: "worker (default) | finisher — use 'finisher' for a MERGE FINISHER that writes into the cwd's checkout instead of its worktree; success is then judged on the declared source landing there, not on worktree edits" })),
      finisherSource: Type.Optional(Type.String({ description: "the branch/tag/sha this finisher must land in cwd (e.g. pi-parallel-<runid>-0). REQUIRED for the landed-evidence path: without it a runtime 'no edits in the worktree' verdict fails the run as usual" })),
      overrideCapacity: Type.Optional(Type.Boolean({ description: "dispatch even though the fleet is at/above maxSlots. Default false = the dispatch is REFUSED at capacity; true is a deliberate over-subscription (recorded in the result) — use it when you know better than the cap, not routinely" })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueDispatch(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "queue_review",
    label: "Queue review",
    description: CONTRACTS.queue_review,
    parameters: Type.Object({
      key: Type.String({ description: "queue key of an `ai-review` item" }),
      task: Type.Optional(Type.String({ description: "optional reviewer prompt; the verdict contract is injected if omitted" })),
      timeoutMs: Type.Optional(Type.Number()),
      model: Type.Optional(Type.String({ description: CONTRACTS.model_param })),
      tier: Type.Optional(Type.String({ description: CONTRACTS.tier_param })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueReview(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "queue_steer",
    label: "Queue steer",
    description: CONTRACTS.queue_steer,
    parameters: Type.Object({
      key: Type.String({ description: "queue key of the running worker/reviewer" }),
      message: Type.String({ description: "the steering instruction (interrupt-and-deliver)" }),
      mode: Type.Optional(Type.String({ description: "steer (interrupt, default) | follow_up (queued until idle)" })),
      ackTimeoutMs: Type.Optional(Type.Number({ description: "ack poll deadline in ms (default 4000)" })),
    }),
    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
      const r = await queueSteer(opsFor(ctx), params as never);
      return { content: [{ type: "text", text: r.text }], details: r.details };
    },
  });

  pi.registerTool({
    name: "flag_for_review",
    label: "Flag for review",
    description: CONTRACTS.flag_for_review,
    parameters: Type.Object({
      summary: Type.String({ description: "What was done (one line)" }),
      risk: Type.Union([Type.Literal("low"), Type.Literal("medium"), Type.Literal("high")], { description: "low = mistake has no real-world impact; medium = contained but visible; high = affects production/data/auth/pipelines" }),
      blast_radius: Type.String({ description: "What breaks if this is wrong — be specific" }),
      review_targets: Type.Array(Type.String(), { description: "WHERE to review: file paths, path:line ranges, commit SHAs, MR/PR links — the human reviews these" }),
      self_reviewed: Type.Boolean({ description: "true if the pre-commit reviewer or a reviewer subagent passed; false if no automated review ran" }),
      review_method: Type.Union([Type.Literal("commit-hook"), Type.Literal("reviewer-subagent"), Type.Literal("none")], { description: "How the self-review happened" }),
      action_needed: Type.Optional(Type.String({ description: "What the user should DO after reviewing (merge X, approve Y, decide A/B)" })),
      residual_risks: Type.Optional(Type.String({ description: "What was NOT verified / could change" })),
      queue_key: Type.Optional(Type.String({ description: "The queue item this flag came from (when orchestrated)" })),
    }),
    async execute(_toolCallId, params: { summary: string; risk: "low" | "medium" | "high"; blast_radius: string; review_targets: string[]; self_reviewed: boolean; review_method: "commit-hook" | "reviewer-subagent" | "none"; action_needed?: string; residual_risks?: string; queue_key?: string }) {
      const { deliveryNote } = flagForReview(params, { logPath: join(stateDir, "reviews.jsonl") });
      return { content: [{ type: "text", text: `Flagged for review. Risk: ${params.risk}. ${deliveryNote}` }], details: {} };
    },
  });

  // -- /autopilot command ----------------------------------------------------

  pi.registerCommand("autopilot", {
    description: "Orchestrator autopilot: on | off [in <duration>] | status | capacity <n>",
    handler: async (args, ctx) => {
      const [cmd, ...rest] = (args ?? "").trim().split(/\s+/);
      const val = rest.length ? rest.join(" ") : undefined; // full remainder — "off in 30m" must reach the shared parser intact
      hostCtx = ctx ?? null;
      try {
        // PREREQUISITE PROBE (fail-closed, BEFORE the toggle writes state):
        // the harness spawns through the subagent backend — without that
        // runtime, dispatch/review would fail downstream in confusing ways.
        // Say so NOW instead of half-activating.
        if (cmd === "on") {
          const fleet = await backend.fleetStatus();
          if (fleet === null) {
            ctx.ui.notify("Autopilot NOT activated: the subagent backend is unreachable (fleet status timed out). Install/enable the pi-subagents extension and retry — the queue tools work without it, but ticks/dispatch do not.", "error");
            return;
          }
        }
        // ONE shared toggle implementation (config.autopilotCommand) — this
        // host keeps only its SIDE EFFECTS: the /orchestrate injection, the
        // activation nudge, the orchestrator mode notice, and the notify.
        const r = autopilotCommand(cmd, val, { stateDir, sessionId });
        if (!r.ok) {
          ctx.ui.notify(r.message, "error");
          return;
        }
        if (r.scheduledOffAt !== undefined && sessionId) {
          // SCHEDULED off: still ON until the deadline — arm the timer (the
          // manager replaces any previous one) + confirm with the absolute time.
          schedules.schedule(sessionId, r.scheduledOffAt);
          ctx.ui.notify(r.message, "info");
        } else if (r.mode === "on") {
          schedules.cancel(sessionId); // explicit toggle cancels a pending schedule
          ensureMigrated(stateDir); // import legacy state.md into queue.json once
          // STATE-DIR PROBE (AUTOPILOT-3) at ACTIVATION: the ON message points
          // the orchestrator at this state dir for workspace facts — verify it
          // is REAL (dir + queue.json + promised autopilot.config.json + host
          // parity vs the profile's autopilot.host.json) BEFORE relying on it.
          // Fail-open: ONE telemetry log + ONE notify; activation proceeds even
          // with findings (a warning costs a notify; blocking would strand the
          // user mid-activation). Never throws.
          const probe = probeStateDir({ stateDir, hostConfigDir: hostConfigDir(), expectWorkspaceConfig: true });
          if (!probe.ok) {
            logStateDirProbe(stateDir, probe, { hook: "autopilot-on" });
            try {
              ctx.ui.notify(`State-dir probe: ${probe.findings.join(" | ")}`, "error");
            } catch {
              // best-effort — the activation itself is already committed
            }
          }
          // STARTUP DOES NOT NUDGE: /orchestrate loads the mode; the ON
          // message (autopilotModeMessage) carries the run-your-loop now
          // directive. The first tick can never race the injection because
          // no tick is fired here.
          // EXCEPT auto-actions: the activation sweep refreshes the fleet and
          // lets the harness fill free slots IMMEDIATELY (approved items never
          // wait for the first settle/timer — AUTOPILOT-9). Its ticks are
          // gated on orchestratorLoaded, so pre-injection it can only dispatch.
          runner.activate();
          maybeInjectOrchestrate();
          informOrchestrator("on", { redundant: !r.changed });
          ctx.ui.notify(`Autopilot ON (session ${sessionId.slice(0, 8)}) — orchestrator mode loaded, capacity ticks enabled`, "info");
        } else if (r.mode === "off") {
          schedules.cancel(sessionId); // explicit toggle cancels a pending schedule
          informOrchestrator("off", { redundant: !r.changed });
          ctx.ui.notify(`Autopilot OFF (session ${sessionId.slice(0, 8)}) — proactive auto-actions stopped; queue tools + informational ticks remain available`, "info");
        } else if (cmd === "status") {
          const st = ensureAutopilot().status(backend); // the SAME derivation the tick/tools use (AUTOPILOT-86)
          const counts = queueLengths(storeOrNew());
          // the helper's mode+capacity line + this host's runtime detail
          ctx.ui.notify(`${r.message} running: ${st.running}, queue: ${JSON.stringify(counts)}`, "info");
        } else if (cmd === "capacity") {
          autopilot = null; // recreate with the new capacity
          ctx.ui.notify(r.message, "info");
        }
      } catch (e) {
        const err = e instanceof Error ? e : new Error(String(e));
        // ALWAYS log the failure (even when autopilot is off) — the notify
        // promises "see autopilot.jsonl"; without this line the promise is a
        // lie and the failure is undebuggable.
        try {
          appendTelemetry(stateDir, JSON.stringify({
            t: new Date().toISOString(),
            type: "command-error",
            command: args,
            error: err.message,
            stack: err.stack?.split("\n").slice(0, 6).join("\n"),
          }));
        } catch {
          // jsonl write failing is not worth a second notification
        }
        ctx.ui.notify("Autopilot command failed (see autopilot.jsonl)", "error");
      }
    },
  });

  // -- /orchestrate command (the PACKAGE owns the procedure) -----------------
  // AUTOPILOT-32 D2: the procedure ships with the plugin and is registered
  // HERE, so there is nothing to project, sync or drift-check. EMPIRICAL
  // PRECEDENCE (see framework/orchestrate-command.ts): on pi this extension
  // command WINS over a same-named prompt template, so a stale projection
  // cannot shadow it — the D4 warning names it anyway.
  pi.registerCommand("orchestrate", {
    description: "Enter orchestrator mode — the packaged procedure (queue, dispatch, review, ship)",
    handler: async (args, ctx) => {
      const procedure = orchestrateProcedure();
      if (!procedure) {
        ctx.ui.notify("Orchestrator procedure unavailable — the packaged prompts/orchestrate.md could not be read. Reinstall/repair the orchestrator-autopilot package.", "error");
        return;
      }
      hostCtx = ctx ?? hostCtx;
      orchestratorLoaded = true; // the mode is loaded by THIS turn — ticks may flow
      const trimmed = (args ?? "").trim();
      const message = trimmed ? `${procedure}\n\nInvocation: ${trimmed}` : procedure;
      try {
        const sent: unknown = pi.sendUserMessage(message, { deliverAs: "followUp" });
        if (sent && typeof (sent as Promise<void>).catch === "function") {
          void (sent as Promise<void>).catch(() => { orchestratorLoaded = false; });
        }
      } catch {
        runner.deferUserMessage(message); // busy-not-streaming — flushes at the settle
      }
    },
  });

  // -- decision panel (/orchestrate-panel) -----------------------------------
  // The user-facing decision inbox: proposals + human-review views over queue
  // state (tab toggles). Pure UI over the shared panels core; fail-closed —
  // if the TUI surface is unavailable the command notifies instead of
  // throwing, and queue_list remains the fallback read.
  try {
    // AUTOPILOT-67: isAutopilotOn(...) gates the panel's tick delivery the
    // same way runner.ts's own enabled() does — the decision itself (a
    // manual human action) still always applies; only the tick riding the
    // orchestrator-autopilot channel is suppressed while OFF.
    registerDecisionPanel(pi, { stateDir: () => stateDir, deliver: deliverTick, enabled: () => isAutopilotOn(stateDir, sessionId) });
  } catch {
    // a broken panel must never affect pi — the command simply won't exist
  }
}