/**
 * The to-spec grooming contract (#679, #772): the one structured result a
 * groomer agent must answer in before anything it says may become a durable
 * grooming verdict. #735 landed the durable grooming table, but no shipped
 * agent contract, strict result parser, or persistence path. This module
 * ships all three: the prompt lives in `briefs/to-spec.md`, the strict JSON
 * Schema the task agent is invoked through is `TO_SPEC_SCHEMA`, and
 * `recordToSpecGrooming` is the only persistence path — parse and validate
 * first, then write through the existing `Store.upsertGrooming`.
 *
 * The store's verdict surface is `promotable | blocked | considered` (#735).
 * The five contract verdicts map onto it — PROMOTABLE → `promotable`;
 * ALREADY DONE and NEEDS DECOMPOSITION → `considered`; BLOCKED and NEEDS
 * PRODUCT DECISION → `blocked` — and a result that cannot be trusted
 * (no answer at all, malformed JSON, a schema violation, no authoritative
 * source, or a stale source) persists as `blocked` and never as
 * `promotable`/`considered`.
 * Reprocessing the same candidate is deterministic: the same input at the
 * same observation time produces the same row, and malformed output never
 * erases a prior valid (promotable/considered) result. Stale or source-less
 * output does replace it — as blocked, because a judgeable refusal is
 * information: the latest pass could not ground a verdict, so the candidate
 * must not keep reading as groomed.
 *
 * The structured result survives a process/tick restart inside the record's
 * existing `evidence` field — a JSON payload with a `kind` discriminator
 * (`to-spec` | `to-spec-failure`) — so no `GroomingRecord` extension and no
 * second store is needed. Nothing here launches a task, edits an issue,
 * creates decomposition children, or touches the queue label: the
 * orchestrator remains the only promotion authority.
 *
 * The invocation contract is OMP's native task-agent output schema: the
 * later tick passes `outputSchema: TO_SPEC_SCHEMA` with `schemaMode:
 * "strict"` to the task tool. Conductor re-validates whatever the agent
 * returned before persisting, because the harness is permissive by design
 * and the store vets only what this module hands it.
 *
 * Who stamps the observation time (#1000): the dispatcher, never the agent.
 * A to-spec groomer runs with reading tools only and therefore has no clock,
 * so the `source.freshAt` it used to be asked for was always a guess — and a
 * guess landing ahead of the daemon's clock discarded the whole verdict as a
 * clock lie, twice in one afternoon on verdicts that were otherwise correct.
 * The schema still accepts the key (it is `.strict()`, so refusing an older
 * prompt's output would recreate exactly that discard) but ignores its value:
 * `parseToSpecResult` stamps the batch's launch time — or the settle time
 * when no launch time is known — and that stamp is what persists.
 *
 * Sizing is evidence, not a feeling (#1041): a `PROMOTABLE` verdict carries
 * `sizingEvidence`, and a `NEEDS DECOMPOSITION` verdict carries an ordered
 * `decomposition`. #1035 was persisted `PROMOTABLE` with nine acceptance
 * criteria across four lifecycle files totalling 28,812 lines; attempt 1
 * exhausted 181 of its 180 turns. The verdict had asserted one-budget fit —
 * what was missing was anything that could falsify the assertion. So the
 * contract now asks for the falsifier itself (distinct behaviours, the
 * production modules and test surfaces each needs, shared-file sequencing)
 * and there is deliberately no numeric threshold anywhere: a line or
 * criterion count that decided verdicts would be gamed by splitting prose,
 * and the judgement is the groomer's.
 *
 * One contract, every surface (#883): this Zod schema is the single source of
 * truth. `TO_SPEC_SCHEMA` generated from it stamps the native task launch,
 * `parseToSpecResult` persists against it, and the drift test pins the
 * rendered brief, the shipped agent role, and the floor's return contract to
 * exactly these fields — a field the prose asks for that the schema refuses
 * is a candidate recorded blocked for following instructions.
 */

import { z } from "zod";

import type { GroomingRecord, GroomingVerdict, Store } from "./types.ts";

/** The five verdicts a to-spec groomer may return (#772). */
export const TO_SPEC_VERDICTS = [
  "ALREADY DONE",
  "PROMOTABLE",
  "NEEDS DECOMPOSITION",
  "BLOCKED",
  "NEEDS PRODUCT DECISION",
] as const;

export type ToSpecVerdict = (typeof TO_SPEC_VERDICTS)[number];

/**
 * How old an authoritative source may be before a result drawn from it no
 * longer counts. A verdict computed over a default branch that has moved on
 * may have missed a landing that invalidates it, so a result whose source was
 * observed longer ago than this persists as `blocked(stale-source)` and can
 * never read as promotable/considered. 24 hours: a groomed candidate must
 * have looked at the code within a day, and a re-grooming pass a day later
 * re-reads the live source anyway.
 *
 * Since #1000 the observation time is the dispatcher's own stamp, so this
 * ceiling can only fire for a batch that genuinely ran longer than a day —
 * a real signal about the batch rather than a verdict on the agent's
 * arithmetic.
 */
