/**
 * The pending-decision ledger's grammar, evaluation and digest (#136).
 *
 * A question the orchestrator sends its operator used to live in exactly one
 * place: the model's context. A compaction, a restart, or a tick that ran long
 * lost the question *and* the fact that one was owed — after which the session
 * either asked again (the operator answers twice) or dropped it silently (the
 * decision never lands, and nothing says a decision is outstanding).
 *
 * The store holds the rows. This module holds the two things around them that
 * are not storage: what a machine-checkable precondition looks like, and how an
 * open row is rendered into a tick prompt.
 */

import type { DecisionRecord, RunRecord, Store, Tracker } from "./types.ts";
import { shellQuote } from "./shell.ts";
import { fetchRateLimit, RATE_LIMIT_COOLDOWN_MS } from "./tracker/github.ts";

/**
 * A precondition whose truth this package can check on its own.
 *
 * Exactly seven kinds, deliberately. Each one is a question the tracker or npm
 * already answers, so the row moves from "parked" to "act on this" without a
 * human re-reading it. Anything richer — a label appearing, a workflow going
 * green — is a follow-on issue rather than a grammar nobody validated.
 */
export type DecisionCondition =
  | { kind: "pr-merged"; url: string }
  | { kind: "issue-closed"; issue: number }
  | { kind: "npm-version"; spec: string }
  | { kind: "pr-checks-green"; url: string }
  | { kind: "pr-review-ready"; url: string }
  | { kind: "pr-mergeable"; url: string }
  | { kind: "rate-limit-reset" };

/** `pkg@version`, including a scoped package (`@scope/pkg@1.2.3`). */
const NPM_SPEC = /^(@[^/@\s]+\/)?[^@\s]+@[^\s]+$/;

/**
 * Check states that count as a green verdict for `pr-checks-green`, copied from
 * the daemon's failure classifier (`failure-class.ts` `SUCCESS_CHECK_STATES`):
 * both spellings are terminally successful, and anything else — a failure, a
 * still-running/pending check, a cancelled or skipped runner — is not.
 */
const GREEN_CHECK_STATES: Record<string, true> = { success: true, neutral: true };

/**
 * The run states a review revision may start from (#795, #1101) — the single
 * definition shared by `conductor_pr_review` and the `pr-review-ready` watch,
 * so the verb's gate and the condition can never name different sets.
 *
 * A revision round resumes the exact run whose row owns the PR, so the
 * revisable states are exactly the terminal runs that own the named PR: a
 * settled `pushed-green` row, a `failed` / `killed` row — a run that capped or
 * failed *after* pushing a green PR — and, since #1101, a `stopped` row that
 * pushed one. Ownership is not carried by the state alone: selection runs
 * through `runsForProjectPr`, so only a row that itself recorded the reviewed
 * PR can ever reach this gate. The PR is the durable artefact, the exact-head
 * green verification is the gate on "green at the reviewed SHA", and a
 * terminal row proves no worker is in flight, so findings are returned
 * without the close-PR → unblock → continuation dance.
 *
 * A `stopped` row has no settle sweep keeping it honest (nothing transitions a
 * stopped row when its PR merges or closes), so like `failed` / `killed` its
 * rounds re-read PR-open and green-at-head decisively before the claim — the
 * dispatch pass refuses to wake a worker against a dead or moved PR.
 *
 * Closed on purpose: a live row (`running` / `claimed`) is already doing its
 * own work, a `pushed-pending` PR is not green yet, and a `blocked` /
 * `orphaned` / `merged` row is not work returned for revision.
 */
export const REVISABLE_RUN_STATES: Record<string, true> = {
  "pushed-green": true,
  failed: true,
  killed: true,
  stopped: true,
};

/**
 * How far back a PR's project-owned run may be looked up. PR verbs and the
 * `pr-review-ready` watch share this bound: a pull request older than it is
 * not one an orchestrator is mid-flight on.
 */
export const PR_LOOKUP_WINDOW_MS = 30 * 24 * 60 * 60_000;

