/**
 * Grooming candidate selection: the one shared rule set that answers "which
 * backlog issues may a to-spec pass be spent on right now".
 *
 * #772 shipped the strict to-spec contract (`TO_SPEC_SCHEMA` +
 * `recordToSpecGrooming`) and #777 shipped the launch half as a per-tick
 * capability token redeemed by a `task` tool-call gate inside the tick
 * extension. That made grooming tick-bound: one batch per low-queue tick, and
 * an operator message that consumed the tick consumed the batch with it
 * (#1040). This module is what remains when the token goes away — the
 * mechanical selection itself, extracted so the dispatch daemon owns launching
 * (#1041) and the tick is free to audit rather than authorize.
 *
 * Where each fact lives decides who enforces it, and every exclusion is
 * enforced mechanically from authoritative data, never by prose:
 *
 *  - Tracker facts — the open-issue pool, the park label, the parent/epic
 *    probe — are read through the existing Tracker adapter
 *    (`listOpenIssues`/`childrenOf`) via {@link ToSpecTrackerSeam}, and turned
 *    into the pool by {@link toSpecPoolFromSnapshot}. If the snapshot cannot be
 *    read, no batch is selected at all: the selection fails closed rather than
 *    trusting a model to self-filter parked/parent/epic candidates.
 *  - Store facts — durable grooming rows (#735), admission's lane/dependency
 *    holds, active runs, in-flight launches — are enforced by
 *    {@link toSpecCandidateExclusion}, the single predicate every reader of
 *    "is this issue already groomed?" goes through.
 *  - The live source ref for each item cannot be known synchronously without a
 *    fresh per-repo read, so the launcher stamps the observation time it
 *    launched at and the strict parser's 24h freshness ceiling re-vets it at
 *    persistence time — the boundary #772 set, tightened by #1000.
 *
 * The module deliberately imports nothing from the tick extension, the harness
 * or the pi surface: both the daemon and the tick read it, and a selection rule
 * that could only run inside a session is how grooming became tick-bound in the
 * first place.
 */

import { repoSlugFor } from "./gitops.ts";
import {
  parseToSpecEvidence,
  parseToSpecFailureEvidence,
  TO_SPEC_MAX_SOURCE_AGE_MS,
  type ToSpecFailure,
  type ToSpecResult,
} from "./to-spec.ts";
import type {
  DispatchSummary,
  GroomingRecord,
  GroomTrigger,
  IssueState,
  ProjectConfig,
  ReadyIssue,
} from "./types.ts";

/** Routable candidates below which the grooming duty fires (#181). */
export const DEFAULT_GROOM_BELOW = 4;

/**
 * Whether the count-based half of the grooming duty fires for one dispatch
 * row's routable count (#988): below a numeric threshold as always, while
 * `"always"` leaves this gate open on purpose — the selectors then answer
 * "is anything left to groom?" from candidate state (ungroomed, unrefused,
 * unparked), and the caller renders the truthful finding either way. Never
 * substitute a large number for `"always"`: a big queue must not silence the
 * duty.
 */
export function groomingDue(routed: number, groomBelow: GroomTrigger): boolean {
  return groomBelow === "always" || routed < groomBelow;
}

/** The `to-spec` agent every batch item runs under (shipped in
 *  `omp/agents/to-spec.md`, discovered through OMP's native task-agent
 *  discovery — never a custom process runtime). */
export const TO_SPEC_AGENT = "to-spec";

/** The maximum number of candidates one batch may carry (#679's "small
 *  per-tick candidate limit"; the daemon's own concurrency bounds the
 *  sessions underneath). */
export const TO_SPEC_BATCH_MAX = 3;

/** The grooming-table verdict row a launched-but-unfinished batch leaves behind
 *  (reason, on a `blocked` verdict): the durable in-flight marker that stops
 *  the next pass — or a restarted daemon — from re-launching the same item.
 *  `blocked` is deliberate: `recordToSpecGrooming` replaces the row when the
 *  result lands, and refusing-to-parse output must *not* be swallowed by the
 *  kept-prior path that protects prior `promotable`/`considered` verdicts. */
export const TO_SPEC_IN_FLIGHT_REASON = "in-flight";

