// runner.ts — the framework RUNNER: host-agnostic TRIGGER ROUTING + tick
// delivery. Both hosts (pi extension + opencode plugin) wire their tool events
// to these methods; everything about WHAT a trigger does (fleet reconcile →
// sweep → tick → reviewTick fallback → gate → deliver) lives here ONCE.
//
// The hosts keep only the WIRING:
//   - their event sources (pi: async-complete/agent_settled/timer//autopilot;
//     opencode: backend process-exit/session.idle/timer)
//   - their gate-state semantics (interactive/loaded/busy/compacting)
//   - their delivery mechanism (pi: sendMessage; opencode: promptAsync)

import type { SubagentBackend } from "../backends/types.ts";
import { join } from "node:path";
import { storeToSnapshot, type Autopilot } from "../core.ts";
import type { CompletionEvent, Tick } from "../types.ts";
import { loadAutopilotConfig, appendTelemetry, underSpecifiedProposals, serializedWithoutDependency } from "../config.ts";
import { loadStore, mutateStore, updateItem, itemByRunId } from "../queue-store.ts";
import { formatDurationMs } from "../duration.ts";
import { flagForReview } from "./flag-review.ts";
import { isRisk, type Risk } from "./item-contract.ts";
import { reviewNotesResidualRisks } from "../review-notes.ts";
import { createTickRouter, premiseForTick, premiseCoalesceKey, type TickDeliveryMeta, type TickHostState, type TickPremise, type TickSendOptions, type TickStoreView } from "./tick-router.ts";
import { tickReasonFromMessage } from "./tick-telemetry.ts";
import { deliverableWhileOff } from "./tick-class.ts";
import { autoDispatchEligible, autoRedispatch, autoReview, dispatchPendingHumanRedispatch, dispatchPendingSpecPass, effectiveOccupiedSlots } from "./auto-dispatch.ts";
import { autoRecoverFails } from "./auto-recovery.ts";
import { decisionTick } from "./panels.ts";
import { checkpointActiveItems, preserveActiveItems, prunePreservedRefs, preserveRunWorktree, recordActiveWorktrees } from "./worktree-preservation.ts";
import { buildReviewTargets, noReviewTargetsFoundLine } from "./review-targets.ts";
import { checkMainWrites } from "./main-write-guard.ts";
import { configChangeTick, detectConfigChanges, recordConfigSnapshot } from "./config-watch.ts";
import { runShippingPass } from "./shipping.ts";
import { runStateEvidence } from "./run-liveness.ts";

export interface RunnerOptions {
  stateDir: string;
  autopilot: Autopilot;
  backend: SubagentBackend;
  /** The tick delivery gate state (what interactive/loaded/busy/compacting
   *  mean in this host). */
  host: TickHostState;
  /** The host's delivery mechanism (pi sendMessage / opencode promptAsync).
   *  May return a promise: a REJECTION means the runtime accepted the call but
   *  the send ultimately failed (pi: the settled-vs-teardown race where the
   *  triggered prompt() throws "already processing" AFTER sendMessage resolved).
   *  Nothing was persisted on that failure path, so the framework re-holds the
   *  message + retries — duplicate-safe by construction. */
  deliver: (message: string) => void | Promise<void>;
  /** Optional user-message delivery for deferred DIRECT sends (the pi host's
   *  /orchestrate injection + toggle messages; opencode has none). */
  deliverUserMessage?: (message: string, options?: Record<string, unknown>) => void;
  /** Host gate: should triggers run at all right now (pi: autopilot-on for
   *  this session; opencode: always). */
  enabled?: () => boolean;
  /** Domain-event sink (orch:item-completed, orch:verdict, ...). */
  emit?: (events: Array<{ name: string; data?: Record<string, unknown> }>) => void;
  cooldownMs?: number; // min gap between delivered ticks (default 1500)
  /** Periodic sweep interval; 0 disables the timer. */
  sweepIntervalMs: number;
  /** Backoff before re-delivering a tick whose send rejected asynchronously
   *  (default 1000ms; tests inject a tiny value). */
  deliveryRetryDelayMs?: number;
  /** Auto-dispatch + auto re-dispatch (default true; AUTOPILOT_AUTO_DISPATCH=0
   *  disables — a real, working opt-out). The framework fills free slots + re-dispatches FAIL items
   *  itself — the orchestrator keeps the judgment (intake, approval,
   *  high-risk checkpoints). */
  autoDispatch?: boolean;
  /** AUTOMATIC RECOVERY of failed items (AUTO-RECOVER-FAILS). Default on with
   *  auto-dispatch; the per-cause policy + bounds live in auto-recovery.ts.
   *  backoffMs/maxRecoveries/degraded window are injectable for tests and for
   *  a slow-provider deployment. */
  recovery?: {
    enabled?: boolean;
    backoffMs?: number;
    maxRecoveries?: number;
    spawnFailureThreshold?: number;
    degradedCooldownMs?: number;
  };
}

export interface FrameworkRunner {
  /** A run completed (worker done / reviewer done / any session subagent).
   *  pi: subagent:async-complete; opencode: backend process-exit. */
  onCompletion(ev: CompletionEvent): void;
  /** The orchestrator's own turn settled (queue may have changed).
   *  pi: agent_settled; opencode: session.idle. Hash-gated sweep. */
  onSettled(): void;
  /** Periodic re-nudge (bypasses the hash — re-ticks a persistent gap). */
  onTimer(): void;
  /** Explicit activation sweep (/autopilot on) — the harness fills free slots
   *  immediately instead of waiting for the first settle/timer (AUTOPILOT-9). */
  activate(): void;
  start(): void;
  stop(): void;
  /** Deferred DIRECT user-message send (the pi host's /orchestrate injection +
   *  toggle/mode messages): held while the agent is busy-not-streaming and
   *  flushed at the next settle. Absent for hosts without a direct send. */
  deferUserMessage(message: string, options?: Record<string, unknown>): void;
  /** Hold a TICK in the SAME shared deferral (KEY: AUTOPILOT-122). The pi
   *  host's ONE delivery surface (`deliverTick`) uses this when its own gate
   *  sees that it must not inject right now (a mid-turn/compacting orchestrator,
   *  or a user prompt inside pi core's pre-run window): the tick is re-delivered
   *  at the next settle flush instead of being swallowed by a caller that only
   *  best-effort catches the deferral signal (the decision panel). Never a drop
   *  — the flush re-applies the live host gate + the OFF class rule. The held
   *  entry carries no premise (it is re-checked as `always`): the host's one-
   *  line ticks (panel decisions, narration) report an event that happened, so
   *  delivering them late is the point — a staleness drop would lose it. */
  deferTick(message: string): void;
}