export const TO_SPEC_MAX_SOURCE_AGE_MS = 24 * 60 * 60 * 1000;

/** `routing: "MULTI"` requires `routingSplit` naming one repo per slice. */
const MULTI_ROUTING = "MULTI" as const;

/**
 * One write-lane entry (#1060): a plausible repository-relative path.
 *
 * A lane entry that is prose — veltro#731's
 * `"docker-compose.dev.yml (only if …)"` parsed as one non-empty string —
 * used to harden into the durable verdict and brick its issue: the ready
 * gate then refuses every promotion whose issue-side lane disagrees, a
 * gate-rejected candidate is deliberately never re-groomed, and the
 * refusal's own remedy (edit the issue to match) would have written the
 * parenthetical into the issue. The grammar is therefore enforced here, at
 * parse time, where a refusal is `malformed`, the retry cooldown can act on
 * it, and nothing unusable becomes an admission contract. Deliberately
 * narrow so no valid lane is newly rejected: dots, hyphens, nested
 * directories, globs and trailing slashes all pass; only a leading `/`
 * (absolute) or whitespace/parentheses inside the entry (prose) is refused.
 */
const LANE_ENTRY_PROSE = /[\s()]/;

/** Every problem with one write lane's entries, each naming the offending
 *  entry verbatim (`field[index] "…"`), in array order — the detail a repair
 *  round hands back and a refusal records must be readable against the next
 *  pass. Empty for a usable lane. */
function laneEntryProblems(field: string, entries: readonly string[]): string[] {
  const problems: string[] = [];
  for (const [index, entry] of entries.entries()) {
    if (entry.startsWith("/")) {
      problems.push(
        `${field}[${index}] ${JSON.stringify(entry)} is absolute — a write lane names repository-relative paths`,
      );
    } else if (LANE_ENTRY_PROSE.test(entry)) {
      problems.push(
        `${field}[${index}] ${JSON.stringify(entry)} is not a path — write-lane entries carry no whitespace or parentheses`,
      );
    }
  }
  return problems;
}

const ToSpecSourceSchema = z
  .object({
    name: z.string().trim().min(1).describe("The authoritative source that was read: repo or tracker, e.g. `TerrifiedBug/conductor`."),
    ref: z.string().trim().min(1).describe("The exact ref (commit/branch/tag) that was reviewed."),
    freshAt: z
      .number()
      .int()
      .optional()
      .describe(
        "Ignored — do not send it. Conductor stamps the observation time itself from the batch window (#1000); a groomer has reading tools only and no clock, so any value here is a guess. The key is still accepted so an older prompt's output is never refused for sending one.",
      ),
  })
  .strict()
  .describe("The authoritative source and ref every verdict must stand on; conductor stamps its freshness (#1000).");

/**
 * One ordered child of a decomposition proposal (#1041). A `NEEDS
 * DECOMPOSITION` verdict used to be free to say "split this up" in prose,
 * which is a verdict nobody can act on: the operator still has to do the
 * decomposition. Each child therefore carries exactly what filing it as an
 * issue needs — a title, the exact write lane it owns, what it waits on, the
 * silent fake it invites, and the commands that prove it — so the proposal is
 * a filing instruction rather than an opinion.
 */
const ToSpecDecompositionChildSchema = z
  .object({
    title: z.string().trim().min(1).describe("The child issue's title — one slice, stated as the behaviour it lands."),
    writeLane: z
      .array(z.string().trim().min(1))
      .min(1)
      .describe("The files/dirs this child alone writes — its `## Exact write lane`."),
    dependsOn: z
      .array(z.union([z.string().trim().min(1), z.number().int().min(1)]))
      .describe(
        "What this child waits on: an earlier child's title, or an existing issue number. Children sharing a file in their write lanes must be serialised here — two concurrent slices writing one core module is the collision the file lane exists to prevent. `[]` for the first child.",
      ),
    likelySilentFake: z
      .string()
      .trim()
      .min(1)
      .describe("The one thing most likely to be silently faked in this child, and how to prove it is not."),
    proofCommands: z
      .array(z.string().trim().min(1))
      .min(1)
      .describe("The focused commands that prove this child, each with its cwd when it matters."),
  })
  .strict()
  .superRefine((child, ctx) => {
    // #1060: a decomposition child carries the identical admission contract,
    // so its write lane obeys the identical grammar — prose here would
    // harden into a child slice nothing can file.
    for (const problem of laneEntryProblems("writeLane", child.writeLane)) {
      ctx.addIssue({ code: "custom", message: problem });
    }
  })
  .describe("One ordered child slice a decomposition proposal names (#1041).");