/**
 * The grooming-row `reason` a `PROMOTABLE` verdict carries when the daemon's
 * ready gate refused to queue it (#1041). The verdict itself stays
 * `promotable` and `promotedAt`/`promotedBy` stay unset: the to-spec pass was
 * valid and must not be re-spent, but the spec is not queueable yet. One
 * constant, two writers-and-readers — the daemon stamps it, the tick's
 * promotion audit renders it — because the last time this fleet hand-copied a
 * reason string into a second file the two drifted apart (`status-render.ts`'s
 * duplicate of {@link TO_SPEC_IN_FLIGHT_REASON}).
 */
export const TO_SPEC_GATE_REJECTED_REASON = "gate-rejected";

/** What the ready gate found missing, as the rejected row records it. */
export interface ReadyGateRejection {
  /** The gate's own `missing[]`, verbatim and in order. */
  missing: string[];
  /** When the gate ran, so an audit can say how stale its finding is. */
  checkedAt: number;
}

/**
 * Recover a ready-gate rejection from a grooming row's evidence, or
 * `undefined` when the row is not one. The daemon writes the rejection as one
 * added top-level sibling of the untouched to-spec `result`
 * (`{kind:"to-spec", result:{…}, readyGate:{missing, checkedAt}}`), so
 * {@link toSpecDurableVerdict} still recovers the verdict and the candidate is
 * not re-groomed — the spec is rejected, not ungroomed.
 *
 * That durability is bounded by the verdict itself (#1060). A rejection
 * beside a verdict the contract still accepts is not re-groomable, because
 * the fault lies in the issue — missing criteria, a missing lane — and only
 * an edit to the issue can fix it. A rejection beside a verdict the contract
 * refuses — veltro#731's prose lane entry, persisted before parse-time lane
 * validation — reads as not-durable through {@link toSpecDurableVerdict},
 * and the candidate becomes re-groomable: no issue edit can ever satisfy a
 * gate whose refusal is baked into the verdict, and complying with the
 * refusal's remedy would write the unusable entry into the issue itself, so
 * a fresh pass is the only repair.
 *
 * Fails closed on anything it cannot read: an empty `missing` array is not a
 * rejection (a gate that found nothing missing passed), and a non-string entry
 * would render as `undefined` in a digest line an operator is asked to act on.
 */