/**
 * One predicate for "a run-owned pull request can be returned for revision",
 * read by both `conductor_pr_review` and the `pr-review-ready` watch so they
 * cannot drift.
 *
 * Ownership is the newest attempt of this project that recorded the PR within
 * the recent-history window (`runsForProjectPr`), never the issue's newest
 * row — a requeued continuation must not hide the PR its predecessor opened
 * (#434). A `stopped` row is transparent to that selection (#870): stopping
 * is terminal and the attempt will never touch the PR again, so a stopped
 * duplicate must not shadow the older revisable owner of the same PR — the
 * row ordering that stranded PR #870's blocking findings. Every other
 * non-revisable state still decides as the newest owner: a live row
 * (`running` / `claimed`) hides an older settled one because a worker is in
 * flight (#844 — PR #838 was green while its newest owner was still
 * `running`, and a watch keyed only to checks woke the orchestrator before
 * `conductor_pr_review` was actionable), `pushed-pending` checks are still
 * settling, `blocked` may resume, `orphaned` is reconciled back to live at
 * startup, and `merged` means the PR lifecycle is over. When every row of the
 * history is `stopped`, the newest one answers — and since #1101 it answers
 * `ready` when it owns the named PR: stopping a worker that had already pushed
 * must leave the PR reviewable rather than stranded.
 *
 * `no-owner` and `not-revisable` both fail closed: a review can never act, so
 * a watch must not wake, even when the checks are green.
 */
export type PrReviewReadiness =
  | { kind: "ready"; run: RunRecord }
  | { kind: "no-owner" }
  | { kind: "not-revisable"; run: RunRecord };

export function prReviewReadiness(store: Store, project: string, prUrl: string, now: number): PrReviewReadiness {
  const runs = store.runsForProjectPr(project, prUrl, now - PR_LOOKUP_WINDOW_MS);
  const owner = runs.find((attempt) => attempt.state !== "stopped") ?? runs.at(0);
  if (owner === undefined) return { kind: "no-owner" };
  return REVISABLE_RUN_STATES[owner.state] === true ? { kind: "ready", run: owner } : { kind: "not-revisable", run: owner };
}

/**
 * Parse a raw condition, or `undefined` when it is not one of the three forms.
 *
 * Parsing on read rather than on write is what keeps a row this build does not
 * understand *listable*: the raw string is stored verbatim, so a future grammar
 * does not make an old row unloadable, and an unparseable one simply never
 * reports itself met.
 */
export function parseCondition(raw: string): DecisionCondition | undefined {
  const text = raw.trim();
  const at = text.indexOf(":");
  if (at <= 0) return undefined;
  const kind = text.slice(0, at);
  const rest = text.slice(at + 1).trim();
  if (rest.length === 0) return undefined;

  if (kind === "pr-merged" || kind === "pr-checks-green" || kind === "pr-review-ready" || kind === "pr-mergeable") {
    return rest.startsWith("https://") ? { kind, url: rest } : undefined;
  }
  if (kind === "issue-closed") {
    if (!/^\d+$/.test(rest)) return undefined;
    const issue = Number.parseInt(rest, 10);
    return issue > 0 ? { kind: "issue-closed", issue } : undefined;
  }
  if (kind === "npm-version") {
    return NPM_SPEC.test(rest) ? { kind: "npm-version", spec: rest } : undefined;
  }
  if (kind === "rate-limit-reset") {
    // Spelled `rate-limit-reset:github`: the grammar requires a non-empty rest,
    // and `github` names the one GraphQL provider the fleet is on.
    return rest === "github" ? { kind: "rate-limit-reset" } : undefined;
  }
  return undefined;
}

/** The seven accepted forms, for a refusal that can be acted on in one turn. */
export const CONDITION_FORMS = [
  "pr-merged:https://github.com/owner/repo/pull/123",
  "issue-closed:123",
  "npm-version:omp-conductor@0.4.3",
  "pr-checks-green:https://github.com/owner/repo/pull/123",
  "pr-review-ready:https://github.com/owner/repo/pull/123",
  "pr-mergeable:https://github.com/owner/repo/pull/123",
  "rate-limit-reset:github",
] as const;

/** Probe for `npm-version`, injectable so tests never reach the network. */
export type NpmProbe = (spec: string) => Promise<boolean>;

/**
 * `npm view <spec> version` — exit 0 with non-empty output means the version is
 * published. Bounded at 10 s because this runs inside a tick: a registry that
 * hangs must cost one unevaluated condition, not the tick.
 */