/**
 * The one strict contract a grooming result must satisfy before it may
 * persist as promotable or considered (and everything else maps onto blocked
 * or is refused). `proposedBrief` and `reasonNotToPromote` are exactly one of
 * the two, chosen by verdict: PROMOTABLE carries the brief, every other
 * verdict states why it must not be promoted.
 */
const ToSpecResultSchema = z
  .object({
    verdict: z.enum(TO_SPEC_VERDICTS).describe("Exactly one of the five to-spec verdicts."),
    routing: z
      .string()
      .trim()
      .min(1)
      .describe("Exactly one `owner/repo`, or `MULTI` with `routingSplit` naming the split."),
    routingSplit: z
      .record(z.string(), z.string())
      .optional()
      .describe("Required when `routing` is `MULTI`: what each slice goes to."),
    source: ToSpecSourceSchema,
    evidence: z
      .array(z.string().trim().min(1))
      .optional()
      .describe(
        "The files/symbols proving the verdict — required for ALREADY DONE (the code that already does the work, never a title match); welcome on every verdict.",
      ),
    laterWorkInvalidates: z
      .boolean()
      .describe("Whether later work (an epic committed after this candidate was filed) invalidated its premise."),
    laterWorkNote: z
      .string()
      .trim()
      .min(1)
      .describe("What was searched for the invalidating-later-work check and what was found — the check's evidence."),
    entryPoints: z
      .array(z.string().trim().min(1))
      .min(3)
      .max(6)
      .describe("3-6 files to change or read first, the discovery a worker budget dies on when absent."),
    existingTests: z
      .array(z.string().trim().min(1))
      .describe("Tests that already exercise the behaviour, by path; empty when none exist."),
    likelySilentFake: z
      .string()
      .trim()
      .min(1)
      .describe("The single thing most likely to be silently faked while implementing and how to prove it is not."),
    proofCommands: z
      .array(z.string().trim().min(1))
      .min(1)
      .describe("Focused commands that prove the work, each with its cwd when relevant."),
    fileLane: z
      .array(z.string().trim().min(1))
      .min(1)
      .describe("The files/dirs this slice writes — the file lane that serialises concurrent work."),
    dependencies: z
      .array(z.union([z.string().trim().min(1), z.number().int().min(1)]))
      .describe(
        'Open prerequisites this work is blocked on, as bare issue numbers (875) or strings ("875"); empty when none.',
      ),
    sizingEvidence: z
      .string()
      .trim()
      .min(1)
      .optional()
      .describe(
        "Required for PROMOTABLE: the source-backed one-budget analysis — the distinct behaviours/state transitions this slice introduces, the production modules and focused test surfaces each of them needs, the shared-file sequencing between them, and why that fits one configured worker attempt. A small file count is not this evidence when the files are high-fanout lifecycle modules with large integration suites (#1041).",
      ),
    decomposition: z
      .array(ToSpecDecompositionChildSchema)
      .min(2)
      .optional()
      .describe(
        "Required for NEEDS DECOMPOSITION and accepted on no other verdict: the ordered children this candidate splits into, first to last, each naming its title, write lane, dependencies, silent fake and proof commands. Children sharing a write-lane path must serialise through `dependsOn` (#1041).",
      ),
    proposedBrief: z
      .string()
      .trim()
      .min(1)
      .optional()
      .describe("Required for PROMOTABLE: the brief a worker would be dispatched with."),
    reasonNotToPromote: z
      .string()
      .trim()
      .min(1)
      .optional()
      .describe("Required for every verdict except PROMOTABLE: why this must not be promoted."),
  })
  .strict()
  .superRefine((value, ctx) => {
    const verdict = value.verdict;
    if (verdict === "PROMOTABLE") {
      if (value.proposedBrief === undefined) {
        ctx.addIssue({ code: "custom", message: "PROMOTABLE requires proposedBrief" });
      }
      if (value.reasonNotToPromote !== undefined) {
        ctx.addIssue({ code: "custom", message: "PROMOTABLE must not carry reasonNotToPromote" });
      }
      // #1041: the one-budget claim is only a claim until something could
      // falsify it. #1035 asserted it and burned 181 of 180 turns.
      if (value.sizingEvidence === undefined) {
        ctx.addIssue({
          code: "custom",
          message:
            "PROMOTABLE requires sizingEvidence: the distinct behaviours, their production modules and test surfaces, the shared-file sequencing, and why that fits one worker attempt",
        });
      }
    } else {
      if (value.reasonNotToPromote === undefined) {
        ctx.addIssue({ code: "custom", message: `${verdict} requires reasonNotToPromote` });
      }
      if (value.proposedBrief !== undefined) {
        ctx.addIssue({ code: "custom", message: `${verdict} must not carry proposedBrief` });
      }
    }
    if (verdict === "ALREADY DONE" && (value.evidence === undefined || value.evidence.length === 0)) {
      ctx.addIssue({
        code: "custom",
        message: "ALREADY DONE requires evidence naming the file/symbol that already does the work",
      });
    }
    // A decomposition proposal is a filing instruction or it is nothing
    // (#1041): NEEDS DECOMPOSITION owes ordered children, and every other
    // verdict owes none — children beside PROMOTABLE would mean the verdict
    // contradicts itself.
    if (verdict === "NEEDS DECOMPOSITION") {
      if (value.decomposition === undefined) {
        ctx.addIssue({
          code: "custom",
          message:
            "NEEDS DECOMPOSITION requires decomposition: the ordered children, each with its title, write lane, dependsOn, likelySilentFake and proofCommands",
        });
      }
    } else if (value.decomposition !== undefined) {
      ctx.addIssue({ code: "custom", message: `decomposition is only valid with verdict "NEEDS DECOMPOSITION", not ${verdict}` });
    }
    // Two children writing one core module concurrently is the collision the
    // file lane exists to prevent, so the later of any file-sharing pair must
    // name what it waits on. Order is the array's order — child n may only
    // depend on something before it.
    const children = value.decomposition ?? [];
    for (let i = 1; i < children.length; i++) {
      const child = children[i]!;
      if (child.dependsOn.length > 0) continue;
      const own = new Set(child.writeLane.map((path) => path.trim()));
      for (let j = 0; j < i; j++) {
        const shared = children[j]!.writeLane.map((path) => path.trim()).find((path) => own.has(path));
        if (shared === undefined) continue;
        ctx.addIssue({
          code: "custom",
          message: `decomposition child ${i + 1} shares ${shared} with child ${j + 1} and must serialise on it through dependsOn`,
        });
        break;
      }
    }
    if (value.routing === MULTI_ROUTING && value.routingSplit === undefined) {
      ctx.addIssue({ code: "custom", message: 'routing "MULTI" requires routingSplit' });
    }
    if (value.routing !== MULTI_ROUTING && value.routingSplit !== undefined) {
      ctx.addIssue({ code: "custom", message: "routingSplit is only valid with routing `MULTI`" });
    }
    // #1060: the lane is the admission contract, so its entries must be
    // plausible repository-relative paths — refused here, at parse time,
    // rather than hardening into a durable verdict no promotion can use.
    for (const problem of laneEntryProblems("fileLane", value.fileLane)) {
      ctx.addIssue({ code: "custom", message: problem });
    }
  })
  .describe("A complete, to-spec grooming result; nothing outside this shape is accepted.");