export function parseReadyGateRejection(evidence: string): ReadyGateRejection | undefined {
  let parsed: unknown;
  try {
    parsed = JSON.parse(evidence);
  } catch {
    return undefined;
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return undefined;
  const row = parsed as Record<string, unknown>;
  if (row["kind"] !== "to-spec") return undefined;
  const gate = row["readyGate"];
  if (gate === null || typeof gate !== "object" || Array.isArray(gate)) return undefined;
  const { missing, checkedAt } = gate as { missing?: unknown; checkedAt?: unknown };
  if (!Array.isArray(missing) || missing.length === 0) return undefined;
  if (!missing.every((entry) => typeof entry === "string" && entry.length > 0)) return undefined;
  return {
    missing: missing as string[],
    checkedAt: typeof checkedAt === "number" && Number.isFinite(checkedAt) ? checkedAt : 0,
  };
}

/**
 * How long a launch row may sit before it is treated as a dead batch and the
 * candidate becomes eligible again. A batch that dies with the process (a
 * daemon stop between launch and settlement) must not park a candidate
 * forever; the 24h ceiling matches the source-freshness ceiling, so a
 * relaunched pass always reads new source evidence anyway.
 */
export const TO_SPEC_IN_FLIGHT_TTL_MS = 24 * 60 * 60 * 1_000;

/**
 * How long a refused pass parks its candidate before another batch may be
 * spent on it. Deliberately the same 24h number as the source-freshness
 * ceiling and the in-flight TTL — one granularity for this whole lifecycle,
 * not a third threshold to keep in sync: within that window neither the
 * authoritative source nor the issue has produced new evidence, so a retry
 * re-runs the identical prompt and refuses the identical way.
 *
 * Without it, a candidate whose delegated pass returns malformed,
 * source-less or stale output is immediately eligible again, so every
 * low-queue pass spends a full delegated batch re-grooming it — measured on
 * this fleet as five permanently-refused rows (#295, #296, #297, #679, #806)
 * re-offered on every pass, and as #807 groomed twice seven minutes apart
 * (#887).
 */
export const TO_SPEC_REFUSED_RETRY_COOLDOWN_MS = TO_SPEC_MAX_SOURCE_AGE_MS;

/**
 * The one durability rule for a grooming row: the validated to-spec result it
 * carries when that result is still fresh, or `undefined` when the row is not
 * durable grooming at all (no to-spec payload — a hand-edited or pre-#772
 * row — or a source observed past the freshness ceiling).
 *
 * Every reader of "is this issue already groomed?" MUST go through this:
 * {@link toSpecCandidateExclusion} (selection), the daemon's promotion gate,
 * and the queue digest's already-considered inventory. Two readers with two
 * predicates is exactly the #887 defect — the digest told the orchestrator
 * "never re-groom these" about rows the mechanical selection was
 * simultaneously offering.
 */
export function toSpecDurableVerdict(row: GroomingRecord, now: number): ToSpecResult | undefined {
  const result = parseToSpecEvidence(row.evidence);
  if (result === undefined) return undefined;
  return now - result.source.freshAt <= TO_SPEC_MAX_SOURCE_AGE_MS ? result : undefined;
}

/** The refusal a row records when its pass produced nothing usable, while the
 *  cooldown above still holds it out of a new batch; `undefined` for any
 *  other row, including a refusal whose cooldown has expired. */
export function toSpecRefusalOnCooldown(row: GroomingRecord, now: number): ToSpecFailure | undefined {
  const failure = parseToSpecFailureEvidence(row.evidence);
  if (failure === undefined) return undefined;
  return now - row.recordedAt <= TO_SPEC_REFUSED_RETRY_COOLDOWN_MS ? failure : undefined;
}

/** One backlog candidate the mechanical gate can judge. The tracker facts
 *  (labels, epics) are read from the authoritative open-issue snapshot at
 *  selection time; they travel in this view so the selector stays
 *  deterministic and testable. */
export interface ToSpecCandidateView {
  issue: number;
  title: string;
  /** The routing target — a routed `owner/repo` (from the issue's one
   *  `routing.labelPrefix<key>` label, resolved through `routing.repos`). */
  routing: string;
  /** Operator-parked (`project.stateLabels.backlog`); read from tracker labels. */
  parked?: boolean;
  /** A parent/epic with no independently runnable slice; read from the
   *  tracker's sub-issue probe. */
  parent?: boolean;
}

/**
 * Why one candidate is not eligible for a batch right now, or undefined when
 * it is. The single rule source for the selection helper and the launcher's
 * own re-vet at dispatch time — one rule, every reader, so a candidate
 * excluded in a digest line is excluded at launch for the same reason.
 *
 *  - `in-flight`: a launch row recorded within the TTL (a dead batch's row
 *    expires and the candidate becomes eligible again);
 *  - `file-lane` / `depends-on`: admission's durable mechanical holds (#735);
 *  - a fresh valid `to-spec` result in the grooming table: the candidate was
 *    already groomed at an observed source within the freshness ceiling, so
 *    re-running it would recompute a verdict that is still valid. New source
 *    evidence reconsiders it: once the recorded `freshAt` crosses the ceiling
 *    the row no longer reads as groomed, and a fresh pass overrides it;
 *  - a refused pass inside {@link TO_SPEC_REFUSED_RETRY_COOLDOWN_MS}: a full
 *    delegated batch was already spent and produced nothing usable
 *    (malformed, source-less or stale output). Retrying inside the cooldown
 *    re-runs the identical prompt against the same source and refuses the
 *    same way, which is how one broken candidate consumed a batch on every
 *    low-queue tick (#887);
 *  - `active`: a run is in flight on the issue right now.
 */
export function toSpecCandidateExclusion(
  candidate: { issue: number },
  facts: { grooming: GroomingRecord | undefined; active: boolean },
  now: number,
): string | undefined {
  const row = facts.grooming;
  if (row !== undefined) {
    if (row.reason === TO_SPEC_IN_FLIGHT_REASON) {
      if (now - row.recordedAt <= TO_SPEC_IN_FLIGHT_TTL_MS) {
        return `#${candidate.issue} is already in a to-spec batch (launched ${new Date(row.recordedAt).toISOString()})`;
      }
    } else if (row.reason === "file-lane" || row.reason === "depends-on") {
      return `#${candidate.issue} is mechanically blocked (${row.reason}) — the hold clears by itself`;
    } else {
      const durable = toSpecDurableVerdict(row, now);
      if (durable !== undefined) {
        return (
          `#${candidate.issue} was already groomed ${durable.verdict} (source ${durable.source.name}@` +
          `${durable.source.ref}, observed ${new Date(durable.source.freshAt).toISOString()}) — re-groom only ` +
          "with new source evidence"
        );
      }
      const refusal = toSpecRefusalOnCooldown(row, now);
      if (refusal !== undefined) {
        const retryAt = new Date(row.recordedAt + TO_SPEC_REFUSED_RETRY_COOLDOWN_MS).toISOString();
        return (
          `#${candidate.issue} already spent a to-spec batch that was refused as ${refusal.kind} ` +
          `(${new Date(row.recordedAt).toISOString()}) — eligible again after ${retryAt}, or once the ` +
          "issue or its source changes"
        );
      }
    }
  }
  if (facts.active) return `#${candidate.issue} has a dispatched run in flight`;
  return undefined;
}

/**
 * The mechanical half of grooming: deterministic, bounded selection of the
 * eligible candidates, smallest issue numbers first, never more than
 * {@link TO_SPEC_BATCH_MAX} per batch, nothing at/above a numeric grooming
 * trigger (`"always"` opens that gate and lets candidate state decide, #988),
 * nothing before the first dispatch summary exists (queue health unknown —
 * the same gate the queue digest uses). Parked and parent views are honored
 * when the caller supplies them.
 *
 * Production reaches this selector through
 * {@link selectToSpecCandidates}, which fills the views from the
 * authoritative tracker snapshot (park label from the open-issue labels,
 * parent/epic from the sub-issue probe) and streams the store-side exclusions
 * before the selector runs — so parked and parent exclusions are enforced on
 * the live path, not only on test inputs.
 */
export function selectToSpecBatch(input: {
  candidates: readonly ToSpecCandidateView[];
  summary: DispatchSummary | undefined;
  groomBelow: GroomTrigger;
  grooming: readonly GroomingRecord[];
  active: readonly { issue: number }[];
  now: number;
}): ToSpecCandidateView[] {
  if (input.summary === undefined) return [];
  if (!groomingDue(input.summary.routed, input.groomBelow)) return [];
  const groomingByIssue = new Map(input.grooming.map((row) => [row.issue, row]));
  const activeIssues = new Set(input.active.map((run) => run.issue));
  const selected: ToSpecCandidateView[] = [];
  for (const candidate of [...input.candidates].sort((a, b) => a.issue - b.issue)) {
    if (candidate.parked) continue;
    if (candidate.parent) continue;
    if (
      toSpecCandidateExclusion(
        { issue: candidate.issue },
        { grooming: groomingByIssue.get(candidate.issue), active: activeIssues.has(candidate.issue) },
        input.now,
      ) !== undefined
    ) {
      continue;
    }
    selected.push(candidate);
    if (selected.length >= TO_SPEC_BATCH_MAX) break;
  }
  return selected;
}

/**
 * The authoritative tracker surface the selection is built from (#777): the
 * two reads through the existing Tracker adapter that answer "which open
 * backlog issues are actually eligible right now". Production implements this
 * with `makeTracker(...).listOpenIssues()` / `.childrenOf(...)`; tests inject
 * deterministic fakes. `listOpenIssues` is the one open-issue snapshot with
 * labels (#203) — it answers the park label and the routing label
 * mechanically — and `childrenOf` answers whether a candidate is a parent/epic
 * (its sub-issues exist, so it has no independently runnable slice of its
 * own). Both fail closed: an unreadable snapshot means no batch is selected,
 * while an unreadable parent probe skips that candidate because eligibility
 * was not mechanically established.
 */
export interface ToSpecTrackerSeam {
  /** Every open issue in the tracker repo, labels included
   *  (`Tracker.listOpenIssues`). Throws when the tracker cannot be read. */
  listOpenIssues(project: ProjectConfig): Promise<ReadyIssue[]>;
  /** Sub-issues of one issue (`Tracker.childrenOf`). */
  childrenOf(project: ProjectConfig, issue: number): Promise<{ number: number; state: IssueState }[]>;
}

/**
 * The tracker half of the candidate views: map one open-issue snapshot onto
 * the pool the selector can judge. Issues still carrying the queue label are
 * already queued — not backlog — and leave the pool; issues with zero or
 * several `routing.labelPrefix` labels (or a label mapping to no configured
 * repo) cannot name an authoritative source to read and leave the pool,
 * exactly like admission's unroutable partition. The park label lands on the
 * view for the selector to drop; the parent/epic probe is separate (one
 * tracker read per candidate) and stays with {@link selectToSpecCandidates},
 * which runs it only for candidates the store-side exclusions did not already
 * reject.
 */
export function toSpecPoolFromSnapshot(
  issues: readonly ReadyIssue[],
  project: ProjectConfig,
): ToSpecCandidateView[] {
  const queueLabel = project.queueLabel;
  const parkLabel = project.stateLabels.backlog;
  const { labelPrefix, repos } = project.routing;
  const views: ToSpecCandidateView[] = [];
  for (const issue of issues) {
    if (issue.labels.includes(queueLabel)) continue;
    const matched = [...new Set(issue.labels.filter((l) => l.startsWith(labelPrefix)))];
    if (matched.length !== 1) continue;
    const key = matched[0]!.slice(labelPrefix.length);
    const target = Object.hasOwn(repos, key) ? repos[key]! : undefined;
    if (target === undefined) continue;
    views.push({
      issue: issue.number,
      title: issue.title,
      routing: repoSlugFor(target),
      parked: issue.labels.includes(parkLabel) ? true : undefined,
    });
  }
  views.sort((a, b) => a.issue - b.issue);
  return views;
}

/**
 * The production selection for one grooming pass: the complete mechanical
 * selection, from the authoritative tracker snapshot through every exclusion,
 * ending in the candidates a launcher may spend to-spec sessions on. All of
 * the following yield an empty batch:
 *
 *  - no dispatch row yet, or the routable queue is at/above a numeric
 *    grooming trigger (`"always"` opens that gate; candidate state decides,
 *    #988 — the same gate the queue digest uses);
 *  - the snapshot cannot be read — the selection fails closed rather than
 *    letting an agent self-filter parked/parent/epic candidates;
 *  - nothing survives the exclusions (parked, parent/epic, already groomed,
 *    in-flight, lane/dependency holds, dispatched runs).
 *
 * The parent/epic probe runs only for candidates the store-side exclusions
 * have not already rejected, in issue order, and stops as soon as
 * {@link TO_SPEC_BATCH_MAX} candidates are collected — a bounded set of
 * tracker reads per pass, never one per open issue.
 */
export async function selectToSpecCandidates(input: {
  project: ProjectConfig;
  seam: ToSpecTrackerSeam;
  summary: DispatchSummary | undefined;
  groomBelow: GroomTrigger;
  grooming: readonly GroomingRecord[];
  active: readonly { issue: number }[];
  /** An open-issue snapshot the caller already read from the tracker. Shared
   *  so one pass cannot describe two queues (#848); absent when that read
   *  failed or was never made, in which case the selection reads its own
   *  snapshot and fails closed on the same terms. */
  issues?: readonly ReadyIssue[];
  now: number;
}): Promise<ToSpecCandidateView[]> {
  if (input.summary === undefined) return [];
  if (!groomingDue(input.summary.routed, input.groomBelow)) return [];
  let issues = input.issues;
  if (issues === undefined) {
    try {
      issues = await input.seam.listOpenIssues(input.project);
    } catch {
      // No authoritative snapshot, no batch: launching without one would make
      // the agent the selector, which is exactly the defect this module
      // removes. The next pass retries the read.
      return [];
    }
  }
  const views = toSpecPoolFromSnapshot(issues, input.project);
  const groomingByIssue = new Map(input.grooming.map((row) => [row.issue, row]));
  const activeIssues = new Set(input.active.map((run) => run.issue));
  const pool: ToSpecCandidateView[] = [];
  for (const view of views) {
    if (view.parked) continue;
    if (
      toSpecCandidateExclusion(
        { issue: view.issue },
        { grooming: groomingByIssue.get(view.issue), active: activeIssues.has(view.issue) },
        input.now,
      ) !== undefined
    ) {
      continue;
    }
    let children: { number: number; state: IssueState }[];
    try {
      children = await input.seam.childrenOf(input.project, view.issue);
    } catch {
      // Eligibility was not mechanically established for this candidate, so it
      // sits out this pass rather than being launched on an unproven fact.
      continue;
    }
    if (children.length > 0) continue; // parent/epic with sub-issues: no runnable slice
    pool.push(view);
    if (pool.length >= TO_SPEC_BATCH_MAX) break;
  }
  // The one shared rule set re-runs on the tracker-vetted pool, so the pure
  // selector — not a prose instruction — answers what may be launched.
  return selectToSpecBatch({
    candidates: pool,
    summary: input.summary,
    groomBelow: input.groomBelow,
    grooming: input.grooming,
    active: input.active,
    now: input.now,
  });
}