export const probeNpmVersion: NpmProbe = async (spec) => {
  const proc = Bun.spawn(["npm", "view", spec, "version"], { stdout: "pipe", stderr: "ignore" });
  const timer = setTimeout(() => {
    proc.kill();
  }, 10_000);
  try {
    const [text, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]);
    return code === 0 && text.trim().length > 0;
  } catch {
    return false;
  } finally {
    clearTimeout(timer);
  }
};

/** Probe for `rate-limit-reset`, injectable so tests never reach the network. */
export type RateLimitProbe = () => Promise<boolean>;

/** The `gh`-shaped runner {@link probeRateLimitReset} drives (#349). */
type RateLimitGh = (argv: string[], stdin?: string) => Promise<string>;

/**
 * Run `gh api rate_limit` under a 10 s bound — a hung `gh` must cost one
 * unevaluated decision, never the tick, the same reason as
 * {@link probeNpmVersion}.
 */
async function ghRateLimit(argv: string[]): Promise<string> {
  const proc = Bun.spawn(["gh", ...argv], { stdout: "pipe", stderr: "pipe" });
  const timer = setTimeout(() => {
    proc.kill();
  }, 10_000);
  try {
    const [text, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]);
    if (code !== 0) throw new Error(`gh rate_limit exited ${code}`);
    return text;
  } finally {
    clearTimeout(timer);
  }
}

/**
 * `gh api rate_limit` reports a budget on every GitHub resource the app's
 * token serves from, and this probe is met only when **both** buckets have
 * quota left. REST `core` exhaustion alongside a still-warm GraphQL budget is
 * the exact false positive this probe used to raise (#349), so one warm bucket
 * must never stand in for recovery on the resource that was actually refused.
 *
 * This is half the recovery evidence. The enforced half — no rate refusal on
 * the tracker's recent breaker ledger — is applied where the store is in view,
 * in {@link evaluateDecisionConditions}. `rate_limit` alone can report quota
 * an app would still refuse, so the probe answers a necessary condition, not a
 * sufficient one.
 */
export async function probeRateLimitReset(runGh: RateLimitGh = ghRateLimit): Promise<boolean> {
  const status = await fetchRateLimit(runGh);
  return status !== undefined && status.graphql.remaining > 0 && status.core.remaining > 0;
}

/**
 * Check every open decision that carries a condition; a met row is only
 * revisited when its condition can stop holding (`pr-checks-green`, #808).
 *
 * Errors are swallowed per row on purpose: this runs fire-and-forget beside a
 * tick, and one deleted PR or one flaky `gh` call must not stop the rest of the
 * pass. A condition that could not be checked is simply not met this time, and
 * the next tick asks again for free.
 *
 * A watch (#459) whose PR condition can no longer be observed — the PR merged
 * or closed before the requested condition was seen — is withdrawn in this same
 * pass rather than left rendering `condition:pending` forever: the mediated
 * surfaces for checks and mergeability do not answer for a settled PR, so the
 * row could only ever stay unmet, and a watch has deliberately no seven-day
 * expiry to close it. The withdrawal is the same durable `state=withdrawn`
 * resolution an operator's `watch withdraw` writes, so the audit trail of what
 * was waiting and why it stopped survives. Operator questions are untouched:
 * they answer to a human and expire on the seven-day clock.
 *
 * A met `pr-checks-green` / `pr-review-ready` row is the exception to "met is
 * final" (#808): both are verdicts *about a commit*, so the met state is bound
 * to the head it was observed at and re-read here. A head change drops the met
 * state (and with it the binding) and the row re-evaluates against the live
 * head — it cannot render `[CONDITION MET]` for a head whose checks are still
 * running or have gone red. A row met before head binding existed has no
 * binding to compare and is preserved as met as-is: nothing is guessed for it,
 * and it is never withdrawn with the never-met reason. Every other condition
 * is monotone or human-answered and keeps its permanent met transition.
 *
 * Returns the rows that just became met, so the caller can log what changed
 * rather than a count.
 */
