// tick-router.ts — the tick DELIVERY GATE, host-agnostic. Both hosts (the pi
// extension + the opencode plugin) route their ticks through this so the
// machinery is IDENTICAL: same gate conditions (interactive/loaded/busy/
// compacting), same cooldown, same message format. The host supplies what
// each state means in that tool + the deliver mechanism (pi: sendMessage;
// opencode: promptAsync injection).
//
// The MESSAGE is delivered as-is — core.ts's ticks already carry the
// `[orch-tick: <reason>]` prefix (dispatch | intake | review | failure |
// budget); hosts must not add another. Runner-crafted ticks (decision,
// harness, main-write, recover, ship) go through the same gate with the same
// marker.
//
// AUTOPILOT-87 — DELIVERY-TIME PREMISE. A tick is composed when an event or a
// sweep decides to nudge; if the host is busy it is HELD and delivered later,
// by which time the store may have moved on (an item merged, blocked,
// re-dispatched). Delivering it verbatim asks the orchestrator to act on a
// state that no longer exists — each costs a turn to refute, and a stale
// FAIL-shaped notice actively invites re-dispatching merged work. So at
// DELIVERY (this gate, the single choke point every tick passes through) a
// tick whose PREMISE no longer holds is DROPPED, with one `tick.dropped-stale`
// telemetry line. A missed tick is recoverable at the next sweep; a stale one
// trains the reader to ignore ticks, which is worse.
//
// Premises (the sender declares what its tick claims; `premiseForTick` maps
// the engine's Tick to one):
//   - status   — every named item must still be in one of `statuses` (review
//                ticks: the item is still ai-review/human-review; a PARKED
//                recover tick: the item is still blocked);
//   - failure  — the item is still `failed` (or `active` with the SAME
//                runId) — a failure/recover notice that outlived its failure;
//   - dispatch — ready > 0 AND a slot is free;
//   - always   — ship policy-inquiries, main-write violations, harness/config
//                info: NEVER dropped (a human question or a violation must
//                always reach the reader).
// A non-`always` premise with no store view is UNVERIFIABLE and is dropped
// (the judgment call: prefer a missed tick to a stale one).

import type { Tick } from "../types.ts";
import { tickReasonFromMessage, logTickDrop } from "./tick-telemetry.ts";

export interface TickHostState {
  /** A live orchestrator session exists (pi: interactive TUI; opencode: a
   *  persistent session, NOT a headless one-shot run). */
  interactive(): boolean;
  /** The orchestrator context is loaded (pi: /orchestrate injected; opencode:
   *  the plugin IS the orchestrator context — always true). */
  loaded(): boolean;
  /** A tick must wait: the agent is mid-turn, OR a turn is about to start. The
   *  pi host reports the second case too (AUTOPILOT-122) — a user prompt inside
   *  pi core's pre-run window, where injecting a triggered turn would claim the
   *  agent and the user's just-submitted message would be refused (lost).
   *  Deferring HERE is what holds such a tick without recursing through the
   *  host's own delivery surface. */
  busy(): boolean;
  /** Compaction in progress (pi only; opencode reports false). */
  compacting(): boolean;
}

/** The live store state a premise is re-checked against. Supplied by the
 *  runner (it reads queue.json); a fake in tests. */
export interface TickStoreSnapshot {
  /** item key → status; an ABSENT key means the item no longer exists. */
  status: Record<string, string | undefined>;
  /** item key → its active run id (null when none). */
  runId: Record<string, string | null>;
  /** Keys currently in ai-review / human-review (the review-state signature). */
  aiReview: string[];
  humanReview: string[];
  /** Dispatch-ready items and free fleet slots (the `dispatch` premise). */
  ready: number;
  slotsFree: number;
}

export interface TickStoreView {
  /** The LIVE store's premise facts, or `undefined` when there is no store to
   *  read at all. "No store" is UNVERIFIABLE, not "an empty store": the gate
   *  drops a non-`always` tick with `unverifiable-no-store` rather than
   *  reading every named item as gone (AUTOPILOT-87). */
  snapshot(): TickStoreSnapshot | undefined;
}

/** What a tick CLAIMS about the store, re-verified at delivery. */
export type TickPremise =
  | { kind: "status"; keys: string[]; statuses: string[]; dedupeByReviewState?: boolean }
  | { kind: "failure"; key: string; runId?: string | null }
  | { kind: "dispatch" }
  | { kind: "always" };

export interface TickDeliveryMeta {
  premise?: TickPremise;
  /** Composition time (ms) — the `ageMs` in the drop telemetry. */
  generatedAt?: number;
}

export interface TickSendOptions extends TickDeliveryMeta {
  bypassCooldown?: boolean;
}

