// shipping.ts — the SHIPPING LANE / merge-finisher (KEY: AUTO-SHIP-ON-DONE).
//
// `done` means human-approved, NOT merged: moving approved work into a base
// branch is the SEPARATE post-approval step this module owns, and it is gated
// by a PER-REPO SHIPPING POLICY (autopilot.config.json → `shipping.repos`,
// keyed by origin-slug or repo basename). NO FALLBACK — the msf incident root
// was a guess; a repo without a policy NEVER gets a guessed flow/base. The
// shipping pass NOTICES, asks the user ONCE (a pending intercom inquiry +
// ticket naming flow/baseBranches + detection-based hints), stays paused, and
// resumes on the sweep AFTER the orchestrator writes the one-time answer.
//
// Deterministic by construction — the lane is framework-owned git, never an
// LLM's judgment:
//   - flow 'mrs'   → for each baseBranch IN ORDER push the approved work's
//                    tip to origin and record an MR entry (base → MR URL /
//                    push ref). Requires a remote; never `--force`.
//   - flow 'merge' → merge the approved work into the LOCAL base branch
//                    (default main). No remote involved.
//   - skip when nothing new (tip already an ancestor of the base).
//   - conflicts → failure + escalation, never a forced merge.
//   - the item's `shippedAt` marker prevents re-merge; clearing it (a human
//     re-open leaves done) re-arms the lane.
//
// The runner calls runShippingPass on every reconcile sweep — one lane, one
// wiring point, both hosts (pi + opencode).

import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { loadStore, mutateStore, updateItem, type QueueItem, type UpdatePatch } from "../queue-store.ts";
import { appendTelemetry, loadAutopilotConfig, type ShippingConfig, type ShippingRepoPolicy, type ShippingFlow } from "../config.ts";
import { reviewPointersFor } from "./worktree-preservation.ts";
import { recordExpectedMainWrite } from "./main-write-guard.ts";
// The repo-key derivation lives in the LEAF src/repo-keys.ts (AUTOPILOT-110)
// so the model level map can match a repo the SAME way without importing this
// module (which would cycle through config.ts). Re-exported here for the
// existing importers/tests that predate the extraction.
import { originSlug, repoPolicyKeys, suggestedPolicyKey } from "../repo-keys.ts";
import { git, gitExitCode } from "./git.ts";
export { originSlug, repoPolicyKeys, suggestedPolicyKey };

export type { ShippingConfig, ShippingRepoPolicy, ShippingFlow };

// ---------------------------------------------------------------------------
// Policy resolution — NO FALLBACK (never guess flow/baseBranches; the msf
// incident root).
// ---------------------------------------------------------------------------

/** Validate a raw policy entry: flow ∈ {mrs, merge} and a non-empty
 *  baseBranches string array (ANY strings — order is the MR sequence). An
 *  invalid entry resolves as NO policy — the inquiry fires instead of a guess. */
export function normalizePolicy(raw: unknown): ShippingRepoPolicy | null {
  if (!raw || typeof raw !== "object") return null;
  const r = raw as Record<string, unknown>;
  const flow = r.flow;
  if (flow !== "mrs" && flow !== "merge") return null;
  const bases = Array.isArray(r.baseBranches) ? r.baseBranches : [];
  if (!bases.length || !bases.every((b) => typeof b === "string" && b.trim().length > 0)) return null;
  return { flow, baseBranches: bases.map((b) => String(b).trim()) };
}

export interface ResolvedPolicy {
  /** The config key that matched (`repos[<key>]`). */
  key: string;
  policy: ShippingRepoPolicy;
}

/** Resolve the policy for one repo. NULL on a policy-less repo — the caller
 *  fires the policy-inquiry instead of shipping anything (NO FALLBACK). */
export function resolveRepoPolicy(stateDir: string, repo: string, cfg?: ReturnType<typeof loadAutopilotConfig>): ResolvedPolicy | null {
  const shipping = (cfg ?? loadAutopilotConfig(stateDir)).shipping;
  const repos = shipping?.repos ?? {};
  for (const key of repoPolicyKeys(repo)) {
    const policy = normalizePolicy(key in repos ? repos[key] : undefined);
    if (policy) return { key, policy };
  }
  return null;
}

/** The merge mode (auto/manual). Default auto. `manual` suppresses the lane —
 *  batch finishers stay an explicit orchestrator act (the tick nudges). */
export function mergeModeOf(cfg: Pick<ReturnType<typeof loadAutopilotConfig>, "shipping">): "auto" | "manual" {
  return cfg.shipping?.mergeMode === "manual" ? "manual" : "auto";
}

// ---------------------------------------------------------------------------
// Policy-inquiry state (the one-time intercom ask; the run stays paused until
// the orchestrator writes the answer).
// ---------------------------------------------------------------------------