export function createFrameworkRunner(opts: RunnerOptions): FrameworkRunner {
  const autopilot = opts.autopilot;
  // ASYNC delivery failures are indistinguishable from success at the call
  // site (the sync verdict says "handed off"), so the wrapper watches the
  // returned thenable: a rejection re-defers the message for a bounded retry.
  // The immediate re-flush is deliberately SKIPPED here — the same race would
  // reject again mid-settle; a short backoff (plus the natural settle/timer
  // flushes) gives the runtime's teardown time to finish.
  const RETRY_DELAY_MS = opts.deliveryRetryDelayMs ?? 1000;
  const RETRY_MAX_ATTEMPTS = 5;
  let retryTimer: ReturnType<typeof setTimeout> | null = null;
  // Attempt counts live HERE, not on the deferred entries: the flush paths
  // remove an entry the moment the host ACCEPTS the send (sync verdict), so a
  // count stored on the entry would reset every cycle and never hit the cap.
  const retryAttempts = new Map<string, number>();
  const deliverWatchingAsync = (message: string, meta?: TickDeliveryMeta): void => {
    try {
      const r = opts.deliver(message) as unknown;
      if (r && typeof (r as Promise<unknown>).then === "function") {
        void (r as Promise<void>).then(
          () => retryAttempts.delete(message), // delivered (or safely queued) — reset
          () => {
            // ASYNC rejection — the runtime accepted the call but the turn
            // trigger failed afterwards (pi: prompt() threw "already
            // processing" post-resolve). Nothing was persisted on that path,
            // so re-hold + retry after a short backoff; bounded attempts.
            // The router's sync verdict said "delivered", so undo the
            // review-state dedupe it committed for THIS send: otherwise the
            // flush retry below is dropped as a phantom duplicate and the nudge
            // never reaches the reader (AUTOPILOT-87).
            router.invalidateReviewState(message);
            const n = (retryAttempts.get(message) ?? 0) + 1;
            if (n > RETRY_MAX_ATTEMPTS) {
              retryAttempts.delete(message);
              return; // permanently rejecting runtime — nudges re-fire anyway
            }
            retryAttempts.set(message, n);
            if (!deferred.some((d) => d.kind === "tick" && d.message === message)) {
              deferred.push({ message, kind: "tick", premise: meta?.premise, generatedAt: meta?.generatedAt });
            }
            // NO immediate flush — the same race would reject again mid-settle.
            if (!retryTimer) {
              retryTimer = setTimeout(() => {
                retryTimer = null;
                flushDeferred();
              }, RETRY_DELAY_MS);
            }
          },
        );
      }
    } catch (e) {
      throw e; // sync throw still propagates — the router maps it to "deferred"
    }
  };
  // DELIVERY-TIME PREMISE (AUTOPILOT-87): the live store, read at the moment
  // the gate is about to deliver a tick that declares a premise. The gate
  // drops the tick when the premise no longer holds — see tick-router.ts.
  // The store dir is the ENGINE's (always the real one; a RunnerOptions call
  // that omitted it still validates against what the engine acts on).
  const premiseStateDir = opts.stateDir || autopilot.stateDir;
  // THE GATE'S LAST-COMPUTED OCCUPANCY (KEY: AUTOPILOT-96). The premise
  // re-check and the delivery-time tick refresh both run OUTSIDE a sweep (at a
  // send/settle), where no fresh fleet RPC is available synchronously — so they
  // project the occupancy the gate most recently acted on, instead of the
  // store-live-only derivation that let a held tick claim a free slot the gate
  // did not have. Set once per sweep right after effectiveOccupiedSlots runs.
  let lastGateOccupied: number | null = null;
  const storeView: TickStoreView = {
    snapshot: () => {
      const store = loadStore(premiseStateDir);
      // NO STORE -> NO VIEW. A premise that cannot be re-checked must be handled
      // as UNVERIFIABLE (the router drops the non-`always` tick with
      // `unverifiable-no-store`), NOT as an empty store — an empty snapshot
      // would read every named item as `item-gone` and misreport the why.
      if (!store) return undefined;
      // The SAME store->facts derivation the engine and the tools use
      // (AUTOPILOT-86) for `ready`, plus the GATE's occupancy (KEY:
      // AUTOPILOT-96) for the free-slot count: the dispatch premise's "slot
      // free" must be the SAME number auto-dispatch is gated on, never the
      // store-live-only view that could claim a slot the gate does not have.
      const snap = storeToSnapshot(store, Date.now(), { backend: opts.backend, zombieGraceMinutes: cfg().zombieGraceMinutes });
      const status: Record<string, string | undefined> = {};
      const runId: Record<string, string | null> = {};
      const aiReview: string[] = [];
      const humanReview: string[] = [];
      for (const [key, it] of Object.entries(store.items)) {
        status[key] = it.status;
        runId[key] = it.runId ?? null;
        if (it.status === "ai-review") aiReview.push(key);
        if (it.status === "human-review") humanReview.push(key);
      }
      return { status, runId, aiReview, humanReview, ready: snap.ready, slotsFree: Math.max(0, cfg().maxSlots - (lastGateOccupied ?? snap.occupied)) };
    },
  };
  const enabled = opts.enabled ?? (() => true);
  const routerInner = createTickRouter(opts.host, deliverWatchingAsync, {
    cooldownMs: opts.cooldownMs,
    store: storeView,
    stateDir: premiseStateDir,
  });
  // AUTOPILOT-120 OFF-SCOPE CLASS GATE — the ONE choke point every runner tick
  // passes: `sendTick`/`sendTickText`/`harnessTick`/the deferred flush all
  // funnel through `router.send`, so the class split is enforced HERE once, not
  // as scattered conditions at each call site. With autopilot OFF an
  // AUTOMATIC-ACTION tick (dispatch/intake/recover) is dropped; an
  // INFORMATIONAL tick (review/failure/budget/decision/harness/ship/
  // main-write/config) still delivers. Reading the class off the MESSAGE — not
  // a passed flag — is what makes a HELD tick re-checked at flush time too:
  // this is the fix for the leak that let a deferred intake/dispatch tick
  // surface after the switch (onSettled/onTimer flush without their own gate).
  const router = {
    send(message: string, o?: TickSendOptions): "delivered" | "dropped" | "deferred" {
      // An UNDETERMINABLE gate (a host whose enabled() throws — AUTOPILOT-24's
      // robustness case) cannot LICENCE an automatic-action tick, but it must
      // not swallow an informational one either: silence is the worse failure.
      let on = false;
      try {
        on = enabled();
      } catch {
        on = false;
      }
      if (!on && !deliverableWhileOff(message)) return "dropped";
      return routerInner.send(message, o);
    },
    invalidateReviewState(message?: string): void {
      routerInner.invalidateReviewState(message);
    },
  };
  // SHARED deferral: messages the runtime rejected (busy-not-streaming —
  // the host's deliver threw) are held here + flushed at the host's
  // settle/idle event (the agent is idle then). This is framework logic;
  // the hosts only wire their idle event to onSettled + supply delivery.
  // DELIVERY-TIME FACTS (AUTOPILOT-6): a deferred dispatch/intake tick keeps
  // its engine Tick and is RE-DERIVED against the live store at each flush —
  // see refreshDeferredFacts below.
  interface DeferredEntry {
    message: string;
    kind: "tick" | "user";
    /** The engine tick a deferred tick entry was generated from (dispatch /
     *  intake refreshes recompute its FLEET/QUEUE facts at delivery; harness
     *  and review ticks have no such facts). */
    tick?: Tick;
    /** Delivery-time refresh decided the nudge no longer applies — drop. */
    drop?: boolean;
    options?: Record<string, unknown>;
    /** AUTOPILOT-87: the premise re-checked at delivery + the composition time
     *  (for the `ageMs` in the drop telemetry). */
    premise?: TickPremise;
    generatedAt?: number;
  }
  const deferred: DeferredEntry[] = [];
  const queueDeferred = (message: string, kind: "tick" | "user", tick?: Tick, options?: Record<string, unknown>, meta?: TickDeliveryMeta): void => {
    deferred.push({ message, kind, ...(tick ? { tick } : {}), ...(options ? { options } : {}), premise: meta?.premise, generatedAt: meta?.generatedAt });
    flushDeferred();
  };
  // Recompute the FLEET/QUEUE facts of every held dispatch/intake tick from
  // the LIVE store before a flush delivers it. Between generation and the
  // busy-deferral flush the store can move (manual dispatch mid-window, an
  // auto-dispatch, a completion flip): "QUEUE: N ready (…)" must never list
  // items that are already active/running, and "FLEET: X free" must never
  // forget runs that started meanwhile. refreshTickFacts returns the CURRENT
  // truthful nudge or null (= no nudge applies anymore → drop the stale text
  // instead of contradicting the live state).
  const refreshDeferredFacts = (): void => {
    for (const d of deferred) {
      if (d.kind !== "tick" || !d.tick || d.drop) continue;
      if (d.tick.reason !== "dispatch" && d.tick.reason !== "intake") continue;
      // PROJECT THE GATE'S OCCUPANCY (KEY: AUTOPILOT-96): pass the number the
      // gate last acted on so the refreshed "X free" can never exceed what the
      // gate would dispatch against. Before this the refresh omitted the fleet
      // term entirely and re-derived from the store alone — the exact drift
      // that let a held tick read "3/6, 3 free" while the gate sat at cap.
      const occupancy = lastGateOccupied !== null ? { occupied: lastGateOccupied } : undefined;
      const fresh = autopilot.refreshTickFacts(d.tick, occupancy, undefined, opts.backend);
      if (!fresh) {
        d.drop = true; // generation-time facts are STALE by delivery — never present them
        continue;
      }
      d.message = fresh.message;
      d.tick = fresh;
      // A refresh can change the tick's REASON (a dispatch nudge becomes an
      // intake nudge once the slots fill) — recompute the premise so the gate
      // validates the CURRENT claim, not the generation-time one.
      d.premise = premiseForTick(fresh);
    }
  };
  /** AUTOPILOT-87 COALESCE: multiple PENDING ticks for the same key+reason are
   *  one event — keep only the NEWEST before delivery. Capacity nudges
   *  (dispatch/intake) collapse the same way; keyless informational ticks are
   *  left untouched (their own content is the identity). */
  const coalesceDeferred = (): void => {
    const byId = new Map<string, number>();
    const keep: DeferredEntry[] = [];
    for (const d of deferred) {
      let id: string | null = null;
      if (d.kind === "tick") {
        const reason = tickReasonFromMessage(d.message);
        const key = premiseCoalesceKey(d.premise);
        if (key !== null) id = `${reason}|${key}`;
        else if (reason === "dispatch" || reason === "intake") id = reason;
      }
      if (id === null) {
        keep.push(d);
        continue;
      }
      const prev = byId.get(id);
      if (prev === undefined) {
        byId.set(id, keep.length);
        keep.push(d);
      } else {
        keep[prev] = d; // newest wins, position preserved
      }
    }
    deferred.length = 0;
    deferred.push(...keep);
  };
  const flushDeferred = (): void => {
    if (!deferred.length) return;
    coalesceDeferred();
    refreshDeferredFacts();
    // De-dupe identical ticks within one flush: a busy window that saw both a
    // nudge and a state-changing sweep can accumulate two dispatch ticks that
    // refresh to the SAME current message — deliver it once.
    const deliveredThisFlush = new Set<string>();
    const hold: typeof deferred = [];
    for (const d of deferred) {
      if (d.drop) continue;
      if (d.kind === "user") {
        if (!opts.deliverUserMessage) continue; // no user-delivery on this host
        try {
          opts.deliverUserMessage(d.message, d.options);
        } catch {
          hold.push(d); // still busy — re-flush at the next settle
        }
        continue;
      }
      if (deliveredThisFlush.has(d.message)) continue; // already delivered this flush
      const r = router.send(d.message, { bypassCooldown: true, premise: d.premise, generatedAt: d.generatedAt });
      if (r === "deferred") hold.push(d); // the runtime still rejected it
      else if (r === "delivered") deliveredThisFlush.add(d.message);
      // "dropped" (interactive/loaded) — permanent, drop; "delivered" — done
    }
    deferred.length = 0;
    deferred.push(...hold);
  };
  /** Host-facing: defer a DIRECT user-message send (the /orchestrate injection,
   *  the toggle message) — delivered via deliverUserMessage at the settle. */
  const deferUserMessage = (message: string, options?: Record<string, unknown>): void => {
    queueDeferred(message, "user", undefined, options);
  };
  /** Host-facing (KEY: AUTOPILOT-122): hold a tick the host's delivery surface
   *  could not inject right now. It rides the SAME deferral + flush path as
   *  the runner's own ticks, so it is re-delivered at the next settle (and
   *  re-checked against the live host gate + the OFF class rule then). */
  const deferTick = (message: string): void => {
    queueDeferred(message, "tick");
  };
  const sendTick = (t: Tick | null | undefined): void => {
    if (!t?.message) return;
    const premise = premiseForTick(t);
    const generatedAt = typeof t.facts.generatedAt === "number" ? t.facts.generatedAt : Date.now();
    const r = router.send(t.message, { premise, generatedAt });
    if (r === "deferred") queueDeferred(t.message, "tick", t, undefined, { premise, generatedAt }); // hold the ENGINE tick — its facts refresh at delivery
  };
  /** Framework-crafted one-line ticks (recovery announcements) through the
   *  SAME gate: deferred → held for the settle flush; never lost. Returns the
   *  router verdict so a caller with once-only state (the config watcher) can
   *  tell "nobody was listening" (dropped) from "held for the settle"
   *  (deferred) and re-announce instead of silently consuming the change. */
  const sendTickText = (message: string, opts?: TickSendOptions): "delivered" | "dropped" | "deferred" => {
    const generatedAt = opts?.generatedAt ?? Date.now();
    const r = router.send(message, { bypassCooldown: true, premise: opts?.premise, generatedAt });
    if (r === "deferred") queueDeferred(message, "tick", undefined, undefined, { premise: opts?.premise, generatedAt });
    return r;
  };

  // The auto-actions opt-out: env (AUTOPILOT_AUTO_DISPATCH=0) or the option.
  const autoDispatchOn = opts.autoDispatch ?? process.env.AUTOPILOT_AUTO_DISPATCH !== "0";
  const cfg = () => loadAutopilotConfig(opts.stateDir);
  // AUTOMATIC RECOVERY (AUTO-RECOVER-FAILS): the failed lane's deterministic
  // policy. Gated with the other auto-actions; the engine (auto-recovery.ts)
  // owns the per-cause plan + bounds, this runner owns DELIVERY (the
  // `[orch-tick: recover]` announcements) and the short backoff TIMER.
  const recoveryOn = autoDispatchOn && (opts.recovery?.enabled ?? true);
  let recoveryTimer: ReturnType<typeof setTimeout> | null = null;
  let recoveryTimerAt = 0;
  const scheduleRecoveryAt = (at: number | null): void => {
    if (at === null || !recoveryOn) return;
    // An earlier (or equal) pass is already armed — leave it.
    if (recoveryTimer && recoveryTimerAt <= at) return;
    if (recoveryTimer) clearTimeout(recoveryTimer);
    recoveryTimerAt = at;
    recoveryTimer = setTimeout(() => {
      recoveryTimer = null;
      recoveryTimerAt = 0;
      void runRecoveryPass();
    }, Math.max(0, at - Date.now()));
  };
  const runRecoveryPass = async (totalActive?: number): Promise<void> => {
    // The recovery timer fires outside the sweep's own gate — honour BOTH the
    // structural opt-out and the live autopilot toggle there too (a pending
    // backoff scheduled before an OFF must not spawn).
    if (!recoveryOn || !enabled()) return;
    let outcome;
    try {
      outcome = await autoRecoverFails(opts.stateDir, opts.backend, {
        backoffMs: opts.recovery?.backoffMs,
        maxRecoveries: opts.recovery?.maxRecoveries,
        spawnFailureThreshold: opts.recovery?.spawnFailureThreshold,
        degradedCooldownMs: opts.recovery?.degradedCooldownMs,
        maxSlots: cfg().maxSlots,
        totalActive,
      });
    } catch {
      return; // recovery must never break a sweep
    }
    for (const r of outcome.recovered) {
      const budget = r.budgetMs ? `, budget ${formatDurationMs(r.budgetMs)}` : "";
      sendTickText(`[orch-tick: recover] re-dispatched ${r.key} (attempt ${r.attempt}, ${r.cause}${budget})`);
    }
    for (const e of outcome.escalated) {
      sendTickText(
        `[orch-tick: recover] ${e.key} exhausted auto-recovery (${e.attempts} attempt${e.attempts === 1 ? "" : "s"}, ${e.cause}) — it STAYS failed. Verify its pi-parallel-* branch, then your call: re-dispatch manually with a bigger budget / apply findings / drop. Not a user request; respond ≤2 lines.`,
        { premise: { kind: "failure", key: e.key } },
      );
    }
    for (const h of outcome.heldForHuman) {
      if (!h.surfaced) continue; // announced once, not on every sweep
      sendTickText(
        h.hold === "finisher"
          ? `[orch-tick: recover] ${h.key} is FINISHER-CLASS and failed (${h.cause}) — auto-recovery will NOT re-dispatch it (a finisher re-run can duplicate a landing, and a conflict-resolved cherry-pick or multi-commit squash leaves no evidence to detect). It STAYS failed. Check the target checkout (git log / git cherry): if the work is IN, close it out (failed → done with an overrideReason); if it is not, re-dispatch it deliberately. Not a user request; respond ≤2 lines.`
          : `[orch-tick: recover] ${h.key} is PARKED (blocked: ${h.blocker ?? "decision"}) after failing (${h.cause ?? "cause unrecorded"}) — auto-recovery will NOT re-dispatch it: retrying cannot supply what it needs. It is HELD for a human decision, not lost. When the blocker is resolved, RELEASE it: queue_update('${h.key}', {status: 'approved'}) then queue_dispatch — or drop it (status: 'rejected'). Not a user request; respond ≤2 lines.`,
        // The producing state differs by hold: a finisher hold names a still-
        // failed item; a PARK names a still-blocked one. A notice whose item
        // has left that state is stale.
        { premise: h.hold === "finisher" ? { kind: "failure", key: h.key } : { kind: "status", keys: [h.key], statuses: ["blocked"] } },
      );
    }
    for (const l of outcome.landedSkipped) {
      if (!l.surfaced) continue; // announced once, not on every sweep
      sendTickText(
        `[orch-tick: recover] ${l.key} is marked failed but its work is EVIDENCED AS LANDED (${l.repo} HEAD ${l.sha.slice(0, 8)}) — auto-recovery will NOT re-dispatch it (a re-run would duplicate a merge that already landed). Verify the landed commit, then close it out (failed → done) or re-open deliberately. Not a user request; respond ≤2 lines.`,
        { premise: { kind: "failure", key: l.key } },
      );
    }
    if (outcome.heldNotice) {
      sendTickText(
        `[orch-tick: recover] DEGRADED WINDOW: ${outcome.spawnFailures} consecutive provider failures (bare 400 / empty api_error / hang-then-die) — auto-recovery is PAUSED for a cooldown instead of churning retries. Failed items are held, not lost; a probe resumes automatically. Tell the user the provider looks degraded. Not a user request; respond ≤2 lines.`,
      );
    }
    scheduleRecoveryAt(outcome.nextAt);
  };
  // QUEUED harness info: auto-actions (dispatched / reviewed / re-dispatched)
  // accumulate into ONE consolidated tick, flushed at the next natural boundary
  // (the agent settles / the timer) instead of interrupting mid-turn. If the
  // router drops the flush (busy / cooldown), the parts are requeued — never
  // lost. The orchestrator learns what the harness DID without any gate bypass.
  let pendingHarness: string[] = [];
  let pendingFleet: number | undefined;
  /** FLEET-RPC DEGRADED EPISODE (AUTOPILOT-24): true while `backend.fleetStatus()`
   *  cannot tell us the fleet — it THREW, or it returned NULL after having
   *  answered before. One-shot like `heldNotified` in recovery-state.ts — the
   *  orchestrator learns the harness went fleet-blind once per episode and once
   *  when it recovers, not on every sweep. Silent degradation ('the
   *  orchestrator just stopped doing anything') is the failure class this
   *  framework keeps getting bitten by. */
  let fleetRpcDegraded = false;
  /** LEARNED CAPABILITY (AUTOPILOT-24): flipped by the first fleetStatus() that
   *  actually answers. The seam's type is `Promise<{totalActive}|null>`, so a
   *  backend that does not implement a fleet view answers `null` FOREVER and is
   *  UNSUPPORTED, not degraded — ticking every sweep for it would be pure spam.
   *  A backend that answered once and now returns null is BROKEN, and that is
   *  the real-world failure mode: the pi backend's `rpc()` resolves a timeout as
   *  `{success:false}` (backends/pi.ts) → fleetStatus() returns null, it never
   *  rejects. Capability is therefore learned at runtime instead of declared. */
  let fleetRpcEverAnswered = false;
  /** SPEC-PASS NUDGE (KEY: AUTOPILOT-36) — one-shot per EPISODE, the SAME
   *  discipline as `fleetRpcDegraded` above and `heldNotified`
   *  (recovery-state.ts): a proposal may be born thin, but it must never be
   *  presented for approval thin without at least a nudge that it needs a
   *  spec pass. This is a NUDGE, not an auto-spend (DELIBERATE, KEY:
   *  AUTOPILOT-36 deliverable 3) — the harness never spends a worker on a
   *  thin proposal by itself (some are about to be rejected; a worker spent
   *  on a drop wastes a slot). The count is re-derived from the live store
   *  every sweep (underSpecifiedProposals, config.ts — the SAME signal the
   *  panel tag and `/autopilot status` use), so the nudge can never drift
   *  from what the panel shows; only the DELIVERY is one-shot: the episode
   *  ends (and the flag resets) the moment the thin count returns to zero,
   *  so a FRESH batch of thin proposals always gets nudged again. */
  let specPassNudged = false;
  /** SERIALIZED-NO-DEPENDENCY RESIDUE NUDGE (KEY: AUTOPILOT-109) — one-shot per
   *  EPISODE, the SAME discipline as `specPassNudged`: a store that still holds
   *  a `blocked`+`serialized` item with NO `after` (the write gate now refuses
   *  to create them, but legacy records remain) is reported ONCE so the
   *  population cannot hide again. The episode ends (and the flag resets) the
   *  moment the count returns to zero, so a FRESH batch is nudged again. The
   *  item is NEVER auto-released — it may be a deliberate indefinite park, so
   *  the human decides. */
  let serializedResidueNudged = false;
  const harnessTick = (parts: string[], fleetTotalActive?: number): void => {
    pendingHarness.push(...parts);
    if (fleetTotalActive !== undefined) pendingFleet = fleetTotalActive;
    flushHarness();
  };
  const flushHarness = (): void => {
    if (!pendingHarness.length) return;
    const parts = pendingHarness;
    pendingHarness = [];
    const fleet = pendingFleet !== undefined ? ` — fleet ${pendingFleet}` : "";
    const r = router.send(
      `[orch-tick: harness] auto: ${parts.join("; ")}.${fleet} Your calls: approvals, high-risk checkpoints, review overrides, flag_for_review. Not a user request; respond ≤2 lines.`,
      { bypassCooldown: true },
    );
    if (r === "deferred") pendingHarness = [...parts, ...pendingHarness]; // runtime rejected — re-flush at the next settle
  };
  const sweep = async (source: "settled" | "activate" | "timer" | "worker-done"): Promise<void> => {
    // AUTOPILOT-120: NOT a blanket toggle gate. With autopilot OFF the
    // PROACTIVE lane below is skipped (see `on`) — nothing that STARTS work on
    // the sweep's own initiative (auto-dispatch, auto-review, auto
    // re-dispatch, recovery, zombie/dead-run flips, the shipping merge) — while
    // the INFORMATIONAL lane still runs so the operator keeps the event feed
    // ("what still tells me things when I switch it off"). The tick CLASS gate
    // in the router wrapper keeps the capacity ticks out regardless.
    const on = enabled();
    // WORKTREE HANDOFF PRESERVATION — capture the active runs' pi-parallel-*
    // tips into the durable journal + keep refs BEFORE anything else looks at
    // the queue. Rides the existing sweep (no new timers); running it on every
    // source means a dispatch is journalled within one sweep and a worker that
    // commits late still gets captured at the next state change. Best-effort.
    try {
      // AUTOPILOT-70: checkpoint FIRST — a worker that staged real edits but
      // never committed leaves work no keep-ref can rescue (a keep-ref
      // journals a COMMIT tip; staged changes are not a git object at all and
      // vanish with the worktree). Auto-committing any dirty worktree here
      // turns that into a normal committed tip, which the very next line then
      // journals in this same sweep pass instead of losing another cycle.
      // AUTOPILOT-120: the sweep's own preservation WRITES are proactive —
      // skipped while OFF (a completing run's tip is still captured
      // reactively, in onCompletion's transition-timed capture).
      if (on) {
        checkpointActiveItems(opts.stateDir);
        preserveActiveItems(opts.stateDir);
        // AUTOPILOT-47 C: and record WHERE each active run's worktree is, so a
        // reaped run's UNCOMMITTED changes are found by reading the item
        // instead of matching run ids to pi-worktree-* directories by hand.
        recordActiveWorktrees(opts.stateDir);
      }
    } catch {
      // preservation must never break the sweep
    }
    // SPEC-PASS NUDGE (KEY: AUTOPILOT-36) — fires on every sweep trigger (like
    // the fleet-degraded notice above) but DELIVERS once per episode.
    try {
      const thin = underSpecifiedProposals(opts.stateDir);
      if (thin.length === 0) {
        specPassNudged = false; // episode ended — the next batch gets a fresh nudge
      } else if (!specPassNudged) {
        specPassNudged = true;
        sendTickText(
          `[orch-tick: harness] ${thin.length} proposal${thin.length === 1 ? "" : "s"} need a spec pass (${thin.map((t) => t.key).join(", ")}) — press 's' in the panel to dispatch a bounded, read-only worker that reads the code and completes the scope, or ask. This is a nudge, not an auto-spend: they still approve as-is if you decide to; nothing here dispatches on its own. Not a user request; respond ≤ 2 lines.`,
        );
      }
    } catch {
      // the nudge must never break the sweep
    }
    // SERIALIZED-NO-DEPENDENCY RESIDUE (KEY: AUTOPILOT-109) — a `blocked` +
    // `serialized` item with NO `after` is a contradiction that nothing can
    // release (see serializedWithoutDependency). The write gate refuses new
    // ones; this reports any that remain ONCE per episode and leaves the
    // decision to a human — never silently released.
    try {
      const residue = serializedWithoutDependency(opts.stateDir);
      if (residue.length === 0) {
        serializedResidueNudged = false; // episode ended — a fresh batch gets a fresh nudge
      } else if (!serializedResidueNudged) {
        serializedResidueNudged = true;
        sendTickText(
          `[orch-tick: harness] ${residue.length} blocked item${residue.length === 1 ? "" : "s"} SERIALIZED with no dependency (${residue.join(", ")}) — nothing can release ${residue.length === 1 ? "it" : "them"} automatically. Re-park as parked/decision, set \`after\`, or release by hand. NOT auto-released: it may be a deliberate park. Not a user request; respond ≤ 2 lines.`,
        );
      }
    } catch {
      // the residue nudge must never break the sweep
    }
    // CONFIG CHANGE SIGNAL (KEY: AUTOPILOT-48) — a fleet-wide config edit
    // (`/autopilot capacity 6` writes autopilot.config.json) used to reach the
    // HUMAN's TUI only: the orchestrator kept dispatching against the cached
    // cap and told the user work was gated after the cap had doubled. The
    // change now rides the SAME [orch-tick] channel as queue decisions, ONCE
    // per change (the surfaced values are snapshotted next to the harness
    // state). A permanently dropped tick (no loaded orchestrator) is NOT
    // recorded as surfaced — the next sweep re-announces it.
    try {
      const current = cfg();
      const changes = detectConfigChanges(opts.stateDir, current);
      const message = configChangeTick(changes);
      if (!message) {
        recordConfigSnapshot(opts.stateDir, current); // baseline / unchanged
      } else if (sendTickText(message) !== "dropped") {
        recordConfigSnapshot(opts.stateDir, current);
      }
    } catch {
      // the config notice must never break the sweep
    }
    // MAIN-IMMUTABILITY GUARD (KEY: MAIN-IMMUTABILITY-GUARD) — the MECHANICAL
    // enforcement of "nothing merges to main before human approval" (the rule
    // is codified in the skills/prompts, but guidance alone can lose — a live
    // incident committed a recovered deliverable to main pre-review). Every
    // reconcile step records the main HEAD of every queue-referenced repo; a
    // worker round that moves a tracked main ref while ≥1 item referencing
    // that repo is pre-done (not human-approved) raises the
    // orch:main-write-pre-approval WARNING event + a loud tick naming the SHA
    // + the offending keys. The ONLY legitimate main-write path is the
    // post-approval merge-finisher (MR when a remote exists, else merge to
    // main) — nothing auto-merges.
    try {
      for (const w of checkMainWrites(opts.stateDir)) {
        if (opts.emit) {
          opts.emit([{ name: "orch:main-write-pre-approval", data: { severity: "warning", ...w } }]);
        }
        const short = (s: string): string => s.slice(0, 8);
        const keys = w.keys.join(", ");
        const message = `[orch-tick: main-write] MAIN-IMMUTABILITY VIOLATION: ${w.ref} moved ${short(w.previousSha)} → ${short(w.sha)} in ${w.repo} while ${keys} ${w.keys.length > 1 ? "are" : "is"} NOT human-approved (pre-done). The ONLY legitimate main-write path is the merge-finisher AFTER the human's done (MR when a remote exists, else merge to main); nothing auto-merges. Verify what landed (git log ${short(w.sha)}); recovery stays on the branch. Not a user request; respond ≤2 lines.`;
        const r = router.send(message, { bypassCooldown: true });
        if (r === "deferred") queueDeferred(message, "tick");
      }
    } catch {
      // the guard must never break the sweep
    }
    // SHIPPING LANE (KEY: AUTO-SHIP-ON-DONE) — the deterministic post-approval
    // merge-finisher. `done` is human-approved, NOT merged; this reconcile step
    // reacts to done: a done item with a declared PER-REPO policy ships by
    // itself (MRs per baseBranch / merge to local main); a done item WITHOUT a
    // policy fires the ONE-TIME policy-inquiry ask and stays paused — nothing
    // merges before the policy is set (NO FALLBACK, never guess); mergeMode
    // manual suppresses the lane (batch finishers stay explicit). mergeMode
    // auto is the default. Every action lands as a `[orch-tick: ship]` tick +
    // a domain event; shippedAt markers prevent re-merge. Best-effort like the
    // guard above: a git/store failure must never break the sweep.
    // ORDERING: this runs AFTER the guard, so a flow 'merge' ship moves main
    // past the baseline the guard just took. The lane closes that gap itself
    // via recordExpectedMainWrite (the narrow expected-write handshake) — its
    // own legitimate merge must never fire a main-immutability violation, and
    // any OTHER main move still does.
    try {
      // AUTOPILOT-120: the merge-finisher MERGES — PROACTIVE, so while OFF it
      // runs in NOTIFY-ONLY mode: it still DETECTS + ANNOUNCES a done item that
      // is ready to ship (an INFORMATIONAL [orch-tick: ship] the operator wants
      // even with the harness off), but never merges/pushes or marks shipped.
      for (const o of runShippingPass(opts.stateDir, undefined, { notifyOnly: !on })) {
        if (o.event && opts.emit) opts.emit([o.event]);
        const r = router.send(o.message, { bypassCooldown: true });
        if (r === "deferred") queueDeferred(o.message, "tick");
      }
    } catch {
      // the shipping lane must never break the sweep — a failure inside is
      // already an outcome of the pass, not an exception
    }
    // The authoritative fleet, fetched ONCE — both the auto-dispatch (A) and
    // the engine sweep use it (one RPC, and the request is emitted synchronously
    // so hosts/replies see it immediately).
    // GUARDED (AUTOPILOT-24): this was the ONE bare await in sweep(). A throwing
    // status RPC used to skip EVERYTHING below it (zombie reconciliation,
    // auto-dispatch, recovery, the engine sweep) AND escape as an unhandled
    // rejection in the host process. A throw now degrades to the UNKNOWN fleet
    // (`null`) — a state every consumer below already handles — so the rest of
    // the sweep still runs off the store/ledger inventory.
    //
    // WHAT COUNTS AS DEGRADED: "could not determine the fleet", not just
    // "threw". The pi backend NEVER rejects — `rpc()` resolves a timeout as
    // `{success:false, error}` and fleetStatus() turns that into `null`
    // (backends/pi.ts) — so a null RETURN is the failure mode that actually
    // happens in production and the throw path is close to unreachable there.
    // But `null` is ALSO the seam's legitimate "I have no fleet view" answer,
    // and a backend that never implements one would tick every single sweep.
    // The two are told apart by LEARNED CAPABILITY (`fleetRpcEverAnswered`):
    // null degrades only once this runner has seen the backend answer at least
    // once (or an episode is already open). A throw always degrades — the seam
    // spells "unsupported" as null, never as an exception.
    let fleet: { totalActive: number; foreignRuns?: number; unattributedRuns?: number } | null = null;
    let fleetFailure: string | null = null; // null = the RPC answered
    let fleetThrew = false;
    try {
      fleet = await opts.backend.fleetStatus();
      if (!fleet) fleetFailure = "returned NULL (no fleet answer — a failed/timed-out status RPC resolves null instead of rejecting)";
    } catch (e) {
      // UNKNOWN fleet, never a fake 0 — zombieReconcile(undefined) refuses to flip.
      fleet = null;
      fleetThrew = true;
      fleetFailure = `THREW (${(e instanceof Error ? e.message : String(e)).slice(0, 200)})`;
    }
    if (fleetFailure === null) {
      fleetRpcEverAnswered = true; // capability proven — nulls from here on are BREAKAGE
      if (fleetRpcDegraded) {
        fleetRpcDegraded = false;
        appendTelemetry(opts.stateDir, JSON.stringify({ t: new Date().toISOString(), type: "fleet-rpc", state: "recovered", source }));
        sendTickText(
          `[orch-tick: harness] fleet RPC RECOVERED: backend.fleetStatus() answers again (fleet ${fleet ? fleet.totalActive : "unknown"} active) — fleet-aware sweeps resume, including zombie reconciliation. Not a user request; respond ≤2 lines.`,
        );
      }
    } else if (fleetThrew || fleetRpcEverAnswered || fleetRpcDegraded) {
      // Not the never-answered (UNSUPPORTED) backend: this one lost a fleet
      // view it demonstrably had, so the blindness is worth saying out loud.
      appendTelemetry(
        opts.stateDir,
        JSON.stringify({ t: new Date().toISOString(), type: "fleet-rpc", state: "failed", source, mode: fleetThrew ? "threw" : "null", error: fleetFailure }),
      );
      if (!fleetRpcDegraded) {
        fleetRpcDegraded = true;
        sendTickText(
          `[orch-tick: harness] DEGRADED: backend.fleetStatus() ${fleetFailure} — the sweep CONTINUES with an unknown fleet: auto-dispatch + recovery still run off the store/ledger inventory, and zombie reconciliation is SUSPENDED (unknown is never read as 0, so no live item is flipped to failed by an RPC blip). One notice per degraded episode; a recovery tick follows when the RPC answers again. Not a user request; respond ≤2 lines.`,
        );
      }
    }
    // FLEET-vS-INVENTORY PARITY (AUTOPILOT-6): the RPC count is the floor, never
    // a replacement for what the engine's own view (the store) knows is
    // running — a transient undercount (status RPC lagging a just-started
    // parent) must not let the harness OVER-SPAWN into phantom free slots.
    // Auto-dispatch fires on every free-slot window now (not just
    // worker-done), so this guard matters more. Zombie reconciliation uses the
    // BACKEND's corrected count (fleetStatus now unions the RPC with the
    // in-flight async-run inventory, so a 0 really means nothing is running
    // anywhere this backend spawns; the undefined guard still covers an RPC
    // failure — never a fake 0).
    // KEY: AUTOPILOT-68/86 — this used to ALSO take `autopilot.status().running`
    // (the engine's run ledger) into the max, the SAME leaked-forever counter
    // AUTOPILOT-61 banned from the MANUAL queue_list/queue_dispatch readers. A
    // ledger entry whose completion event never arrives leaks forever, and one
    // leaked entry poisoned this THIRD computation (auto-dispatch's and
    // recovery's shared input) regardless of what the real fleet or store said
    // (live-observed: occupied 12 / ledger 12 / fleetTotalActive 2 against
    // maxSlots 6 — real active fleet 0 — so nothing dispatched for 20+ min).
    // The ledger term is gone, AND the store term is now the SAME derivation
    // the tools and the tick use (effectiveOccupiedSlots → liveTrackedSlots):
    // it counts `active` items whose run is live plus `ai-review` items with a
    // live reviewer, so a stopped/killed run releases its slot here too instead
    // of a separate storeToSnapshot count that could diverge from the tick.
    // Spawn-race tradeoff considered and rejected (AUTOPILOT-68): fleetStatus()
    // already unions the RPC with the in-flight async-run inventory on disk,
    // and autoDispatchEligible writes `active` synchronously before the next
    // slot check in the same batch, so no free-slot window relied on the
    // ledger to close a real gap.
    const effectiveTotalActive = effectiveOccupiedSlots(opts.stateDir, {
      backend: opts.backend,
      zombieGraceMinutes: cfg().zombieGraceMinutes,
      fleetTotalActive: fleet?.totalActive ?? null,
    });
    // KEY: AUTOPILOT-96 — remember the GATE's occupancy for this pass. The tick
    // sweep below is handed THIS number (projection), and the premise re-check
    // + delivery-time refresh project it too, so no reader can disagree with
    // what auto-dispatch just gated on.
    lastGateOccupied = effectiveTotalActive;
    // AUTOPILOT-120: the timer branch is PROACTIVE — retention + the
    // zombie/dead-run safety nets all MUTATE active items. Skipped while OFF.
    if (source === "timer" && on) {
      // RETENTION: drop keep refs whose job is done (item terminal, or tip
      // reachable from main) so preserved refs don't accumulate forever.
      try {
        prunePreservedRefs(opts.stateDir);
      } catch {
        // never let retention break the sweep
      }
      // Zombie reconciliation FIRST (deterministic safety net): flip active
      // items whose run is provably gone (fleet idle past grace) so a lost
      // completion event cannot wedge an item in active forever. Any flips
      // ride the consolidated harness tick — the orchestrator learns WHAT was
      // flipped and checks pi-parallel-* branches before re-dispatching. The
      // preservation capture above already ran, so even a timed-out worker's
      // last commits were journalled while its branch still existed.
      try {
        // DEAD-RUN reconciliation runs FIRST and is NOT fleet-gated (AUTOPILOT-47).
        // zombieReconcile below needs an authoritatively IDLE fleet, which a busy
        // queue never has — one live worker made every dead run in the store
        // immune (observed: 117m and 287m stale items while other workers ran).
        // This pass condemns a run on its OWN status.json saying a terminal
        // phase, and on nothing weaker: an unreadable/absent/unknown status is
        // absence of evidence and is left alone (see run-liveness.ts).
        const dead = autopilot.deadRunReconcile((runId) => runStateEvidence(opts.backend, runId));
        if (dead.flippedKeys.length) {
          harnessTick([`dead-run reconciliation flipped ${dead.flippedKeys.join(", ")} to failed (the run's own status.json reports a terminal state — its completion event was lost; verify partial work on pi-parallel-* branches before re-dispatch)`], fleet?.totalActive);
          for (const key of dead.flippedKeys) {
            const decision = decisionTick({ key, action: "failed", from: "active", to: "failed", note: "dead-run" });
            if (!decision) continue;
            const sendResult = router.send(decision, { bypassCooldown: true });
            if (sendResult === "deferred") queueDeferred(decision, "tick");
          }
        }
      } catch {
        // never let the safety net break the sweep
      }
      try {
        const zombies = autopilot.zombieReconcile(fleet?.totalActive); // the corrected count (RPC + async-inventory union) — only a genuine 0 may trigger the net
        if (zombies.flippedKeys.length) {
          harnessTick([`zombie reconciliation flipped ${zombies.flippedKeys.join(", ")} to failed (no live run — verify partial work on pi-parallel-* branches before re-dispatch)`], fleet?.totalActive);
          // DECISION TICK per flip — the canonical one-line move record, in
          // addition to the consolidated harness line above (which keeps the
          // verify-branch guidance). Router-gated like every other harness
          // tick: a busy deferral holds it for the settle; a permanent drop
          // is covered by the harness line — never lost, never a surprise.
          for (const key of zombies.flippedKeys) {
            const decision = decisionTick({ key, action: "failed", from: "active", to: "failed", note: "zombie" });
            if (!decision) continue;
            const sendResult = router.send(decision, { bypassCooldown: true });
            // REVIEW/COMPLETE are already ticked (review ticks on verdict
            // routing; the harness tick on worker completion/auto-actions) —
            // zombie is the one harness flip that only got the batch line.
            if (sendResult === "deferred") queueDeferred(decision, "tick");
          }
        }
      } catch {
        // never let the safety net break the sweep
      }
    }
    // A — fill free slots with auto-dispatchable items before deciding ticks,
    // so the dispatch nudge fires only for the MANUAL cases (high-risk or
    // incomplete scope/cwd). Fires on EVERY free-slot window — the activation
    // path, the turn-end settle, the periodic sweep, AND worker completions:
    // the docs/skill contract is "an approved item fills a free slot by itself"
    // (no worker-done qualifier), and the worker-done-only gate stranded
    // approved items next to idle slots until a completion happened (observed
    // live: early-ON window fired ZERO auto-dispatches; idle free slots needed
    // the manual dispatch tick — AUTOPILOT-9). The eligible-item + slot guards
    // make the extra triggers inert when there is nothing to do.
    // AUTOPILOT-120: PROACTIVE — starts worker/reviewer/spec-pass runs;
    // skipped while OFF.
    if (autoDispatchOn && on) {
      try {
        const dispatched = await autoDispatchEligible(opts.stateDir, opts.backend, cfg().maxSlots, effectiveTotalActive, fleet?.foreignRuns, fleet?.unattributedRuns);
        if (dispatched.length) harnessTick([`dispatched ${dispatched.map((d) => d.key).join(", ")}`], fleet?.totalActive);
      } catch {
        // silent — the tick still nudges the manual cases
      }
      // B2 — HUMAN RE-DISPATCH (KEY: AUTOPILOT-59): the decision panel's
      // "redispatch" action already moved the item human-review → active and
      // recorded the human's findings (the single point of truth against
      // double-dispatch — both hosts reach this SAME pass, whether the panel
      // ran in-process (pi) or only mutated the shared store (opencode's
      // client-side TUI, which has no backend to spawn through directly).
      // This is the harness consumer: it spawns the redo through B
      // (autoRedispatch) — no slot gate, exactly like a review-FAIL
      // re-dispatch, because the item already occupies the slot its own move
      // to active just counted against effectiveTotalActive.
      try {
        const redispatched = await dispatchPendingHumanRedispatch(opts.stateDir, opts.backend);
        if (redispatched.length) harnessTick([`re-dispatched ${redispatched.map((d) => d.key).join(", ")} with the human's findings`], fleet?.totalActive);
      } catch {
        // silent — the panel already confirmed the findings were recorded; the
        // next sweep retries
      }
      // SPEC PASS (KEY: AUTOPILOT-36): the harness consumer for the panel's
      // `s` action — spawns a bounded, READ-ONLY worker against ONE
      // under-specified proposal (pendingSpecPass marker — see panels.ts,
      // dispatchPendingSpecPass in auto-dispatch.ts). Reuses the SAME dispatch
      // + worktree machinery as every other lane; it never touches status —
      // this can never approve anything.
      try {
        const specPassed = await dispatchPendingSpecPass(opts.stateDir, opts.backend);
        if (specPassed.length) harnessTick([`started a spec pass on ${specPassed.map((d) => d.key).join(", ")} (read-only — will write a completed scope back when done, still a proposal)`], fleet?.totalActive);
      } catch {
        // silent — the nudge above re-surfaces the item; the next sweep retries
      }
    }
    // AUTOMATIC RECOVERY of failed items (AUTO-RECOVER-FAILS): the failed lane
    // re-dispatches by policy AFTER the backoff; running it after auto-dispatch
    // lets the engine's slot math see the just-filled slots. The pass arms its
    // own backoff timer for the next due attempt.
    // Guarded at the CALL SITE too: runRecoveryPass's own try/catch covers the
    // autoRecoverFails engine call, but the tick delivery AFTER it (sendTickText
    // → the deferral flush → refreshTickFacts against the live store) is outside
    // that guard, so a failure there would still abort the engine sweep below.
    // AUTOPILOT-120: PROACTIVE — auto-recovery re-dispatches failed items;
    // skipped while OFF (runRecoveryPass also re-checks the toggle for its own
    // backoff timer).
    if (on) {
      try {
        await runRecoveryPass(effectiveTotalActive);
      } catch {
        // recovery announcements must never break the sweep
      }
    }
    // THE TICK IS A PROJECTION OF THE GATE'S OCCUPANCY (KEY: AUTOPILOT-96):
    // pass the number auto-dispatch just gated on so the FLEET line and the
    // free-slot count can never diverge from the dispatch decision.
    const result = autopilot.sweep(source, Date.now(), { occupied: effectiveTotalActive, ...(fleet ? { totalActive: fleet.totalActive } : {}) }, opts.backend);
    // AUTOPILOT-120: the engine sweep is queue-read-only; it yields a
    // dispatch/intake capacity tick (AUTOMATIC-ACTION) or the informational
    // budget warning. Deliver it when it is informational or when ON; a
    // SUPPRESSED capacity tick must NOT mask the informational stuck-review
    // fallback below.
    const engineTickDeliverable = result.tick ? on || deliverableWhileOff(result.tick.message) : false;
    if (result.tick && engineTickDeliverable) sendTick(result.tick);
    // timer/activate safety net: stuck ai-review items get a review nudge —
    // the "stuck" wording (reviewers may still be running; never claim completion).
    // The runEvidence seam (AUTOPILOT-47's terminal-status distinction, AUTOPILOT-50)
    // lets reviewTick skip items whose reviewer is PROVABLY still running instead of
    // nudging on every ai-review item regardless of liveness.
    else if (source === "timer" || source === "activate") sendTick(autopilot.reviewTick(undefined, "stuck", (runId) => runStateEvidence(opts.backend, runId)));
  };

  /** One-shot dedupe for the sweep-failure tick (same episode discipline as
   *  `fleetRpcDegraded`): cleared by the next sweep that completes. */
  let sweepFailureNotified = false;
  /** EVERY sweep trigger goes through here (AUTOPILOT-24). A rejected sweep
   *  used to be a bare `void sweep(...)` — an UNHANDLED PROMISE REJECTION
   *  inside the host process (pi extension / opencode plugin), which can take
   *  the host down or vanish silently depending on the runtime. The guarded
   *  fleet RPC above removed today's known thrower; this records tomorrow's
   *  unknown one instead of losing it. */
  const runSweep = (source: "settled" | "activate" | "timer" | "worker-done"): void => {
    void sweep(source)
      .then(() => {
        sweepFailureNotified = false; // a completed sweep ends the failing episode
      })
      .catch((e: unknown) => {
        try {
          const reason = (e instanceof Error ? e.message : String(e)).slice(0, 200);
          appendTelemetry(opts.stateDir, JSON.stringify({ t: new Date().toISOString(), type: "sweep-failure", source, error: reason }));
          if (sweepFailureNotified) return; // one notice per failing episode, not per sweep
          sweepFailureNotified = true;
          sendTickText(
            `[orch-tick: harness] SWEEP FAILED (${source}): ${reason} — the harness caught it instead of losing it to an unhandled rejection, but THIS sweep did no work (dispatch, recovery and ticks were skipped). Later triggers retry; if it repeats, the harness is effectively stopped. Not a user request; respond ≤2 lines.`,
          );
        } catch {
          // the failure recorder itself must never reject — that would recreate
          // the very unhandled rejection this wrapper exists to prevent
        }
      });
  };

  let timer: ReturnType<typeof setInterval> | null = null;

  return {
    onCompletion(ev) {
      // AUTOPILOT-120: a completion is REACTIVE. With autopilot OFF the harness
      // still RECORDS the outcome and emits the INFORMATIONAL tick
      // (review/failure) — the operator keeps the event feed — while the
      // PROACTIVE lane below (auto-review dispatch, auto re-dispatch, the
      // freed-slot dispatch sweep) is skipped. The tick CLASS gate in the
      // router wrapper keeps any automatic-action tick out.
      const on = enabled();
      const result = autopilot.handleAsyncComplete(ev);
      if (opts.emit) opts.emit(result.domainEvents);
      // TRANSITION-TIMED preservation capture: the active→ai-review/failed
      // flip is the last queue-state change this run gets — journal its
      // parallel-branch tip NOW so commits that landed after the previous
      // sweep are still recorded (best-effort: if the runtime's own cleanup
      // already removed the branch, this no-ops and the sweep captures hold).
      // AUTOPILOT-85: pass the completion event's OWN reported branch (when the
      // runner names one) through as `knownBranch` — live-repo discovery alone
      // (parallelBranches) finds nothing once the runtime's cleanup has already
      // deleted the worktree/branch, which is exactly the gap that let a `done`
      // item silently never journal a tip.
      try {
        const doneRunId = String((ev as { runId?: string }).runId ?? "");
        const store = doneRunId ? loadStore(opts.stateDir) : null;
        const item = store ? itemByRunId(store, doneRunId) : null;
        if (item?.cwd)
          preserveRunWorktree({
            stateDir: opts.stateDir,
            repo: item.cwd,
            runId: doneRunId,
            key: item.key,
            knownBranch: (ev as { branch?: string | null }).branch ?? null,
          });
      } catch {
        // preservation never breaks completion routing
      }
      // B + C, ONE consolidated harness tick. B — a review FAIL
      // auto-re-dispatches with the findings (up to the cap; the engine
      // already flipped to active when attempts < cap). C — a worker
      // completion (active → ai-review) auto-dispatches its review with the
      // SAME fields the dispatch used (KEY + scope + cwd). The orchestrator
      // learns what happened from the single harness tick and can still
      // queue_review/queue_dispatch to override.
      // AUTOPILOT-120: PROACTIVE — starts a review or a redo run; skipped
      // while OFF (the completion itself is still recorded + ticked above).
      if (autoDispatchOn && on) {
        const verdict = result.domainEvents.find((e) => e.name === "orch:verdict" && e.data?.verdict === "FAIL");
        const completed = result.domainEvents.find(
          (e) => e.name === "orch:item-completed" && e.data?.outcome === "ai-review",
        );
        const tasks: Array<Promise<{ part: string; ok: boolean }>> = [];
        if (verdict) {
          const findings = (Array.isArray(ev.results) ? ev.results : [])
            .map((r) => (r as { agent?: string; output?: string })?.output ?? "")
            .filter(Boolean)
            .join("\n\n");
          const key = String(verdict.data?.key ?? "");
          if (key && findings) {
            tasks.push(autoRedispatch(opts.stateDir, opts.backend, key, findings).then((ok) => ({ part: `re-dispatched ${key} with findings`, ok })));
          }
        }
        if (completed) {
          const key = String(completed.data?.key ?? "");
          if (key) {
            tasks.push(
              autoReview(opts.stateDir, opts.backend, cfg().reviewerAgents[0] ?? "orchestrator-reviewer", key).then((runId) => ({
                part: `reviewer for ${key}${runId ? ` (run ${String(runId).slice(0, 8)})` : ""}`,
                ok: !!runId,
              })),
            );
          }
        }
        if (tasks.length) {
          void Promise.all(tasks)
            .then((results) => {
              const parts = results.filter((r) => r.ok).map((r) => r.part);
              if (parts.length) harnessTick(parts);
            })
            .catch(() => {
              // silent — the review tick nudges the orchestrator
            });
        }
      }
      
      // DETERMINISTIC HANDOVER: on a review PASS the framework auto-flags for
      // the user's review, built from the ITEM (title/scope/risk/cwd — the
      // orchestrator wrote those at approval, so no judgment is needed here).
      // The user ALWAYS gets the notification; the orchestrator's manual
      // flag_for_review becomes the refinement (specific files/commits).
      const pass = result.domainEvents.find((e) => e.name === "orch:verdict" && e.data?.verdict === "PASS");
      if (pass) {
        try {
          const key = String(pass.data?.key ?? "");
          const store = loadStore(opts.stateDir);
          const item = store?.items[key];
          if (item) {
            // The risk LABEL is the item's CLOSED-set value, parsed by the ONE
            // parser (item-contract.ts's isRisk) — the vocabulary lives there,
            // not copied here (AUTOPILOT-128 review caveat). Any item that
            // reached a verdict passed the approval + dispatch contract, so its
            // risk IS a member; the `medium` fallback is only for a legacy/
            // bypassed record and affects this NOTIFICATION label alone (never
            // admission — the contract gates scope/cwd/risk before both).
            const risk: Risk = isRisk(item.risk) ? item.risk : "medium";
            const scopeHead = (item.scope ?? "").split(/\n/)[0].trim().slice(0, 80);
            // REVIEW TARGETS (AUTOPILOT-57): rebuilt fresh for THIS attempt —
            // branch+commits, reviewer-cited file:line anchors, changed doc
            // paths, MR/PR link — ranked + capped, never fabricated when
            // nothing resolves. Persisted onto the ITEM (not only this
            // rendered message) so the handover survives a session, a
            // reload and a scroll; the decision panel reads the same field.
            const targets = buildReviewTargets(opts.stateDir, key, item, item.reviewNotes);
            mutateStore(opts.stateDir, (s) => {
              if (s.items[key]) updateItem(s, key, { reviewTargets: targets });
            });
            // THE REVIEWER'S OWN CAVEATS (AUTOPILOT-27): captured at verdict
            // parse onto the item, replayed here so the notification and the
            // approval screen show the SAME material — the automatic flag used
            // to pass no residual_risks at all because the item did not know
            // them.
            const residualRisks = reviewNotesResidualRisks(item.reviewNotes);
            flagForReview(
              {
                summary: `${item.title || key} — agent review PASSED, awaiting your approval (human-review). Accept with queue_update status: done${scopeHead ? ` (${scopeHead}…)` : ""}.`,
                risk,
                blast_radius: `The reviewed work lands in the repo at ${item.cwd ?? "?"} — if wrong, ${item.title || key} is affected.`,
                review_targets: targets.length ? targets : [noReviewTargetsFoundLine(item, key)],
                self_reviewed: true,
                review_method: "reviewer-subagent",
                queue_key: key,
                origin: "automatic",
                ...(residualRisks ? { residual_risks: residualRisks } : {}),
              },
              { logPath: join(opts.stateDir, "reviews.jsonl") },
            );
          }
        } catch {
          // the PASS tick below still nudges the orchestrator to flag manually
        }
      }
      sendTick(result.tick);
      if (result.freedSlot) {
        // ANY run completing frees a slot → dispatch sweep with the
        // authoritative fleet count.
        runSweep("worker-done");
      }
      if (result.reviewerCompleted) {
        sendTick(autopilot.reviewTick());
      }
    },
    deferUserMessage,
    deferTick,
    onSettled() {
      // flush the shared deferral + the queued harness info FIRST (the agent
      // just settled — the busy gate is clear), then the sweep's nudges.
      flushDeferred();
      flushHarness();
      runSweep("settled");
    },
    onTimer() {
      flushDeferred(); // backstop if the agent never settles
      flushHarness();
      runSweep("timer"); // never lets the timer break the host (sync or async)
    },
    activate() {
      runSweep("activate");
    },
    start() {
      if (opts.sweepIntervalMs > 0 && !timer) {
        timer = setInterval(() => this.onTimer(), opts.sweepIntervalMs);
      }
    },
    stop() {
      if (timer) clearInterval(timer);
      timer = null;
      if (retryTimer) clearTimeout(retryTimer);
      retryTimer = null;
      if (recoveryTimer) clearTimeout(recoveryTimer);
      recoveryTimer = null;
    },
  };
}