export interface TickRouterOptions {
  cooldownMs?: number;
  /** Live-store reader for the delivery-time premise re-check. */
  store?: TickStoreView;
  /** Where to append `tick.dropped-stale`. Absent = no telemetry. */
  stateDir?: string;
  now?: () => number;
}

/** The premise a tick's FACTS imply (the engine's own ticks; runner-crafted
 *  one-line ticks pass an explicit premise). Review ticks carry the state they
 *  were produced from — a PASS is produced in human-review, a FAIL-at-cap in
 *  failed, a spec pass in a proposal, the plural nudge in the live review
 *  sets. A tick with nothing to re-check is `always` (never dropped). */
export function premiseForTick(tick: Tick): TickPremise {
  const f = tick.facts;
  const key = typeof f.key === "string" && f.key ? f.key : null;
  if (tick.reason === "dispatch") return { kind: "dispatch" };
  if (tick.reason === "failure") {
    return key ? { kind: "failure", key, runId: typeof f.runId === "string" ? f.runId : null } : { kind: "always" };
  }
  if (tick.reason === "review") {
    const ai = asKeys(f.aiReview);
    const human = asKeys(f.humanReview);
    if ((ai.length || human.length) && !key) {
      // The SET-based nudge: fires at most once per (ai-review set,
      // human-review set) change — deduped at delivery against the LIVE sets
      // (not a hash that includes a changing field like the run ledger).
      return { kind: "status", keys: [...new Set([...ai, ...human])], statuses: ["ai-review", "human-review"], dedupeByReviewState: true };
    }
    if (!key) return { kind: "always" };
    if (f.specPass) return { kind: "status", keys: [key], statuses: ["proposal"] };
    if (f.verdict === "PASS") return { kind: "status", keys: [key], statuses: ["human-review", "ai-review"] };
    if (f.verdict === "FAIL" && f.cap !== undefined) return { kind: "status", keys: [key], statuses: ["failed"] };
    if (f.verdict === "FAIL") return { kind: "status", keys: [key], statuses: ["active"] };
    if (f.failureOverridden) return { kind: "status", keys: [key], statuses: ["ai-review"] };
    // Unparseable-verdict / generic single-item review nudge: the item must
    // still be somewhere in the review lane.
    return { kind: "status", keys: [key], statuses: ["ai-review", "human-review"] };
  }
  return { kind: "always" };
}

/** The key a pending tick coalesces on ("multiple pending ticks for the same
 *  key collapse to the newest"). Keyless ticks return null. */
export function premiseCoalesceKey(premise: TickPremise | undefined): string | null {
  if (!premise) return null;
  if (premise.kind === "failure") return premise.key;
  if (premise.kind === "status" && premise.keys.length === 1) return premise.keys[0];
  return null;
}

function asKeys(v: unknown): string[] {
  return Array.isArray(v) ? v.filter((x): x is string => typeof x === "string" && x.length > 0) : [];
}

interface Verdict {
  deliver: boolean;
  key: string | null;
  why: string;
}

/** Re-read the live store and decide whether the premise still holds. */
function evaluatePremise(premise: TickPremise, snap: TickStoreSnapshot | undefined): Verdict {
  if (premise.kind === "always") return { deliver: true, key: null, why: "" };
  if (!snap) return { deliver: false, key: premiseKey(premise), why: "unverifiable-no-store" };
  if (premise.kind === "dispatch") {
    if (snap.ready > 0 && snap.slotsFree > 0) return { deliver: true, key: null, why: "" };
    return { deliver: false, key: null, why: `no-capacity (ready ${snap.ready}, free ${snap.slotsFree})` };
  }
  if (premise.kind === "failure") {
    const st = snap.status[premise.key];
    if (st === undefined) return { deliver: false, key: premise.key, why: "item-gone" };
    if (st === "failed") return { deliver: true, key: premise.key, why: "" };
    if (st === "active" && premise.runId && snap.runId[premise.key] === premise.runId) {
      return { deliver: true, key: premise.key, why: "" };
    }
    return { deliver: false, key: premise.key, why: `item-${st}` };
  }
  for (const k of premise.keys) {
    const st = snap.status[k];
    if (st === undefined) return { deliver: false, key: k, why: "item-gone" };
    if (!premise.statuses.includes(st)) return { deliver: false, key: k, why: `item-${st}` };
  }
  return { deliver: true, key: premise.keys[0] ?? null, why: "" };
}

function premiseKey(premise: TickPremise): string | null {
  if (premise.kind === "failure") return premise.key;
  if (premise.kind === "status") return premise.keys[0] ?? null;
  return null;
}

