/**
 * The daemon's half of grooming: launch to-spec scouts when the queue runs low,
 * persist what they answer, and promote what passes the ready gate.
 *
 * Named `groom-pass` against the top-level `groom.ts` because the two halves
 * must stay apart. `groom.ts` is the rule set — selection, exclusions, the
 * in-flight reasons — pure and shared, and the only place a candidate is ever
 * ruled in or out. This file is the pass: sessions, durable rows, label ops and
 * the wake. A rule that drifted in here would be a second answer to "should this
 * issue be groomed", which is precisely the failure the mechanical pipeline was
 * built to remove.
 *
 * `TO_SPEC_BRIEF_PATH` resolves through `PACKAGE_SRC_DIR` for the same reason
 * the worker template does: the prompt stayed in `src/briefs/` when this pass
 * moved a directory deeper.
 */
import { randomUUID } from "node:crypto";
import { mkdirSync } from "node:fs";
import { join } from "node:path";
import { resolveGroomRole, stateDir } from "../config.ts";
import { DEFAULT_GROOM_BELOW, TO_SPEC_AGENT, TO_SPEC_GATE_REJECTED_REASON, TO_SPEC_IN_FLIGHT_REASON, TO_SPEC_IN_FLIGHT_TTL_MS, groomingDue, selectToSpecCandidates, toSpecDurableVerdict, type ToSpecCandidateView } from "../groom.ts";
import { errText, log } from "../log.ts";
import { materializeOmpSettings } from "../omp-settings.ts";
import { readyGate } from "../ready-gate.ts";
import { recordToSpecGrooming, type ToSpecEvidence } from "../to-spec.ts";
import type { GroomingRecord, IssueComment, ProjectConfig, ReadyIssue } from "../types.ts";
import { wakeDispatch } from "../wake.ts";
import { TO_SPEC_MAX_TURNS, renderBrief, runToSpecScout } from "../worker.ts";
import type { WorkerPool } from "./admission-pass.ts";
import { PACKAGE_SRC_DIR, recordDaemonSessionSpend, repoSlug, type Deps } from "./deps.ts";

// ==================================================== daemon-driven grooming
// (#1041, #1040). The dispatcher watches its own queue depth, launches to-spec
// scouts when it runs low, and promotes what passes the ready gate — the three
// steps that used to need an orchestrator turn each, and therefore used to wait
// up to a whole tick interval apiece while the queue sat empty.
//
// The division of labour, and why each half is where it is:
//
//  - SELECTION is `groom.ts`: pure, shared, and the only rule set. The daemon
//    supplies the authoritative reads (its own routable count, the open-issue
//    snapshot, the durable grooming rows, its live runs) and takes back a
//    bounded candidate list. Nothing here re-derives an exclusion.
//  - LAUNCH is this module: a read-only session per candidate, the
//    `handleReviewAdjudication` shape exactly — per-item session directory
//    under the state root, staged omp settings, a socket, the configured role,
//    and a durable settle on every branch including "no answer".
//  - PERSISTENCE is `recordToSpecGrooming`: still the only validator, and it
//    stamps the daemon's own launch time as the observation window (#1000).
//  - PROMOTION is `readyGate` plus the epic-scope read below it, and it is the
//    one place a queue label is added: provenance first (the latch), then the
//    label op, then the wake.
//
// The in-flight grooming row is the single-flight latch for the batch. It is
// written BEFORE any session starts, so a second pass in the same window and a
// restart mid-batch both read the claim; a launch that dies with the process
// leaves the row to expire on {@link TO_SPEC_IN_FLIGHT_TTL_MS} rather than
// parking its candidate forever.

/** The grooming prompt, read per launch so an edit applies to the next batch
 *  rather than the next daemon restart — the same contract `buildBrief` has. */
export const TO_SPEC_BRIEF_PATH = join(PACKAGE_SRC_DIR, "briefs", "to-spec.md");

/** One batch's identity, carried into every item of it. */
export interface ToSpecBatch {
  /** Short opaque id: it names the batch in logs, in the in-flight evidence and
   *  in each item's session directory, so a spend line, a durable row and a
   *  transcript can be tied together after the fact. */
  id: string;
  /** The daemon's own launch stamp. It — never anything the agent wrote — is
   *  the source observation time every verdict in this batch persists with
   *  (#1000). */
  launchedAt: number;
}

/**
 * Launch the to-spec grooming batch this project's queue depth calls for
 * (#1041).
 *
 * `routable` is the count this pass just measured, not a stored one: the whole
 * point of moving grooming into the daemon is that the drought and the response
 * happen in the same pass. The durable dispatch row still has to exist, because
 * its absence means no pass has ever completed here and the shared selector
 * refuses to judge a queue nobody has routed.
 *
 * Never blocks the tick: with a worker pool the launches ride it exactly like
 * adjudications, so a batch that takes minutes cannot delay the dispatch that
 * follows it. Without one (a `--once` pass, a test) it awaits, which is what
 * makes the whole pipeline assertable in one call.
 */