export async function evaluateDecisionConditions(
  store: Store,
  project: string,
  tracker: Tracker,
  probes: { npm: NpmProbe; rateLimit: RateLimitProbe },
  now: () => number,
): Promise<DecisionRecord[]> {
  const met: DecisionRecord[] = [];
  for (const decision of store.openDecisions(project)) {
    if (decision.condition === undefined) continue;
    const condition = parseCondition(decision.condition);
    if (condition === undefined) continue;
    if (decision.conditionMetAt !== undefined) {
      // A met row is normally final: the digest's "act on this now" is the
      // whole point, and flapping a verdict would reopen a question the
      // orchestrator may already be acting on. The two head-bound conditions
      // — `pr-checks-green` and `pr-review-ready` — are the ones whose
      // observation can stop holding: they are about a commit, and a PR head
      // change makes the old verdict stale.
      if (condition.kind !== "pr-checks-green" && condition.kind !== "pr-review-ready") continue;
      // A row met before head binding existed has no binding to compare and no
      // way to prove a head moved. It is preserved as met rather than cleared
      // and re-verified: a clear would let the settled-PR path below withdraw
      // it as "before its checks were seen", which the durable ledger itself
      // contradicts (review round 2, #808). No head is ever guessed for it.
      if (decision.conditionHead === undefined) continue;
      let head: string | undefined;
      try {
        head = await tracker.prHead(condition.url);
      } catch {
        continue;
      }
      // An unreadable head is not proof a head changed: flip the row only on a
      // definitive observation, or it flaps on every flaky `gh` call.
      if (head === undefined) continue;
      if (decision.conditionHead === head) {
        // A stable head is the whole story for `pr-checks-green` — checks are
        // a verdict about a commit, and the commit did not move. For
        // `pr-review-ready` it is only half: the row also gates on which run
        // owns the PR, and a review revision resumes the owner to live
        // (`running` / `claimed`) without moving the head. So a met
        // review-ready row falls through to the branch below, whose shared
        // readiness gate clears it the moment the verb can no longer accept —
        // the repeated wake this condition exists to prevent (#844 review 1).
        if (condition.kind === "pr-checks-green") continue;
      } else {
        // The head moved — or the row predates head binding — so the old met
        // observation is no longer the PR's state. Clear it and let the pass
        // below judge the live head: still green, the row re-marks with the
        // new binding and wakes the orchestrator; pending or red, it stays
        // pending until it is not.
        store.clearDecisionConditionMet(decision.id);
      }
    }
    let satisfied = false;
    let satisfiedHead: string | undefined;
    try {
      if (condition.kind === "pr-merged") {
        const state = await tracker.prState(condition.url);
        if (state === "merged") {
          satisfied = true;
        } else if (decision.kind === "watch" && state === "closed") {
          // The PR closed without merging: `pr-merged` can never be observed.
          store.resolveDecision(decision.id, "withdrawn", "PR closed without merging", now());
          continue;
        }
      } else if (condition.kind === "issue-closed") {
        satisfied = (await tracker.issueState(condition.issue)) === "closed";
      } else if (condition.kind === "npm-version") {
        satisfied = await probes.npm(condition.spec);
      } else if (condition.kind === "pr-checks-green") {
        // A watch whose PR settled before its checks were seen green cannot
        // ever be met: the checks surface does not answer for a merged or
        // closed PR, so the row would render `pending` forever (#664). A row
        // that was already met — then had its head move and its PR settle
        // before the new head's checks — did see its checks green at the
        // earlier head, and its withdrawal must record that, or the durable
        // ledger gains a fact it contradicts (review round 2, #808).
        if (decision.kind === "watch") {
          const state = await tracker.prState(condition.url);
          if (state === "merged" || state === "closed") {
            const seenGreen = decision.conditionMetAt !== undefined;
            store.resolveDecision(
              decision.id,
              "withdrawn",
              seenGreen ? `PR ${state} after its checks were seen green` : `PR ${state} before its checks were seen`,
              now(),
            );
            continue;
          }
        }
        // The head anchor is read BEFORE the checks: a verdict is only ever
        // bound to a head that did not move while the checks were read, and
        // the anchor-first order is what closes the push race in both
        // directions. Reading the checks first would let a push that landed
        // after the checks answered bind the old head's green verdict to the
        // new, untested commit — both confirming head reads would agree on a
        // head the checks were never read for (review round 1, #808).
        const before = await tracker.prHead(condition.url);
        // The same conclusion values the daemon's failure classifier treats as
        // a green verdict (`success` / `neutral`, lowercased): a non-empty list
        // in which every check is terminally successful and none is failing or
        // pending (#189).
        const checks = await tracker.checkConclusions(condition.url);
        if (
          before !== undefined &&
          checks.length > 0 &&
          checks.every((c) => GREEN_CHECK_STATES[c.state.trim().toLowerCase()] === true)
        ) {
          // Confirm the anchor still holds after the checks read: a push that
          // landed anywhere inside the window surfaces as a mismatch, so the
          // green list is bound only to a head it was actually read for. A
          // list without a readable, stable anchor is not yet met — an
          // unbound verdict is exactly the staleness this fixes, and the next
          // tick can ask again for free.
          const after = await tracker.prHead(condition.url);
          if (before === after) {
            satisfied = true;
            satisfiedHead = before;
          }
        }
      } else if (condition.kind === "pr-review-ready") {
        // A watch whose PR settled before it was seen review-ready can never
        // be met: neither the checks surface nor a revisable run answers for
        // a merged or closed PR, so the row would render `pending` forever
        // (#664). A row that was already met — then had its head move and its
        // PR settle before the new head's checks — did see its PR ready at
        // the earlier head, and its withdrawal must record that, the same
        // history-preserving convention as pr-checks-green (review round 2,
        // #808).
        if (decision.kind === "watch") {
          const state = await tracker.prState(condition.url);
          if (state === "merged" || state === "closed") {
            const seenReady = decision.conditionMetAt !== undefined;
            store.resolveDecision(
              decision.id,
              "withdrawn",
              seenReady ? `PR ${state} after it was review-ready` : `PR ${state} before it was review-ready`,
              now(),
            );
            continue;
          }
        }
        // The store-owned half of review readiness, shared with the verb
        // (#844): the newest project-owned run that recorded this PR must be
        // in a revisable state (`pushed-green` / `failed` / `killed`). A PR
        // with no owner, or whose newest owner is live (`claimed` / `running`
        // — the original worker or an earlier review revision) or otherwise
        // non-revisable, stays pending even while its checks are green: the
        // watch must not wake the orchestrator before `conductor_pr_review`
        // can legally act.
        const readiness = prReviewReadiness(store, project, condition.url, now());
        if (readiness.kind !== "ready") {
          // A met row whose owner stopped being revisable — the revision
          // dispatched and the owner is live again (`running` / `claimed`)
          // or the owner ended non-revisable — returns to pending on the
          // same tick: the watch exists precisely so the orchestrator is
          // not woken while `conductor_pr_review` would refuse. The clear
          // is idempotent, and a revision that settles at the same head
          // re-marks the row on a later tick.
          if (decision.conditionMetAt !== undefined) store.clearDecisionConditionMet(decision.id);
          continue;
        }
        // The tracker half, with the same fail-closed exact-head anchor as
        // pr-checks-green (#808): the current head must be readable, every
        // current check terminally green, and the head must not move while
        // the checks were read. An unreadable or unstable head is not yet
        // ready — never a verdict fabricated from a head the checks were not
        // read for.
        const before = await tracker.prHead(condition.url);
        const checks = await tracker.checkConclusions(condition.url);
        if (
          before !== undefined &&
          checks.length > 0 &&
          checks.every((c) => GREEN_CHECK_STATES[c.state.trim().toLowerCase()] === true)
        ) {
          const after = await tracker.prHead(condition.url);
          if (before === after) {
            satisfied = true;
            satisfiedHead = before;
          }
        }
      } else if (condition.kind === "pr-mergeable") {
        // A watch whose PR settled before it was seen mergeable is the same
        // unobservable case as checks: the mergeability literal exists only
        // while the PR is open (#664).
        if (decision.kind === "watch") {
          const state = await tracker.prState(condition.url);
          if (state === "merged" || state === "closed") {
            store.resolveDecision(decision.id, "withdrawn", `PR ${state} before it was seen mergeable`, now());
            continue;
          }
        }
        // `clean` is the tracker's "this PR can merge" literal; `unknown` and a
        // conflict are both unsatisfied (#189).
        satisfied = (await tracker.mergeable(condition.url)) === "clean";
      } else {
        // `rate-limit-reset` (#349). The probe above (both GitHub buckets warm)
        // is only half the evidence; recovery is proven by the enforced refusal
        // ledger — no rate refusal inside the breaker's cooldown window.
        // `/rate_limit` alone can report quota an app would still refuse, which
        // is exactly the false positive this fixes, so a recent refusal on the
        // tracker's breaker keeps the row unmet. A store that cannot report
        // refusals cannot prove the enforced half, so it fails closed too.
        const refusals = store.ghRefusalsSince?.(now() - RATE_LIMIT_COOLDOWN_MS);
        satisfied = (await probes.rateLimit()) && refusals !== undefined && refusals.count === 0;
      }
    } catch {
      continue;
    }
    if (!satisfied) continue;
    if (store.markDecisionConditionMet(decision.id, now(), satisfiedHead)) {
      met.push({
        ...decision,
        conditionMetAt: now(),
        ...(satisfiedHead === undefined ? {} : { conditionHead: satisfiedHead }),
      });
    }
  }
  return met;
}