export interface PolicyInquiry {
  key: string;         // the config key suggested for `repos`
  repo: string;
  question: string;
  hints: string[];
  ts: string;
}

interface ShippingState {
  version: 1;
  /** Pending per-repo policy inquiries (one-time intercom asks). A present
   *  entry means the question is already out — no re-ask, no shipping. */
  inquiries: Record<string, PolicyInquiry>;
/** One-time suppression for non-shipping nudges (manual-mode nudges, failure
 *  floors). The value is the item's `updatedAt` at notify time — RE-ARMED when
 *  the item is re-approved (a status change re-stamps updatedAt, so a fresh
 *  done cycle gets ONE new escalation instead of a permanent silence). */
  notified: Record<string, string>;
}

export function shippingStatePath(stateDir: string): string {
  return join(stateDir, "shipping.json");
}

function readShippingState(stateDir: string): ShippingState {
  try {
    const p = shippingStatePath(stateDir);
    if (!existsSync(p)) return { version: 1, inquiries: {}, notified: {} };
    const raw = JSON.parse(readFileSync(p, "utf8")) as ShippingState;
    return { version: 1, inquiries: raw?.inquiries ?? {}, notified: raw?.notified ?? {} };
  } catch {
    return { version: 1, inquiries: {}, notified: {} };
  }
}

function writeShippingState(stateDir: string, s: ShippingState): void {
  try {
    const p = shippingStatePath(stateDir);
    mkdirSync(stateDir, { recursive: true });
    const tmp = `${p}.tmp-${process.pid}`;
    writeFileSync(tmp, JSON.stringify(s, null, 2) + "\n", "utf8");
    renameSync(tmp, p);
  } catch {
    // best-effort — shipping still proceeds on the next sweep after a policy lands
  }
}

/** Detection-based candidate hints for the policy ask (setup help ONLY — what
 *  the repo offers, never a recommendation): the remote default branch
 *  (refs/remotes/origin/HEAD), local + origin branch names. Branch protection
 *  is not detectable locally — named so the orchestrator knows it is a blind
 *  spot, not a guess. */