export async function dispatchToSpecGrooming(
  d: Deps,
  routable: number,
  pool?: WorkerPool,
): Promise<void> {
  const project = d.project;
  const groomBelow = project.groomBelow ?? DEFAULT_GROOM_BELOW;
  if (!groomingDue(routable, groomBelow)) return;
  // The stop fence (#374), for the same reason every other launch honours it:
  // a batch started into a draining daemon is spend with nowhere to land — the
  // sessions die with the process and their rows expire unanswered.
  if (d.drain?.draining === true) {
    log("grooming held: the daemon is draining");
    return;
  }
  const prior = d.store.latestDispatch(project.name);
  if (prior === undefined) return;
  const now = Date.now();
  const grooming = d.store.groomingVerdicts(project.name);
  // One batch at a time. The per-candidate exclusion inside the selector stops
  // one issue being groomed twice; this stops a second batch existing at all,
  // which is the bound on what an unattended dispatcher may spend per window.
  const inFlight = grooming.filter(
    (row) => row.reason === TO_SPEC_IN_FLIGHT_REASON && now - row.recordedAt <= TO_SPEC_IN_FLIGHT_TTL_MS,
  );
  if (inFlight.length > 0) {
    log(
      `grooming held: ${inFlight.length} candidate(s) still in a launched batch ` +
        `(${inFlight.map((row) => `#${row.issue}`).join(" ")})`,
    );
    return;
  }
  let candidates: ToSpecCandidateView[];
  try {
    candidates = await selectToSpecCandidates({
      project,
      seam: {
        listOpenIssues: () => d.tracker.listOpenIssues(),
        childrenOf: (_project, issue) => d.tracker.childrenOf(issue),
      },
      // The selector reads two things off this: that a pass has completed here
      // at all, and the routable depth. The depth is this pass's own
      // measurement — the row supplies the rest of the shape and is never
      // written back.
      summary: { ...prior, routed: routable },
      groomBelow,
      grooming,
      active: d.store.liveRuns(project.name),
      now,
    });
  } catch (err) {
    log(`grooming selection failed: ${errText(err)} — retrying next pass`);
    return;
  }
  if (candidates.length === 0) return;

  const batch: ToSpecBatch = { id: randomUUID().slice(0, 8), launchedAt: now };
  const claim = JSON.stringify({
    kind: "to-spec-in-flight",
    launchedAt: batch.launchedAt,
    batch: batch.id,
    agent: TO_SPEC_AGENT,
  });
  // Every claim before any launch, deliberately: a session that starts while a
  // sibling's claim is still unwritten is a candidate a concurrent pass can
  // select twice.
  for (const candidate of candidates) {
    d.store.upsertGrooming({
      project: project.name,
      issue: candidate.issue,
      verdict: "blocked",
      reason: TO_SPEC_IN_FLIGHT_REASON,
      evidence: claim,
      at: batch.launchedAt,
    });
  }
  log(
    `grooming batch ${batch.id}: routable ${routable} below ${String(groomBelow)} — ` +
      `launching ${candidates.map((c) => `#${c.issue}`).join(" ")} as ${resolveGroomRole(project)}`,
  );
  const launches = candidates.map((candidate) => handleToSpecGrooming(d, candidate, batch));
  if (pool !== undefined) {
    for (const launch of launches) pool.launch(launch);
    return;
  }
  await Promise.allSettled(launches);
}

/**
 * Groom one candidate to a durable verdict, and promote it when it earns that
 * (#1041).
 *
 * Every branch settles the row. That is the whole reason the in-flight marker
 * can be a latch rather than a lease: a scout that answered, one that answered
 * garbage, and one whose session never started all replace their own claim, so
 * the only way a candidate stays parked is the daemon dying mid-flight — which
 * the TTL covers.
 */