/** The strict JSON Schema the OMP task tool validates the agent's structured
 *  output with (`outputSchema` + `schemaMode: "strict"`, #772). */
export const TO_SPEC_SCHEMA: object = z.toJSONSchema(ToSpecResultSchema) as object;

/**
 * The validated result every reader gets. `source.freshAt` is a required
 * number here even though the schema no longer asks the agent for one:
 * {@link parseToSpecResult} stamps the dispatcher's own observation time onto
 * every result it returns (#1000), so a persisted verdict always carries a
 * freshness witness — just never the agent's guess.
 */
export type ToSpecResult = Omit<z.infer<typeof ToSpecResultSchema>, "source"> & {
  source: { name: string; ref: string; freshAt: number };
};

/** Why a result could not be trusted; each persists as a blocked record.
 *
 * `no-answer` is the session-level refusal (#1064): the pass produced no
 * payload at all — the scout was killed at its turn ceiling, or the session
 * ended without ever yielding. It is a property of the session, never of
 * output, so it is decided by the pass before anything is parsed and is never
 * `malformed`. */
export type ToSpecFailure =
  | { kind: "malformed"; detail: string }
  | { kind: "no-answer"; detail: string }
  | { kind: "missing-source"; detail: string }
  | { kind: "stale-source"; detail: string };

export type ParseToSpecOutcome = { ok: true; result: ToSpecResult } | { ok: false; failure: ToSpecFailure };

/** The row's `evidence` when a result passed validation. */
export interface ToSpecEvidence {
  kind: "to-spec";
  result: ToSpecResult;
  /**
   * How the answer arrived (#1064): through the `yield` tool, or recovered
   * from the session's text because it never yielded. `"text"` is the record's
   * admission that the verdict did not come through the yield contract — the
   * answer was complete and valid, so it stands, but a later reader can tell
   * the two apart. Absent (or `"yield"`) means it came through the contract.
   */
  via?: "yield" | "text";
  /**
   * Whether the one bounded repair round produced the answer (#1064): the
   * first answer parsed but violated the contract, the validation error was
   * handed back, and the re-answer is what this record carries.
   */
  repaired?: boolean;
  /**
   * What the ready gate found missing when a valid PROMOTABLE verdict was
   * refused mechanical promotion (#1041). A sibling of `result`, never a
   * replacement for it: the verdict stays recoverable, so the candidate reads
   * as groomed-but-rejected rather than ungroomed, and the tick can name what
   * an operator must fix. Absent on every promoted or unjudged row.
   */
  readyGate?: { missing: string[]; checkedAt: number };
}