export function createTickRouter(
  host: TickHostState,
  deliver: (message: string, meta?: TickDeliveryMeta) => void,
  options: TickRouterOptions = {},
): {
  send(message: string, opts?: TickSendOptions): "delivered" | "dropped" | "deferred";
  /** AUTOPILOT-87: undo the review-state dedupe commit for a send the gate
   *  accepted synchronously but whose triggered turn then failed (pi's
   *  settled-vs-teardown race). See `send`'s commit rule. */
  invalidateReviewState(message?: string): void;
} {
  const cooldownMs = options.cooldownMs ?? 1500;
  const now = options.now ?? (() => Date.now());
  let lastSentAt = 0;
  // The last review STATE delivered — the dedupe key for the set-based review
  // nudge (AUTOPILOT-87): an unchanged (ai-review, human-review) pair is the
  // same state, so its nudge fires at most once per state change.
  let lastReviewStateKey: string | null = null;
  // The message that committed `lastReviewStateKey`, so an async rejection can
  // invalidate exactly that commit (never a later, genuinely delivered one).
  let lastReviewStateMessage: string | null = null;
  const emitDrop = (message: string, key: string | null, generatedAt: number | undefined, why: string): void => {
    if (!options.stateDir) return;
    logTickDrop({
      stateDir: options.stateDir,
      reason: tickReasonFromMessage(message),
      key,
      ageMs: generatedAt !== undefined ? now() - generatedAt : 0,
      why,
    });
  };
  return {
    /** Apply the shared gate; deliver when it passes.
     *  "delivered" — the host accepted the message.
     *  "dropped"   — PERMANENT (interactive/loaded) or a cooldown nudge, or a
     *                stale/duplicate tick whose premise no longer holds: the
     *                nudges re-fire on the next sweep; do not hold.
     *  "deferred"  — TRANSIENT (busy/compacting, or the host's deliver threw —
     *                the runtime rejected the send in the busy-not-streaming
     *                window). The runner holds it + flushes at the host's
     *                settle/idle event — the premise is re-checked THEN, at
     *                the actual delivery, not now. */
    send(message: string, opts?: TickSendOptions): "delivered" | "dropped" | "deferred" {
      if (!host.interactive() || !host.loaded()) return "dropped"; // permanent
      if (host.busy() || host.compacting()) return "deferred"; // transient — hold + flush at settle
      const t = now();
      if (!opts?.bypassCooldown && t - lastSentAt < cooldownMs) return "dropped"; // nudge — re-fires
      const premise = opts?.premise ?? { kind: "always" };
      const snap = premise.kind === "always" ? undefined : options.store?.snapshot();
      const verdict = evaluatePremise(premise, snap);
      if (!verdict.deliver) {
        emitDrop(message, verdict.key, opts?.generatedAt, verdict.why);
        return "dropped";
      }
      // State-change dedupe: only the pure set-based review nudge asks for it,
      // and it keys on the LIVE (ai-review set, human-review set) — an
      // unchanged pair is the SAME state, so its nudge fires once.
      let reviewSig: string | null = null;
      if (premise.kind === "status" && premise.dedupeByReviewState && snap) {
        reviewSig = `review-sets:${[...snap.aiReview].sort().join(",")}|${[...snap.humanReview].sort().join(",")}`;
        if (lastReviewStateKey === reviewSig) {
          emitDrop(message, verdict.key, opts?.generatedAt, "duplicate-state");
          return "dropped";
        }
      }
      try {
        deliver(message, { premise: opts?.premise, generatedAt: opts?.generatedAt });
      } catch {
        return "deferred"; // the runtime rejected the send — hold + flush at settle
      }
      // COMMIT ONLY ON CONFIRMED DELIVERY. The review-state key gates the NEXT
      // nudge, so a send the runtime did not accept must never consume it: the
      // sync throw above returned "deferred" without committing, and the runner
      // calls invalidateReviewState when a send it accepted (sync) is rejected
      // asynchronously — otherwise the flush retry is misread as a DUPLICATE and
      // the nudge never reaches the reader (AUTOPILOT-87).
      lastSentAt = t;
      if (reviewSig !== null) {
        lastReviewStateKey = reviewSig;
        lastReviewStateMessage = message;
      }
      return "delivered";
    },
    invalidateReviewState(message?: string): void {
      // Undo the dedupe commit for a send that resolved as accepted but whose
      // triggered turn then failed (pi: the settled-vs-teardown race). Match by
      // message so a LATER, genuinely-delivered state commit is left intact.
      if (message === undefined || message === lastReviewStateMessage) {
        lastReviewStateKey = null;
        lastReviewStateMessage = null;
      }
    },
  };
}