export async function handleToSpecGrooming(
  d: Deps,
  candidate: ToSpecCandidateView,
  batch: ToSpecBatch,
): Promise<void> {
  const project = d.project;
  const issue = candidate.issue;
  // The body verbatim, read now rather than carried from the selection
  // snapshot: the brief's whole premise is that the groomer judges the issue as
  // it stands against the source as it stands. An unreadable body is said out
  // loud in the brief — a scout that silently groomed a blank issue would
  // produce a confident verdict about nothing.
  let body: string | undefined;
  try {
    body = (await d.tracker.getIssue(issue))?.body;
  } catch {
    body = undefined;
  }
  const brief = renderBrief(await Bun.file(TO_SPEC_BRIEF_PATH).text(), {
    TRACKER_REPO: project.tracker.repo,
    ISSUE_NUMBER: String(issue),
    CANDIDATE_TITLE: candidate.title,
    ISSUE_BODY:
      body === undefined || body.trim() === ""
        ? "(the issue body could not be read — say so in reasonNotToPromote rather than guessing what it asked for)"
        : body,
    SOURCE: candidate.routing,
    SOURCE_REF: groomSourceRef(project, candidate.routing),
  });

  const sessionDir = join(stateDir(), "sessions", `to-spec-${batch.id}-${issue}`);
  mkdirSync(sessionDir, { recursive: true });
  const ompSettingsFile = materializeOmpSettings(project, sessionDir);
  const result = await (d.runToSpecScoutImpl ?? runToSpecScout)({
    brief,
    // No worktree: a scout has no branch, and everything about the candidate is
    // in the brief. The session directory doubles as its confined cwd.
    cwd: sessionDir,
    sessionDir,
    model: resolveGroomRole(project),
    ...(ompSettingsFile === undefined ? {} : { ompSettingsFile }),
    socketPath: join(sessionDir, "ipc.sock"),
    maxTurns: TO_SPEC_MAX_TURNS,
  });

  // Recorded before the verdict is validated: an unparseable answer cost
  // whatever it cost, and a refusal that is invisible in the accounting is how
  // grooming spend became unattributable in the first place.
  const groomRole = resolveGroomRole(project);
  recordDaemonSessionSpend(d, {
    role: "groom",
    issue,
    ...(groomRole === undefined ? {} : { model: groomRole }),
    ...(result.model === undefined ? {} : { resolvedModel: result.model }),
    turns: result.turns,
    spendUsd: result.spendUsd,
    at: Date.now(),
  });

  // The scout's answer, unjudged, through the one validator (#1064). Only the
  // structured payload is graded: a session that produced no payload at all is
  // its own refusal class (`no-answer`), with the narration recorded as
  // context, and a payload recovered from the text carries its `via` marker so
  // the record says it did not come through the yield contract. `raw: ""`
  // therefore means "no answer", and the pass never grades `result.report` as
  // if it were one.
  const outcome = recordToSpecGrooming(d.store, {
    project: project.name,
    issue,
    input: result.raw,
    via: result.via === "text" ? "text" : result.raw === "" ? undefined : "yield",
    report: result.report,
    repaired: result.repaired,
    killedAtCeiling: result.killedAtCeiling,
    maxTurns: TO_SPEC_MAX_TURNS,
    turns: result.turns,
    launchedAt: batch.launchedAt,
  });
  const ran = result.model === undefined ? "" : ` by ${result.model}`;
  // "$0.00" for a session that took turns and reported nothing is the misleading
  // zero #46 named: say "unmetered" instead, matching what the durable row now
  // stores and what the stats lane renders.
  const cost =
    result.turns > 0 && result.spendUsd === 0 ? "unmetered" : `$${result.spendUsd.toFixed(2)}`;
  const repairNote = result.repaired === true ? " (repaired)" : "";
  log(
    `#${issue} groomed ${outcome.record.verdict}(${outcome.record.reason})${ran} in batch ${batch.id} — ` +
      `${result.turns} turn(s), ${cost}${repairNote}`,
  );
  if (outcome.kind !== "persisted" || outcome.record.verdict !== "promotable") return;
  await promoteGroomedVerdict(d, outcome.record);
}

/**
 * The ref a grooming brief tells its scout to read the source at (#1041).
 *
 * The routed repository's configured default branch — a name, not a sha,
 * because conductor no longer needs the scout to witness freshness: the
 * dispatcher stamps the observation window itself (#1000), so resolving a head
 * here would cost a network read per candidate to produce a fact nothing reads.
 * `HEAD` is the honest fallback for a routing target this config cannot
 * resolve: it names the repository's own default branch without claiming to
 * know which branch that is.
 */
export function groomSourceRef(project: ProjectConfig, routing: string): string {
  const target = Object.values(project.routing.repos).find((repo) => repoSlug(repo) === routing);
  return target?.defaultBranch ?? "HEAD";
}