/** The row's `evidence` when a result was refused. */
export interface ToSpecFailureEvidence {
  kind: "to-spec-failure";
  failure: ToSpecFailure;
  /** The raw agent output, bounded, so a refusal is diagnosable. */
  input: string;
}

/**
 * How much raw agent output a failure record may carry (#1064).
 *
 * Sized for the whole refused payload, not a sample of it: a complete verdict
 * with a proposed brief and sizing analysis routinely exceeds 2 KB, and a
 * discarded verdict should be recoverable from the row rather than
 * re-derived at full session cost. 8 KB holds a full payload with headroom
 * while still keeping a single refused row bounded.
 */
export const TO_SPEC_INPUT_SAMPLE_MAX = 8_000;

/** Refused results always land on the same store surface, named by failure. */
const FAILURE_REASON: Record<ToSpecFailure["kind"], string> = {
  malformed: "malformed",
  "no-answer": "no-answer",
  "missing-source": "missing-source",
  "stale-source": "stale-source",
};

/** The five contract verdicts onto the store's three (#735). */
function mapStoreVerdict(verdict: ToSpecVerdict): GroomingVerdict {
  switch (verdict) {
    case "PROMOTABLE":
      return "promotable";
    case "ALREADY DONE":
    case "NEEDS DECOMPOSITION":
      return "considered";
    case "BLOCKED":
    case "NEEDS PRODUCT DECISION":
      return "blocked";
  }
}

/** The groomer's own label the record carries (the #679 half of `reason`). */
function mapStoreReason(verdict: ToSpecVerdict): string {
  switch (verdict) {
    case "ALREADY DONE":
      return "already-done";
    case "PROMOTABLE":
      return "promotable";
    case "NEEDS DECOMPOSITION":
      return "needs-decomposition";
    case "BLOCKED":
      return "blocked";
    case "NEEDS PRODUCT DECISION":
      return "needs-product-decision";
  }
}