/** Whole hours, then days — the granularity an operator reads a digest at. */
function age(since: number, now: number): string {
  const minutes = Math.max(0, Math.round((now - since) / 60_000));
  if (minutes < 60) return `${minutes}m`;
  const hours = Math.round(minutes / 60);
  return hours < 48 ? `${hours}h` : `${Math.round(hours / 24)}d`;
}

/**
 * The open-decision block appended to a tick prompt, or `""` when nothing is
 * owed.
 *
 * This is the half that makes the ledger worth having: the session is told what
 * it still owes on every tick, from the store, so "what did I ask them?" stops
 * being a memory question. A row whose condition has come true is flagged, not
 * merely listed — that is the difference between a parked question and one to
 * act on now.
 *
 * The two kinds render under their own headings (#459). Only rows a human must
 * answer — `kind === "question"`, whatever condition they carry — appear under
 * the operator heading; a watch the orchestrator set for itself goes under the
 * watch heading with no instruction to resolve it, so a fleet at rest behind
 * GitHub's checks is never mistaken for a fleet that is waiting on its
 * operator. A met watch still surfaces to the orchestrator with its note and
 * the same `[CONDITION MET]` flag a met question gets. Every watch line also
 * names the command that closes it (`watch withdraw <id> --project <project>`),
 * because the surface that shows a watch is where the reader learns how to end
 * one (#664). The row's own project is always named — a digest is per-project,
 * and a bare `watch withdraw <id>` is ambiguous on a host with several (#810).
 */