/**
 * Promote one fresh `promotable` verdict, or record exactly why not (#1041,
 * #1040).
 *
 * Mechanical promotion is the phase's whole point and also its whole risk, so
 * the shape is: judge with a pure function, refuse with a durable list, and
 * mutate through the outbox only after the provenance stamp has been won.
 *
 *  - The judgement is {@link readyGate} over the issue as it stands right now.
 *    A tracker read that fails becomes `"unread"`, which the gate refuses
 *    rather than treating as an empty issue: nothing here was queued by a
 *    human, so every unknown is a refusal.
 *  - The epic-scope check is this function's own, because the gate is pure and
 *    approval is a store fact. A child of an unapproved epic is not promotable
 *    however complete its brief: an orchestrator can file a decomposition, but
 *    only an operator agrees to its scope. No decision row is created here —
 *    net-new scope stays the orchestrator's proposal to make.
 *  - A refusal re-records the SAME verdict with `readyGate.missing` beside it,
 *    under {@link TO_SPEC_GATE_REJECTED_REASON}. The verdict was valid; the
 *    spec is simply not queueable, and the tick's promotion audit renders the
 *    whole list so one pass fixes every problem instead of four passes fixing
 *    one each.
 *  - A pass stamps provenance FIRST. `markGroomingPromoted` returning false is
 *    the latch: another pass already promoted this verdict, so this one adds no
 *    label and fires no wake, and a restart mid-promotion cannot queue an issue
 *    twice.
 *
 * Returns whether this call promoted, for the caller's own log; the durable
 * facts are the row and the label op.
 */
export async function promoteGroomedVerdict(d: Deps, row: GroomingRecord): Promise<boolean> {
  const project = d.project;
  const issue = row.issue;
  const now = Date.now();
  const result = toSpecDurableVerdict(row, now);
  if (result === undefined) {
    // Not durable grooming: no recoverable to-spec payload, or a source
    // observed past the freshness ceiling. Either way there is nothing to
    // promote ON, and the candidate is re-groomable by the same rule.
    log(`#${issue} not promoted: its verdict carries no fresh to-spec result`);
    return false;
  }
  let snapshot: ReadyIssue | undefined;
  try {
    snapshot = await d.tracker.getIssue(issue);
  } catch {
    snapshot = undefined;
  }
  let comments: IssueComment[] | "unread" = "unread";
  if (snapshot !== undefined) {
    try {
      comments = await d.tracker.listComments(issue);
    } catch {
      comments = "unread";
    }
  }
  const verdict = readyGate({
    result,
    issueBody: snapshot?.body ?? "",
    comments: snapshot === undefined ? "unread" : comments,
    labels: snapshot?.labels ?? [],
    project,
  });
  const missing = [...(verdict.ok ? [] : verdict.missing), ...(await epicScopeMisses(d, issue))];
  if (missing.length > 0) {
    const evidence: ToSpecEvidence = { kind: "to-spec", result, readyGate: { missing, checkedAt: now } };
    d.store.upsertGrooming({
      project: project.name,
      issue,
      verdict: row.verdict,
      reason: TO_SPEC_GATE_REJECTED_REASON,
      evidence: JSON.stringify(evidence),
      // The verdict's own recording time is preserved: the gate ran now, but
      // the judgement is the one the scout made, and its freshness window is
      // what decides re-grooming.
      at: row.recordedAt,
    });
    log(`#${issue} not promoted (${missing.length} gate refusal(s)): ${missing[0]}`);
    return false;
  }
  if (!d.store.markGroomingPromoted(project.name, issue, now, "daemon")) {
    log(`#${issue} promotion skipped: this verdict already carries a promotion`);
    return false;
  }
  // The outbox owns labels — never a direct tracker write. The projector
  // applies it and retries until GitHub takes it, and the eligibility overlay
  // keeps this pass coherent in the meantime (#201).
  d.store.enqueueLabelOps(project.name, [{ issue, op: "add", label: project.queueLabel }]);
  try {
    log(`#${issue} promoted by the ready gate — ${await (d.wake ?? wakeDispatch)(project.name)}`);
  } catch (err) {
    log(`#${issue} promoted but the dispatch wake failed (${errText(err)}) — the next scheduled pass will claim`);
  }
  return true;
}

/**
 * The epic-scope half of promotion: nothing, or the one reason this slice's
 * scope has not been agreed to (#1041).
 *
 * An issue with no parent is not epic-serialized work and needs no approval —
 * it exists in the tracker, which means somebody filed it. A child does: a
 * decomposition can be proposed without its scope being accepted, and
 * mechanically queueing children of an unapproved epic is how an orchestrator's
 * proposal becomes a fleet's week of work without anyone saying yes.
 *
 * Fails closed on an unreadable parent, matching every other epic read in this
 * package: "the tracker could not say" is not evidence of standalone work.
 */
export async function epicScopeMisses(d: Deps, issue: number): Promise<string[]> {
  let parent: number | undefined;
  try {
    parent = await d.tracker.parentOf(issue);
  } catch (err) {
    return [
      `the parent/epic lookup failed (${errText(err)}), so the scope this slice belongs to could not be established`,
    ];
  }
  if (parent === undefined) return [];
  if (d.store.epicApproval(d.project.name, parent) !== undefined) return [];
  return [
    `it is a child of epic #${parent}, whose scope nobody approved — ` +
      `\`omp-conductor epic approve ${parent}\` once the operator agrees to it`,
  ];
}