/** True for a JSON object value (null and arrays excluded). */
function isObject(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

/**
 * Pulls one fenced JSON block out of an agent answer, exactly like the setup
 * probes: a model's text, parsed as JSON. Anything unparseable is malformed —
 * no seed at all, and the candidate persists as blocked rather than groomed.
 */
function extractJson(input: string): string {
  const fenced = /```(?:json)?\s*([\s\S]*?)```/.exec(input);
  return (fenced?.[1] ?? input).trim();
}

/**
 * The three-way gate the scout runner decides a payload against (#1064):
 * whether a string is a payload at all, and whether that payload already
 * conforms to the contract or is a single re-answer away from it.
 *
 *  - `conforming` — a JSON object that satisfies the strict contract. For
 *    text-recovered blocks this is the criterion that separates "a complete
 *    verdict written outside the yield tool" from mere narration.
 *  - `repairable` — a JSON object that violates the contract (the #1062
 *    shape: every required field present, one key misnamed). The failure
 *    carries the first validation issue, which is what the repair round hands
 *    back to the scout.
 *  - `not-a-payload` — prose, truncated JSON, a non-object: nothing a
 *    re-answer can repair, so the pass records no answer rather than grading
 *    it.
 *
 * Same parser as {@link parseToSpecResult}, deliberately: the runner probes
 * with the one contract so its repair decision can never drift from what the
 * persistence layer will accept — this is the one validator used twice, not a
 * second, weaker one.
 */
export type ToSpecPayloadGate =
  | { state: "conforming"; body: string }
  | { state: "repairable"; body: string; failure: ToSpecFailure }
  | { state: "not-a-payload" };

export function gateToSpecPayload(payload: string): ToSpecPayloadGate {
  const body = extractJson(payload);
  if (body.length === 0) return { state: "not-a-payload" };
  let raw: unknown;
  try {
    raw = JSON.parse(body);
  } catch {
    return { state: "not-a-payload" };
  }
  if (!isObject(raw)) return { state: "not-a-payload" };
  const parsed = ToSpecResultSchema.safeParse(raw);
  if (!parsed.success) {
    // The field path rides in front of the message: zod says "Required" for a
    // missing field and "Unrecognized key(s)…" for an extra one, and the
    // repair round's whole value is that the scout sees WHICH field failed.
    const first = parsed.error.issues[0];
    const detail =
      first === undefined
        ? "schema violation"
        : first.path.length === 0
          ? first.message
          : `${first.path.join(".")}: ${first.message}`;
    return { state: "repairable", body, failure: { kind: "malformed", detail } };
  }
  return { state: "conforming", body };
}

/**
 * Parse and validate a groomer's raw output against the to-spec contract.
 * `now` is the settle time, explicit so staleness and determinism are
 * testable: the same input over the same batch window always yields the same
 * outcome. On failure the outcome says *why* in one of the three refusal
 * classes: unparseable/schema-breaking output is `malformed`, output that
 * never names an authoritative source (or lacks name/ref) is
 * `missing-source`, and a batch whose window is wider than
 * `TO_SPEC_MAX_SOURCE_AGE_MS` is `stale-source`.
 *
 * The observation time is the dispatcher's, never the agent's (#1000).
 * `opts.launchedAt` is when the batch was launched — the conservative (older)
 * end of its wall-clock window, so staleness is never understated — and
 * `now` stands in when no launch time is known. Whatever the agent put in
 * `source.freshAt` is overwritten by that stamp before the result is
 * returned, so the persisted evidence carries the dispatcher's own witness
 * and a guessed timestamp can no longer discard a correct verdict.
 */
export function parseToSpecResult(
  input: string,
  now: number,
  opts?: { launchedAt?: number },
): ParseToSpecOutcome {
  const body = extractJson(input);
  if (body.length === 0) {
    return { ok: false, failure: { kind: "malformed", detail: "empty answer — no JSON found" } };
  }
  let raw: unknown;
  try {
    raw = JSON.parse(body);
  } catch {
    return { ok: false, failure: { kind: "malformed", detail: "output is not valid JSON" } };
  }
  if (!isObject(raw)) {
    return { ok: false, failure: { kind: "malformed", detail: "output is not a JSON object" } };
  }
  const parsed = ToSpecResultSchema.safeParse(raw);
  if (!parsed.success) {
    const detail = parsed.error.issues[0]?.message ?? "schema violation";
    const source = raw.source;
    // `freshAt` is no longer part of this test: the dispatcher stamps it, so
    // a source is complete when it names what was read and at which ref.
    if (!isObject(source) || source.name === undefined || source.ref === undefined) {
      return { ok: false, failure: { kind: "missing-source", detail: `no authoritative source — ${detail}` } };
    }
    return { ok: false, failure: { kind: "malformed", detail } };
  }
  // The launch bound when the dispatcher knows it and it precedes the settle
  // time — a launch stamped after `now` is clock skew, not evidence, so the
  // settle time stands in rather than producing a negative age (#1000).
  const launchedAt = opts?.launchedAt;
  const observedAt =
    launchedAt !== undefined && Number.isFinite(launchedAt) && launchedAt <= now ? launchedAt : now;
  if (now - observedAt > TO_SPEC_MAX_SOURCE_AGE_MS) {
    return {
      ok: false,
      failure: {
        kind: "stale-source",
        detail:
          `the batch that read the source launched at ${observedAt}, ${now - observedAt}ms before this item ` +
          `settled — older than the ${TO_SPEC_MAX_SOURCE_AGE_MS}ms ceiling`,
      },
    };
  }
  return { ok: true, result: { ...parsed.data, source: { ...parsed.data.source, freshAt: observedAt } } };
}

/** What one grooming pass did to the row, so the caller can report it. */
export type ToSpecGroomingOutcome =
  | { kind: "persisted"; record: GroomingRecord }
  | { kind: "kept-prior"; record: GroomingRecord };

/** One candidate's grooming pass. */
export interface ToSpecGroomingRequest {
  project: string;
  issue: number;
  /** The agent's structured answer, exactly as returned: the yield payload,
   *  or a conforming block recovered from its text. Empty when the pass
   *  produced no answer at all — which is its own refusal class, decided here
   *  before anything is parsed (#1064). */
  input: string;
  /** How the answer arrived: through the yield tool, or recovered from the
   *  session's text because it never yielded. `"text"` marks the record so it
   *  stays visible that the verdict did not come through the yield contract;
   *  omitted reads as a yield, and an empty `input` with no `via` reads as no
   *  answer. */
  via?: "yield" | "text";
  /** The session's last narration, recorded as the refusal's context when no
   *  answer arrived — never parsed as a verdict (#1064). */
  report?: string;
  /** Whether the one bounded repair round produced the carried answer
   *  (#1064); recorded on the row so a repaired verdict is distinguishable. */
  repaired?: boolean;
  /** Whether the session was killed at its turn ceiling before it answered
   *  (#1064) — half of the no-answer shape naming. */
  killedAtCeiling?: boolean;
  /** The turn ceiling the pass ran under, and the turns it took: the other
   *  half of the no-answer shape naming (`turns >= maxTurns` reads as a cap
   *  kill even when the runner did not flag it). */
  maxTurns?: number;
  turns?: number;
  /** Observation time for staleness and the recorded row; `Date.now()` when
   *  omitted (tests pass it to keep reprocessing deterministic). */
  now?: number;
  /** When the batch that produced this output was launched, on the
   *  dispatcher's own clock (#1000). It becomes the source observation time in
   *  preference to `now`; omit it and the settle time stands in. */
  launchedAt?: number;
}

function failureEvidence(failure: ToSpecFailure, input: string): string {
  const sample =
    input.length > TO_SPEC_INPUT_SAMPLE_MAX ? `${input.slice(0, TO_SPEC_INPUT_SAMPLE_MAX)}…[truncated]` : input;
  const payload: ToSpecFailureEvidence = { kind: "to-spec-failure", failure, input: sample };
  return JSON.stringify(payload);
}

/**
 * The only way a groomer's output becomes a durable grooming verdict: parse
 * and validate, then upsert through the existing store — no second store, no
 * `GroomingRecord` extension. Valid results map to the store's verdict
 * surface with the full structured result serialized into `evidence`
 * (`{"kind":"to-spec","result":…}`), which is what survives a restart. A
 * refused result persists as `blocked` with the failure named in
 * `evidence` (`to-spec-failure`) — never as promotable/considered — except
 * when a prior valid result exists: malformed reprocessing then keeps that
 * prior row instead of erasing it. Nothing here touches labels or issues.
 *
 * `request.launchedAt` is the dispatcher's launch stamp for the batch (#1000);
 * it, not anything the agent wrote, becomes the recorded source observation
 * time.
 */
/**
 * The reason a via-text verdict carries: the groomer's own label with a
 * visible marker that the answer arrived outside the yield contract (#1064),
 * so `status` reads `#13 promotable via text` instead of hiding the
 * distinction in an evidence string. Yielded verdicts keep the bare label.
 */
function mapStoreReasonWithVia(verdict: ToSpecVerdict, via: "yield" | "text" | undefined): string {
  const label = mapStoreReason(verdict);
  return via === "text" ? `${label} via text` : label;
}

/**
 * The no-answer detail names which of the two session shapes produced it
 * (#1064): a turn-ceiling kill (`killedAtCeiling`, or `turns >= maxTurns`
 * when the runner did not flag it) or a session that ended without ever
 * yielding.
 */
function noAnswerDetail(request: ToSpecGroomingRequest): string {
  const killed =
    request.killedAtCeiling === true ||
    (request.maxTurns !== undefined && request.turns !== undefined && request.turns >= request.maxTurns);
  const turns = request.turns === undefined ? "" : ` after ${request.turns} turn(s)`;
  if (killed) {
    const ceiling = request.maxTurns === undefined ? "its turn ceiling" : `its ${request.maxTurns}-turn ceiling`;
    return `the scout was killed at ${ceiling}${turns} and never yielded an answer`;
  }
  return `the scout session ended${turns} without yielding an answer`;
}

/**
 * Persist one refusal, or keep a prior valid verdict when the new pass
 * produced something no-answer-like or malformed. `input` is what the failure
 * record carries: the raw payload of the refused pass — or, for a no-answer
 * pass, the session's narration as context (#1064).
 */
function persistGroomingRefusal(
  store: Store,
  request: ToSpecGroomingRequest,
  failure: ToSpecFailure,
  input: string,
  now: number,
): ToSpecGroomingOutcome {
  const prior = store.grooming(request.project, request.issue);
  if (
    (failure.kind === "malformed" || failure.kind === "no-answer") &&
    prior !== undefined &&
    (prior.verdict === "promotable" || prior.verdict === "considered")
  ) {
    return { kind: "kept-prior", record: prior };
  }
  store.upsertGrooming({
    project: request.project,
    issue: request.issue,
    verdict: "blocked",
    reason: FAILURE_REASON[failure.kind],
    evidence: failureEvidence(failure, input),
    at: now,
  });
  return { kind: "persisted", record: store.grooming(request.project, request.issue)! };
}

/**
 * The only way a groomer's output becomes a durable grooming verdict: parse
 * and validate, then upsert through the existing store — no second store, no
 * `GroomingRecord` extension. Valid results map to the store's verdict
 * surface with the full structured result serialized into `evidence`
 * (`{"kind":"to-spec","result":…}`), which is what survives a restart. A
 * refused result persists as `blocked` with the failure named in
 * `evidence` (`to-spec-failure`) — never as promotable/considered — except
 * when a prior valid result exists: malformed and no-answer reprocessing then
 * keep that prior row instead of erasing it. Nothing here touches labels or
 * issues.
 *
 * The refusal classes come from the whole pass, not just its output (#1064):
 *
 *  - `no-answer` — the session produced no payload at all (no yield, no
 *    conforming text block): a turn-ceiling kill or a session end without a
 *    yield, recorded by shape, never as `malformed`. The narration is kept as
 *    context only. A text-carried block that fails validation is also
 *    `no-answer` — such a session never yielded, and the block is not an
 *    answer — except a batch-window staleness, which is a property of the
 *    pass rather than the answer.
 *  - `malformed` / `missing-source` — a *yielded* payload (or an unjudged
 *    one) that violates the contract, with the raw payload persisted so the
 *    discarded verdict stays recoverable.
 *  - `stale-source` — the batch window outlived the freshness ceiling.
 *
 * `request.launchedAt` is the dispatcher's launch stamp for the batch (#1000);
 * it, not anything the agent wrote, becomes the recorded source observation
 * time.
 */
export function recordToSpecGrooming(store: Store, request: ToSpecGroomingRequest): ToSpecGroomingOutcome {
  const now = request.now ?? Date.now();
  if (request.input === "") {
    return persistGroomingRefusal(
      store,
      request,
      { kind: "no-answer", detail: noAnswerDetail(request) },
      request.report ?? "",
      now,
    );
  }
  const parsed = parseToSpecResult(request.input, now, { launchedAt: request.launchedAt });
  if (parsed.ok) {
    const evidence: ToSpecEvidence = {
      kind: "to-spec",
      result: parsed.result,
      ...(request.via === undefined ? {} : { via: request.via }),
      ...(request.repaired === true ? { repaired: true } : {}),
    };
    store.upsertGrooming({
      project: request.project,
      issue: request.issue,
      verdict: mapStoreVerdict(parsed.result.verdict),
      reason: mapStoreReasonWithVia(parsed.result.verdict, request.via),
      evidence: JSON.stringify(evidence),
      at: now,
    });
    return { kind: "persisted", record: store.grooming(request.project, request.issue)! };
  }
  // A text-carried answer that fails validation was never yielded and does
  // not conform: that is a no-answer pass, with the block kept as the
  // failure's context — never malformed (#1064). The one exception is a
  // batch-window staleness, whose class is about the pass, not the answer.
  const failure =
    request.via === "text" && parsed.failure.kind !== "stale-source"
      ? { kind: "no-answer" as const, detail: noAnswerDetail(request) }
      : parsed.failure;
  return persistGroomingRefusal(store, request, failure, request.input, now);
}

/**
 * Recover the validated result from a record's evidence (the restart
 * round-trip). Returns undefined for anything that is not a `to-spec`
 * payload or fails validation again, so a hand-edited row cannot read back
 * as groomed.
 */
export function parseToSpecEvidence(evidence: string): ToSpecResult | undefined {
  let raw: unknown;
  try {
    raw = JSON.parse(evidence);
  } catch {
    return undefined;
  }
  if (!isObject(raw) || raw.kind !== "to-spec") return undefined;
  const parsed = ToSpecResultSchema.safeParse(raw.result);
  if (!parsed.success) return undefined;
  const freshAt = parsed.data.source.freshAt;
  // A row with no stamped observation time cannot have its staleness judged,
  // and freshness is precisely the field nobody may forge (#1000) — so the row
  // reads as not-durably-groomed and becomes re-groomable, rather than
  // carrying a fabricated timestamp that would read fresh forever. Every row
  // `recordToSpecGrooming` writes is stamped; this is the hand-edited case.
  if (freshAt === undefined) return undefined;
  return { ...parsed.data, source: { ...parsed.data.source, freshAt } };
}

/**
 * The mirror of {@link parseToSpecEvidence} for the refusal payload
 * {@link failureEvidence} writes: the recorded failure when a row is a
 * refused pass, `undefined` for anything else (a valid verdict, a mechanical
 * hold, an in-flight marker, a hand-edited row).
 *
 * A refusal is durable evidence too — it says a full delegated batch was
 * already spent on this candidate and produced nothing usable — so the
 * selection side needs to read it back rather than seeing only "not groomed"
 * and re-spending immediately (#887).
 */
export function parseToSpecFailureEvidence(evidence: string): ToSpecFailure | undefined {
  let raw: unknown;
  try {
    raw = JSON.parse(evidence);
  } catch {
    return undefined;
  }
  if (!isObject(raw) || raw.kind !== "to-spec-failure") return undefined;
  const failure = raw.failure;
  if (!isObject(failure) || typeof failure.detail !== "string") return undefined;
  if (
    failure.kind !== "malformed" &&
    failure.kind !== "no-answer" &&
    failure.kind !== "missing-source" &&
    failure.kind !== "stale-source"
  ) {
    return undefined;
  }
  return { kind: failure.kind, detail: failure.detail };
}