export function detectionHints(repo: string): string[] {
  const hints: string[] = [];
  const head = git(repo, ["symbolic-ref", "--short", "-q", "refs/remotes/origin/HEAD"]);
  if (head) hints.push(`remote default branch: ${head.replace(/^origin\//, "")}`);
  const local = (git(repo, ["for-each-ref", "--format=%(refname:short)", "refs/heads/"]) ?? "")
    .split("\n").map((s) => s.trim()).filter(Boolean);
  if (local.length) hints.push(`local branches: ${local.slice(0, 8).join(", ")}`);
  const remote = (git(repo, ["for-each-ref", "--format=%(refname:short)", "refs/remotes/origin/"]) ?? "")
    .split("\n").map((s) => s.trim()).filter((s) => s && !s.endsWith("/HEAD")).map((s) => s.replace(/^origin\//, ""));
  if (remote.length) hints.push(`origin branches: ${remote.slice(0, 8).join(", ")}`);
  hints.push("branch protection is NOT detectable locally — supply it yourself if it matters (setup help only)");
  return hints;
}

/** The one-time inquiry question — the exact phrasing the contract names. */
export function policyQuestion(repo: string, key: string): string {
  return `shipping policy for ${key}? flow (mrs|merge), baseBranches (order = MR sequence) — repo: ${repo}, config key: shipping.repos['${key}']`;
}

/** Write the pending one-time inquiry (idempotent: re-asking until the policy
 *  lands is suppressed by the existing entry). Returns the inquiry + whether
 *  it is FRESH (the caller emits the ask once). */
export function ensurePolicyInquiry(stateDir: string, repo: string, now = new Date().toISOString()): { inquiry: PolicyInquiry; fresh: boolean } {
  const s = readShippingState(stateDir);
  const key = suggestedPolicyKey(repo);
  const existing = s.inquiries[key];
  if (existing) return { inquiry: existing, fresh: false };
  const inquiry: PolicyInquiry = { key, repo, question: policyQuestion(repo, key), hints: detectionHints(repo), ts: now };
  s.inquiries[key] = inquiry;
  writeShippingState(stateDir, s);
  return { inquiry, fresh: true };
}

/** Clear a policy inquiry once the policy exists — the run RESUMES. */
export function clearPolicyInquiry(stateDir: string, repoKey: string): void {
  const s = readShippingState(stateDir);
  if (s.inquiries[repoKey]) {
    delete s.inquiries[repoKey];
    writeShippingState(stateDir, s);
  }
}

/** True when the nudge is still fresh; a STALE entry (the item was re-stamped
 *  since — a new approval cycle) is cleared and re-armed. */
function nudgeSuppressed(stateDir: string, id: string, itemUpdatedAt: string): boolean {
  const s = readShippingState(stateDir);
  const mark = s.notified[id];
  if (mark === undefined) return false;
  if (mark !== itemUpdatedAt) {
    delete s.notified[id]; // the item was re-approved — re-arm the one-time nudge
    writeShippingState(stateDir, s);
    return false;
  }
  return true;
}

function markNotified(stateDir: string, id: string, itemUpdatedAt: string): void {
  const s = readShippingState(stateDir);
  s.notified[id] = itemUpdatedAt;
  writeShippingState(stateDir, s);
}

// ---------------------------------------------------------------------------
// The shipping lane (deterministic git; merge-finisher semantics)
// ---------------------------------------------------------------------------

export interface WorkTip {
  branch: string;
  sha: string;
  source: WorkTipSource;
}

/** Which of the four resolution sources produced a WorkTip — recorded in the
 *  ship event (AUTOPILOT-85) so a silent journal miss is diagnosable after the
 *  fact instead of just "it shipped, somehow". */
export type WorkTipSource = "journal" | "finisherSource" | "runWorktree" | "branch-match";

function existingRef(repo: string, ref: string): string | null {
  return git(repo, ["rev-parse", "--verify", "--quiet", `${ref}^{commit}`]);
}

/** The ONE base ref used to test "already on base" for tip sources (c)/(d)
 *  below — the first declared baseBranch, preferring the LOCAL branch (the
 *  'merge' flow's target); falling back to the remote-tracking ref ('mrs').
 *  Sources (a) journal and (b) finisherSource are NOT gated on this — they
 *  resolve unconditionally and shipItem's own ancestor check decides
 *  already-on-base for them (see the explicit announcement in the
 *  runShippingPass skip path). */
function primaryBaseRef(repo: string, policy: ShippingRepoPolicy): string | null {
  const base = policy.baseBranches[0];
  if (!base) return null;
  if (existingRef(repo, `refs/heads/${base}`)) return `refs/heads/${base}`;
  if (existingRef(repo, `refs/remotes/origin/${base}`)) return `refs/remotes/origin/${base}`;
  return null;
}

/**
 * The approved work's tip for one item — FOUR sources, in priority order
 * (AUTOPILOT-85: the journal alone is NOT reliable — a missed preservation
 * sweep leaves a `done` item with no journal entry, and it just never ships,
 * silently, with nothing telling anyone):
 *   (a) the journal (worktree-preservation's handoffs.jsonl) — unchanged,
 *       still first: the pre-emptive capture this framework relies on, and
 *       it stays authoritative when present.
 *   (b) `item.finisherSource` — a branch/sha declared directly (queue_update
 *       validates it resolves before accepting it on a `done` item), now
 *       honoured for ANY item, not only dispatchClass 'finisher'.
 *   (c) `item.runWorktree.branch` — recorded by an earlier ACTIVE sweep
 *       (recordActiveWorktrees). Trusted ONLY when it still resolves to a
 *       commit, is NOT already an ancestor of the declared base, AND
 *       `item.runWorktree.runId === (item.runId ?? item.lastRunId)`
 *       (AUTOPILOT-85): a re-dispatch (auto-dispatch's autoRedispatch,
 *       auto-recovery, human/manual re-dispatch) assigns a fresh runId but
 *       does not itself refresh `runWorktree` — that only happens when a
 *       LATER sweep sees a LIVE worktree for the CURRENT run. A record from a
 *       superseded attempt is otherwise indistinguishable from the current
 *       run's own worktree, and trusting it would ship a FAILED attempt's
 *       divergent commits. `item.runId` is ALWAYS null on a `done` item
 *       (updateItem's terminal-transition clearing nulls it on every entry to
 *       done/human-review/failed) — the single-attempt happy path (one run,
 *       PASS, approve) needs `item.lastRunId` (stamped by that SAME clearing
 *       with the run being cleared) to recognise its own run's worktree; a
 *       re-dispatched item's `lastRunId` instead reflects whichever run LAST
 *       reached a terminal state, so a superseded attempt is still told
 *       apart. A stale record (no match, including the case where NEITHER
 *       `item.runId` NOR `item.lastRunId` is set — nothing to compare
 *       against) is never a positive resolution — it falls through to (d) and
 *       is recorded as `tipSource: "runWorktree-stale"` telemetry (`reason:
 *       "no run identity"` for the neither-set case) so an operator can see
 *       why (c) was skipped.
 *   (d) a branch named exactly `<key>` or `<key>-*`, with a tip not yet on
 *       the base — ONLY when EXACTLY ONE such branch exists. Two or more
 *       candidates is an AMBIGUITY, never a guess (NO FALLBACK, the same rule
 *       shipping policy resolution already lives by) — resolves to null,
 *       same as finding nothing.
 * Returns null when none of the four resolves — the caller emits the NO TIP
 * tick instead of silently skipping the item forever.
 */
export function resolveWorkTip(
  stateDir: string,
  item: Pick<QueueItem, "key" | "finisherSource" | "runWorktree" | "runId" | "lastRunId">,
  repo: string,
  policy: ShippingRepoPolicy,
): WorkTip | null {
  try {
    const pointers = reviewPointersFor(stateDir, item.key);
    if (pointers.length) {
      const last = pointers[pointers.length - 1]; // newest branch; its tip is the delivered state
      if (git(repo, ["cat-file", "-e", `${last.tipSha}^{commit}`]) !== null)
        return { branch: last.branch, sha: last.tipSha, source: "journal" };
    }
  } catch {
    // fall through to the other sources — a broken journal must not hide them
  }
  const declared = item.finisherSource?.trim();
  if (declared) {
    const sha = existingRef(repo, declared);
    if (sha) return { branch: declared, sha, source: "finisherSource" };
  }
  const baseRef = primaryBaseRef(repo, policy);
  const rwBranch = item.runWorktree?.branch?.trim();
  if (rwBranch) {
    // The identity of the run that produced `runWorktree`: `item.runId` when a
    // run is CURRENTLY live, else `item.lastRunId` — the run a terminal
    // transition (done/human-review/failed) most recently cleared (KEY:
    // AUTOPILOT-85; see the doc comment above). Both null/undefined (no run
    // has ever been attributed) means there is nothing to compare against, so
    // it can never match — the `&& effectiveRunId` below forces that case
    // through the rejection branch rather than an accidental
    // undefined === undefined pass.
    const effectiveRunId = item.runId ?? item.lastRunId ?? null;
    if (effectiveRunId && item.runWorktree?.runId === effectiveRunId) {
      const sha = existingRef(repo, `refs/heads/${rwBranch}`);
      if (sha && !(baseRef && isAncestor(repo, sha, baseRef)))
        return { branch: rwBranch, sha, source: "runWorktree" };
    } else {
      // AUTOPILOT-85: item.runWorktree survives a re-dispatch unchanged (only a
      // later sweep's LIVE-worktree match refreshes it), so a mismatch here
      // means this record belongs to a SUPERSEDED attempt (e.g. a review-FAILed
      // run) — trusting its branch would ship that failed attempt's divergent
      // commits. Never a positive resolution; recorded so an operator can see
      // (c) was skipped and why. `reason` distinguishes an actual identity
      // mismatch ("stale") from the case where there is no run identity to
      // compare against at all.
      appendTelemetry(
        stateDir,
        JSON.stringify({
          t: new Date().toISOString(),
          type: "ship.tip-source-rejected",
          key: item.key,
          repo,
          tipSource: "runWorktree-stale",
          reason: effectiveRunId ? "stale" : "no run identity",
          runWorktreeRunId: item.runWorktree?.runId ?? null,
          itemRunId: item.runId ?? null,
          itemLastRunId: item.lastRunId ?? null,
        }),
      );
    }
  }
  const named = (git(repo, ["for-each-ref", "--format=%(refname:short)", "refs/heads/"]) ?? "")
    .split("\n")
    .map((b) => b.trim())
    .filter(Boolean)
    .filter((b) => b === item.key || b.startsWith(`${item.key}-`));
  const fresh = named
    .map((b) => ({ branch: b, sha: existingRef(repo, `refs/heads/${b}`) }))
    .filter((c): c is { branch: string; sha: string } => !!c.sha)
    .filter((c) => !(baseRef && isAncestor(repo, c.sha, baseRef)));
  const only = fresh.length === 1 ? fresh[0] : undefined;
  if (only) return { branch: only.branch, sha: only.sha, source: "branch-match" };
  return null; // nothing resolved, or (d) was ambiguous — never guess
}

function isAncestor(repo: string, tip: string, ref: string): boolean {
  // exit 0 exactly when reachable; any other outcome (incl. 128 / unspawnable) → false
  return gitExitCode(repo, ["merge-base", "--is-ancestor", tip, ref]) === 0;
}

export interface ShipMerge {
  base: string;
  sha: string;
}

export interface ShipMr {
  /** The MR target base branch (ORDER = the declared baseBranches sequence). */
  base: string;
  /** The pushed source branch (the approved work's branch). */
  source: string;
  /** The pushed tip. */
  sha: string;
  /** MR/create URL when the remote's web layout is derivable, else null (the
   *  push ref is the pointer). */
  url: string | null;
}

export interface ShipResult {
  ok: boolean;
  flow: ShippingFlow;
  /** flow 'merge': the local base merges applied (after a successful ship). */
  merges?: ShipMerge[];
  /** flow 'mrs': one entry per baseBranch, in order. */
  mrs?: ShipMr[];
  /** true when nothing new (tip already on the base / already pushed) — a
   *  legitimate no-op, marked shipped. */
  skipped?: boolean;
  /** failure reason (never forced): conflict, checkout/base missing, no
   *  remote, non-fast-forward push, … */
  reason?: string;
  /** true when the failure was a merge conflict (the never-force case). */
  conflict?: boolean;
}

/** Best-effort MR-create URL for a known forge layout; null when the remote's
 *  web layout would be a guess (local/file remotes, unknown hosts). Never
 *  fabricated — the push ref remains the pointer. */
export function mrCreateUrl(repo: string, source: string, base: string): string | null {
  const slug = originSlug(repo);
  if (!slug) return null;
  const url = git(repo, ["remote", "get-url", "origin"]);
  if (!url) return null;
  let host = "";
  try {
    const u = new URL(url);
    if (u.protocol !== "https:" && u.protocol !== "http:") return null;
    if (u.port) return null;
    host = u.hostname;
  } catch {
    const scp = url.replace(/^[^@]+@/, "");
    const i = scp.indexOf(":");
    if (i <= 0) return null;
    host = scp.slice(0, i);
  }
  const enc = (p: string): string => encodeURIComponent(p);
  if (/gitlab/i.test(host)) {
    return `https://${host}/${slug}/-/merge_requests/new?merge_request[source_branch]=${enc(source)}&merge_request[target_branch]=${enc(base)}`;
  }
  // github + the generic compare layout
  return `https://${host}/${slug}/compare/${enc(base)}...${enc(source)}?expand=1`;
}

/** The merge-finisher ITSELF. Deterministic git, per the declared policy:
 *  flow 'merge' merges the work into the local base (default main); flow 'mrs'
 *  pushes the work to origin, one MR entry per baseBranch in the declared
 *  order. Skips when the tip is already on the base. NEVER forces: a merge
 *  conflict or a rejected push is a FAILURE with the reason — the item stays
 *  unshipped and the orchestrator is escalated to. */
export function shipItem(input: {
  repo: string;
  key: string;
  branch: string;
  sha: string;
  policy: ShippingRepoPolicy;
}): ShipResult {
  const { repo, sha, policy } = input;
  if (policy.flow === "merge") {
    const base = policy.baseBranches[0] ?? "main"; // 'merge' = merge to main (the declared flow, not a guess)
    if (git(repo, ["rev-parse", "--verify", "--quiet", `refs/heads/${base}^{commit}`]) === null) {
      return { ok: false, flow: "merge", reason: `local base branch '${base}' does not exist — fix the policy baseBranches or create the branch` };
    }
    if (isAncestor(repo, sha, `refs/heads/${base}`)) return { ok: true, flow: "merge", skipped: true }; // nothing new
    // The lane owns the base checkout — post-approval, a clean state is the
    // declared precondition (a dirty tree makes checkout/merge fail → failure).
    if (git(repo, ["checkout", "-q", base]) === null) {
      return { ok: false, flow: "merge", reason: `could not checkout '${base}' (dirty tree or missing ref) — resolve locally, then re-ship` };
    }
    if (git(repo, ["merge", "--no-ff", "-m", `ship ${input.key} (post-approval merge-finisher)`, sha]) === null) {
      git(repo, ["merge", "--abort"]); // never leave a conflicted tree around; never force the merge
      return { ok: false, flow: "merge", reason: `merge conflict on '${base}' — resolve on the branch, then re-ship`, conflict: true };
    }
    const newSha = git(repo, ["rev-parse", "--verify", "refs/heads/" + base]);
    if (!newSha) return { ok: false, flow: "merge", reason: "merge applied but base HEAD unreadable" };
    return { ok: true, flow: "merge", merges: [{ base, sha: newSha }] };
  }
  // flow 'mrs' — a remote is REQUIRED (one MR per baseBranch; no remote = failure, never a merge guess)
  if (git(repo, ["remote", "get-url", "origin"]) === null) {
    return { ok: false, flow: "mrs", reason: "policy flow is 'mrs' but the repo has no 'origin' remote — fix the remote or the policy" };
  }
  const source = input.branch; // the approved work's branch (journaled pi-parallel-*)
  const mrs: ShipMr[] = [];
  for (const base of policy.baseBranches) {
    const baseRef = `refs/remotes/origin/${base}`;
    if (git(repo, ["rev-parse", "--verify", "--quiet", `${baseRef}^{commit}`]) !== null && isAncestor(repo, sha, baseRef)) {
      continue; // already on that base — nothing new for THIS MR
    }
    // never --force: a rejected (non-fast-forward) push is a failure
    if (git(repo, ["push", "origin", `${sha}:refs/heads/${source}`]) === null) {
      return { ok: false, flow: "mrs", reason: `push of ${source} to origin rejected (non-fast-forward?) — never forcing; reconcile then re-ship` };
    }
    git(repo, ["fetch", "-q", "origin"]); // keep local tracking refs truthful
    mrs.push({ base, source, sha, url: mrCreateUrl(repo, source, base) });
  }
  if (!mrs.length) return { ok: true, flow: "mrs", skipped: true }; // nothing new anywhere
  return { ok: true, flow: "mrs", mrs };
}

// ---------------------------------------------------------------------------
// The per-sweep pass (runner wiring target)
// ---------------------------------------------------------------------------

export interface ShippingOutcome {
  /** "no-tip" (AUTOPILOT-85): none of resolveWorkTip's four sources found
   *  anything shippable for a `done` item — the item stays unshipped and the
   *  tick names the fix (set finisherSource).
   *  "ready" (AUTOPILOT-120): the autopilot-OFF notify-only mode resolved a
   *  shippable tip and announced "ready to ship" WITHOUT merging. */
  kind: "shipped" | "skipped" | "policy-inquiry" | "manual" | "failure" | "no-tip" | "ready";
  key: string;
  repo: string;
  message: string;
  event?: { name: string; data?: Record<string, unknown> };
}

function shortSha(sha: string): string {
  return sha.slice(0, 8);
}

/** AUTOPILOT-120: pass options. `notifyOnly` is the autopilot-OFF mode — the
 *  lane DETECTS + ANNOUNCES but never merges; see `runShippingPass`. */
export interface ShippingPassOptions {
  notifyOnly?: boolean;
}

/** One sweep's shipping decisions. Deterministic; the DEFAULT mode is
 *  side-effectful (git merges / pushes, store shippedAt markers, inquiry
 *  state) and returns what the runner should TICK. Never throws — a shipping
 *  failure is an outcome, never a broken sweep.
 *
 *  `notifyOnly` (AUTOPILOT-120, autopilot OFF): DETECT + ANNOUNCE only. The
 *  lane still emits its informational notices (the policy inquiry, NO TIP, and
 *  a "ready to ship" notice for a resolvable tip) but NEVER runs the
 *  merge-finisher — no merge, no push, no shippedAt write. The merge is
 *  PROACTIVE (suppressed while OFF); the notices are INFORMATIONAL (kept). */
export function runShippingPass(stateDir: string, now = new Date().toISOString(), opts: ShippingPassOptions = {}): ShippingOutcome[] {
  const cfg = loadAutopilotConfig(stateDir);
  const mode = mergeModeOf(cfg);
  const store = loadStore(stateDir);
  if (!store) return [];
  const done = Object.values(store.items)
    .filter((i) => i.status === "done" && typeof i.cwd === "string" && i.cwd.trim().length > 0 && !i.shippedAt)
    .sort((a, b) => (a.updatedAt < b.updatedAt ? -1 : 1)); // oldest first
  const outcomes: ShippingOutcome[] = [];
  // The pass runs git (merges/pushes) between decisions, so it cannot hold the
  // store's mutation window open: markers are STAGED against this snapshot and
  // merged onto the current store in one mutateStore at the end. Each patch is
  // a function of the item it lands on, so a note another writer appended while
  // git ran survives instead of being overwritten.
  const staged: Array<{ key: string; patch: (item: QueueItem) => UpdatePatch }> = [];
  for (const item of done) {
    const repo = item.cwd as string;
    const resolved = resolveRepoPolicy(stateDir, repo, cfg);
    if (!resolved) {
      // POLICY-INQUIRY: the run NOTICES + ASKS the user once (intercom flag,
      // like OTP) — nothing merges before the policy is set (NO FALLBACK).
      const { inquiry, fresh } = ensurePolicyInquiry(stateDir, repo, now);
      if (!fresh) continue; // already asked — paused, no re-ask, no guess
      const hints = inquiry.hints.length ? ` Hints: ${inquiry.hints.join(" · ")}` : "";
      outcomes.push({
        kind: "policy-inquiry",
        key: item.key,
        repo,
        message:
          `[orch-tick: ship] ${inquiry.question}. ASK THE USER directly (one-time): relay their answer and write it to ` +
          `autopilot.config.json → shipping.repos['${inquiry.key}'] = { flow, baseBranches } (e.g. { flow: "mrs", baseBranches: ["main"] } for one MR, ["dev","master"] for two). ` +
          `Nothing ships and nothing merges until the policy is set.${hints} Not a user request; respond ≤2 lines.`,
        event: { name: "orch:shipping-policy-inquiry", data: { repo, key: item.key, policyKey: inquiry.key, question: inquiry.question, hints: inquiry.hints } },
      });
      continue;
    }
    clearPolicyInquiry(stateDir, resolved.key); // the policy landed — the run resumes
    if (mode === "manual") {
      // mergeMode manual: the lane stays EXPLICIT (batch finishers). Nudge once.
      const id = `manual:${item.key}`;
      if (nudgeSuppressed(stateDir, id, item.updatedAt)) continue;
      markNotified(stateDir, id, item.updatedAt);
      const plan = resolved.policy.flow === "mrs"
        ? resolved.policy.baseBranches.map((b) => `MR into ${b}`).join(" then ")
        : `merge into ${resolved.policy.baseBranches[0] ?? "main"}`;
      outcomes.push({
        kind: "manual",
        key: item.key,
        repo,
        message:
          `[orch-tick: ship] ${item.key} is DONE and ready to ship — mergeMode is manual, so the merge-finisher stays YOUR call: ` +
          `declared policy shipping.repos['${resolved.key}'] = ${JSON.stringify(resolved.policy)} → ${plan}. Ship it explicitly (run the finisher); it will not run itself. Not a user request; respond ≤2 lines.`,
      });
      continue;
    }
    // auto lane
    const tip = resolveWorkTip(stateDir, item, repo, resolved.policy);
    if (!tip) {
      // NO TIP (AUTOPILOT-85): none of resolveWorkTip's four sources found
      // anything shippable — most commonly a missed preservation sweep. The
      // item stays DONE, unshipped, and SILENT otherwise: nothing else in the
      // lane would ever say so. Dedupe like the policy inquiry (one tick per
      // state change, keyed by updatedAt) — a genuinely stuck item does not
      // spam every sweep, but a fresh done cycle (re-approval) gets one new
      // nudge.
      const id = `no-tip:${item.key}`;
      if (nudgeSuppressed(stateDir, id, item.updatedAt)) continue;
      markNotified(stateDir, id, item.updatedAt);
      appendTelemetry(stateDir, JSON.stringify({ t: now, type: "ship.no-tip", key: item.key, repo }));
      outcomes.push({
        kind: "no-tip",
        key: item.key,
        repo,
        message: `[orch-tick: ship] NO TIP ${item.key} — nothing journaled; set finisherSource (branch/sha) on the item and the next sweep ships. Not a user request; respond ≤2 lines.`,
        event: { name: "orch:ship-no-tip", data: { key: item.key, repo } },
      });
      continue;
    }
    if (opts.notifyOnly) {
      // AUTOPILOT-120: autopilot OFF — a shippable tip was RESOLVED, so the item
      // is DONE and ready to ship. That is an INFORMATIONAL report (it starts
      // no work; see framework/tick-class.ts), so the lane announces it and
      // leaves the finisher to the operator. The MERGE itself is PROACTIVE and
      // stays suppressed: no shipItem call, no shippedAt write. Dedupe like the
      // manual-mode nudge — one notice per done cycle (a re-approval re-stamps
      // updatedAt and re-arms it).
      const id = `ready:${item.key}`;
      if (nudgeSuppressed(stateDir, id, item.updatedAt)) continue;
      markNotified(stateDir, id, item.updatedAt);
      const plan = resolved.policy.flow === "mrs"
        ? resolved.policy.baseBranches.map((b) => `MR into ${b}`).join(" then ")
        : `merge into ${resolved.policy.baseBranches[0] ?? "main"}`;
      outcomes.push({
        kind: "ready",
        key: item.key,
        repo,
        message:
          `[orch-tick: ship] ${item.key} is DONE and ready to ship — autopilot is OFF, so the merge-finisher stays YOUR call: ` +
          `declared policy shipping.repos['${resolved.key}'] = ${JSON.stringify(resolved.policy)} → ${plan}. Ship it explicitly (run the finisher); it will not run itself while OFF. Not a user request; respond ≤2 lines.`,
      });
      continue;
    }
    const result = shipItem({ repo, key: item.key, branch: tip.branch, sha: tip.sha, policy: resolved.policy });
    if (!result.ok) {
      const id = `fail:${item.key}`;
      if (nudgeSuppressed(stateDir, id, item.updatedAt)) continue; // one escalation per failure floor
      markNotified(stateDir, id, item.updatedAt);
      const neverForce = result.conflict ? " A CONFLICT — never forced." : " Shipping FAILED — never forced.";
      outcomes.push({
        kind: "failure",
        key: item.key,
        repo,
        message:
          `[orch-tick: ship] SHIPPING FAILED ${item.key} (${result.reason ?? "unknown"}).${neverForce} The item stays DONE but unshipped (no shippedAt marker — it will be retried only after you escalate). Resolve on the branch, then re-ship; response ≤2 lines.`,
        event: { name: "orch:ship-failed", data: { key: item.key, repo, reason: result.reason ?? "unknown", conflict: result.conflict ?? false } },
      });
      continue;
    }
    if (result.skipped) {
      // ALREADY ON BASE — the "direct deliverable" case, now EXPLICIT
      // (AUTOPILOT-85): the resolved tip is already an ancestor of the base, so
      // there is nothing to merge — but this is a real shipping decision, not a
      // silent no-op, and gets its own source marker + announcement
      // (previously: staged the marker with zero outcome, so a human watching
      // ticks never saw this item resolve at all).
      const skipNote = `[shipped] ${now} — already on base (tip source: ${tip.source}); nothing new to merge.`;
      staged.push({ key: item.key, patch: (fresh) => ({ shippedAt: now, notes: fresh.notes ? `${fresh.notes}\n\n${skipNote}` : skipNote }) });
      outcomes.push({
        kind: "shipped",
        key: item.key,
        repo,
        message: `[orch-tick: ship] ${item.key} already on base — direct deliverable, nothing to merge (tip source: ${tip.source}). Marked shipped. Not a user request; respond ≤2 lines.`,
        event: {
          name: "orch:item-shipped",
          data: { key: item.key, repo, flow: result.flow, policyKey: resolved.key, sha: tip.sha, source: "already-on-base", tipSource: tip.source },
        },
      });
      continue;
    }
    // SHIPPED — mark + announce. The shippedAt marker prevents re-merge.
    //
    // First, the EXPECTED-WRITE HANDSHAKE with the main-immutability guard: a
    // flow 'merge' ship moves the LOCAL base, and this sweep's guard pass
    // already took its baseline before the merge — so the guard must be told
    // about this one write or it flags its own lane on the next sweep. The
    // exemption is narrow by construction: only the exact sha this merge just
    // produced, only for this human-approved item's base (the guard itself
    // refuses anything that is not the ref's current head sitting directly on
    // the recorded baseline).
    for (const m of result.merges ?? []) recordExpectedMainWrite(stateDir, repo, m.base, m.sha);
    const sha = result.flow === "merge" ? result.merges![0].sha : tip.sha;
    const flowNote = result.flow === "merge"
      ? `merged into ${result.merges![0].base} @ ${shortSha(sha)}`
      : `MR(s): ${result.mrs!.map((m) => (m.url ? `${m.url}` : `pushed ${m.source} → ${m.base} (${shortSha(m.sha)})`)).join("; ")}`;
    const note = `[shipped] ${now} — ${flowNote} (flow ${result.flow}, policy shipping.repos['${resolved.key}'])`;
    staged.push({ key: item.key, patch: (fresh) => ({ shippedAt: now, notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note }) });
    const message = result.flow === "merge"
      ? `[orch-tick: ship] merged ${item.key} @ ${shortSha(sha)} (into ${result.merges![0].base}, policy ${resolved.key}, flow merge). Done deterministically shipped via the declared policy. Not a user request; respond ≤2 lines.`
      : `[orch-tick: ship] ${item.key} shipped — ${result.mrs!.map((m) => `MR into ${m.base}: ${m.url ?? `push ${m.source}@${shortSha(m.sha)}`}`).join("; ")} (policy ${resolved.key}, flow mrs). Not a user request; respond ≤2 lines.`;
    outcomes.push({
      kind: "shipped",
      key: item.key,
      repo,
      message,
      event: {
        name: "orch:item-shipped",
        data: {
          key: item.key,
          repo,
          flow: result.flow,
          policyKey: resolved.key,
          sha,
          tipSource: tip.source,
          ...(result.mrs ? { mrs: result.mrs.map((m) => ({ base: m.base, source: m.source, sha: m.sha, url: m.url })) } : {}),
          ...(result.merges ? { merges: result.merges } : {}),
        },
      },
    });
  }
  if (staged.length) {
    mutateStore(stateDir, (s) => {
      for (const { key, patch } of staged) {
        const fresh = s.items[key];
        if (fresh) updateItem(s, key, patch(fresh));
      }
    });
  }
  return outcomes;
}

// ---------------------------------------------------------------------------
// The one-time-answer writer (orchestrator writes the policy → the run resumes)
// ---------------------------------------------------------------------------

/** Write (or overwrite) ONE repo's policy in autopilot.config.json, preserving
 *  every other field + repo. This is the "orchestrator writes the one-time
 *  answer" path — the next sweep resolves the policy and ships. */
export function writeShippingPolicy(stateDir: string, repoKey: string, policy: ShippingRepoPolicy): void {
  const p = join(stateDir, "autopilot.config.json");
  let raw: Record<string, unknown> = {};
  try {
    if (existsSync(p)) raw = JSON.parse(readFileSync(p, "utf8")) as Record<string, unknown>;
  } catch {
    raw = {};
  }
  const shipping = (raw.shipping && typeof raw.shipping === "object" ? raw.shipping : {}) as Record<string, unknown>;
  const repos = (shipping.repos && typeof shipping.repos === "object" ? shipping.repos : {}) as Record<string, unknown>;
  repos[repoKey] = policy;
  shipping.repos = repos;
  raw.shipping = shipping;
  mkdirSync(stateDir, { recursive: true });
  const tmp = `${p}.tmp-${process.pid}`;
  writeFileSync(tmp, JSON.stringify(raw, null, 2) + "\n", "utf8");
  renameSync(tmp, p);
}