export function formatDecisionDigest(open: readonly DecisionRecord[], now = Date.now()): string {
  const questions = open.filter((d) => d.kind !== "watch");
  const watches = open.filter((d) => d.kind === "watch");
  const lines: string[] = [];
  if (questions.length > 0) {
    lines.push(
      `Open operator decisions (${questions.length}) — resolve or withdraw each with omp-conductor decision resolve|withdraw <id>:`,
    );
    for (const d of questions) {
      lines.push(decisionLine(d, now));
    }
  }
  if (watches.length > 0) {
    lines.push(
      `Watches (${watches.length}) — conditions the orchestrator set for itself; no operator action needed:`,
    );
    for (const d of watches) {
      // Each line names the verb that closes it, qualified by the row's stored
      // project so a copied command runs verbatim on a multi-project host
      // (#810). The rows are project-scoped by construction; the stored
      // project is the one the command must name, shell-quoted because a
      // project name may contain characters a shell would read.
      lines.push(decisionLine(d, now, `watch withdraw ${d.id} --project ${shellQuote(d.project)}`));
    }
  }
  return lines.join("\n");
}

/**
 * One digest row: id, age, what it blocks, the met flag, and the note.
 * `closing` names the command that ends the row, rendered after the note.
 */
function decisionLine(d: DecisionRecord, now: number, closing?: string): string {
  const flag = d.conditionMetAt === undefined ? "" : " [CONDITION MET — act on this now]";
  const end = closing === undefined ? "" : ` — end with: omp-conductor ${closing}`;
  return `- ${d.id} (${age(d.askedAt, now)}, blocks ${d.blocks ?? "nothing"})${flag}: ${d.question}${end}`;
}