# Concepts — diagnosis as a textual gradient

> A reward tells you *how much* a candidate failed. An editor can only act on
> *why*. diagnose converts traces into that "why" — the learning signal the
> optimizer edits against. Implementation: `diagnose()` in this skill's
> `scripts/run.py`.

## Why a scalar reward is not enough

Reinforcement learning turns a scalar reward into a parameter update via a
gradient. Prompt/tool/skill optimization has no weights to nudge — the artifact
is text. The substitute for the gradient is **natural-language feedback**: a
diagnosis of what went wrong that an LLM optimizer can read and translate into a
concrete edit. GEPA's central finding is that language is a *richer* learning
signal than a scalar reward — it carries direction and cause, not just
magnitude — which is why reflective evolution can match or beat RL using far
fewer rollouts. diagnose is where that signal is manufactured.

## The reflective dataset

For each failing task, diagnose emits the triple GEPA calls a reflective dataset:

- **Inputs** — what the task asked.
- **Generated Outputs** — what the agent actually produced.
- **Feedback** — the scorer's diagnosis of the failure.
- **Trajectory** — a *path* to the full trace (reasoning, tool calls). A path and
  not parsed content, because the trace format belongs to the runner: the location
  comes from `Adapter.trajectories(split)`, which is documented to return "*any*
  structure, files in *any* format". Anything that parsed it here would bind this
  phase to one runner, which a phase skill may never do.

Giving the optimizer this triple instead of a bare score is the difference
between "you got 0.4" and "on these inputs you called the wrong tool because you
misread field X; here is the trace." Only the latter tells it what to change.

## What makes feedback *actionable*

Three properties, in order of importance:

1. **Specific** — name the concrete failure (wrong tool, skipped step, misread
   field), not "incorrect".
2. **Causal** — point at the decision that produced it, so the edit has a target.
3. **General** — describe the *pattern*, not the single instance, so the fix
   transfers to unseen tasks.

### The no-leak rule (non-negotiable)
Feedback must never quote the gold/target answer. If it does, the optimizer can
"fix" a task by hard-coding the answer into the prompt — it memorizes the eval set
instead of learning the capability. Val climbs, the sealed test number collapses,
and the run's headline becomes a lie. Keep feedback at the level of *what class of
mistake was made*, never *what the right answer is*. This is the same discipline
the scorer must honor at intake.

## Clustering: fix classes, not instances

Listing failures flatly invites the optimizer to patch each one individually —
which overfits to the val set and bloats the artifact. Grouping failures by a
shared signature turns the signal **actionable at scale**: ten tasks failing for
one reason become one generalizing edit. This is the through-line of the
diagnose-then-edit optimizer family:

- **GEPA** — *actionable side information* distilled from trajectories, then
  combined across a Pareto frontier of variants.
- **Trace-analysis optimizers** — parallel analysts read traces and surface
  recurring failure modes.
- **Evolutionary loops** — cluster issues so each generation targets a class of
  defect rather than a single example.

A good round produces a *few* clusters; "one cluster per task" means the signature
is too fine and no generalization is happening.

### Why the signature is an overlap test, not a string match

SKILL.md gives the procedure; this is why it has the shape it does. A signature
built by hashing the leading words of the scorer's feedback fails in both
directions, and both failures are silent:

- **Fragmentation.** One root cause reaches the scorer in many phrasings — "did not
  confirm the change", "omitted required confirmation step", "missing confirmation
  before the write". Under string equality that is three clusters, so the optimizer
  writes three narrow patches for one defect and overfits val three times over.
  Requiring only *overlap* between the content-word keys keeps them together.
- **Collapse.** A scorer whose every message opens with a fixed preamble ("Grading
  failed for this trajectory because the expected outcome was not met: …") makes
  every failure share its first dozen tokens. Under a prefix key the entire signal
  becomes one cluster. Stripping the prefix that *all* failures share removes the
  boilerplate without anyone having to configure a per-benchmark pattern.

Both directions are regression-tested in `scripts/check.py`; the mechanism is
`scripts/cluster.py`.

## Hand-off to the gate

diagnose also emits `kept_good` — the tasks the current candidate already passes.
This is the baseline the gate's **no-regression** check protects: a fix for one
failure cluster must not silently break a passing task. Diagnosis and acceptance
are two halves of one honest loop — diagnose says what to change; the gate refuses
changes that trade a real pass for an aggregate bump.

## Sources
- GEPA: Reflective Prompt Evolution Can Outperform RL (Agrawal et al., 2025) —
  reflective dataset, natural-language feedback as the learning signal, Pareto
  combination of lessons: https://arxiv.org/abs/2507.19457
- DSPy / MIPRO lineage (Opsahl-Ong et al., 2024) — bootstrapped
  instruction/demonstration optimization from execution feedback:
  https://arxiv.org/abs/2406.11695
- τ-bench (Yao et al., 2024) — trajectory-level failures that only trace
  inspection reveals: https://arxiv.org/abs/2406.12045

"""The 'diagnose' phase composes the project adapter + shared harness; it declares no
abstract methods of its own. check.py verifies the wiring instead of stubs."""

"""Contract: diagnose emits a well-formed reflective dataset (the real task INPUT, a
pointer to the full trace) and clusters failures by root cause — one cause under
several phrasings must NOT fragment, and several causes under one scorer preamble
must NOT collapse.
"""

from __future__ import annotations

import sys
import tempfile
from pathlib import Path

import _bootstrap  # noqa: F401

from cap_evolve.skillcheck import Checker, import_run, temp_run_dir, write_val_rollout


def _clusters_of(run, feedbacks: dict[str, str]) -> list[dict]:
    """Cluster a set of ``task_id -> feedback`` failures, records-free."""
    return run.diagnose([{"input": {}, "rollout": {"output": ""},
                          "score": {"task_id": t, "reward": 0.0, "feedback": f}}
                         for t, f in sorted(feedbacks.items())])["clusters"]


def main() -> int:
    c = Checker("diagnose")
    run = import_run()
    c.require_main(run)

    with tempfile.TemporaryDirectory() as d:
        rd, _ = temp_run_dir(Path(d))
        # two failures with the SAME root cause but different numbers, one pass.
        write_val_rollout(rd, "a", reward=0.0, feedback="Expected 5 but got 7",
                          task_input={"expr": "2+3"}, output="7")
        write_val_rollout(rd, "b", reward=0.0, feedback="Expected 9 but got 2",
                          task_input={"expr": "4+5"}, output="2")
        write_val_rollout(rd, "c", reward=1.0, feedback="ok",
                          task_input={"expr": "1+1"}, output="2")

        records = run._load_records(rd, "seed")
        result = run.diagnose(records)

        rd_set = result["reflective_dataset"]
        c.check(len(rd_set) == 2, f"expected 2 failing entries, got {len(rd_set)}")
        c.check(result["kept_good"] == ["c"], f"passing task not kept-good: {result['kept_good']}")

        # Inputs must be the real task INPUT, not the task id.
        entry = next(e for e in rd_set if e["task_id"] == "a")
        c.check(entry["Inputs"] == {"expr": "2+3"},
                f"Inputs carries the wrong thing (should be the task input): {entry['Inputs']}",
                note="reflective dataset carries the actual task input")
        c.check(entry["Generated Outputs"] == "7", "Generated Outputs missing the rollout output")

        # Every entry must be traceable back to its FULL trace, or the failure SITE
        # (which is half the cluster key) is unrecoverable from the entry alone.
        c.check(Path(entry["Trajectory"]).exists(),
                f"Trajectory pointer does not resolve: {entry['Trajectory']}",
                note="each reflective entry points at its full trace")
        # ...and when the RUNNER has a native trace store, that pointer comes from
        # adapter.trajectories(split) — never from a hardcoded trace layout. A phase
        # skill that guessed the path would be bound to one runner.
        proj = Path(d) / "project"
        (proj / "adapters").mkdir(parents=True)
        (proj / "adapters" / "adapter.py").write_text(
            "from pathlib import Path\n"
            "from cap_evolve import CapabilityAdapter, Rollout, Score\n"
            "class Adapter(CapabilityAdapter):\n"
            "    def tasks(self, split): return []\n"
            "    def run_target(self, task, ctx, *, seed=0):\n"
            "        return Rollout(task_id='x', output='')\n"
            "    def score(self, task, rollout): return Score(task_id='x', reward=0.0)\n"
            "    def trajectories(self, split, ctx=None):\n"
            "        return Path('/native') / split\n", encoding="utf-8")
        native = run.trace_dir(str(proj), "val")
        c.check(native is not None and native.endswith("val"),
                f"adapter.trajectories(split) was not used for the trace path: {native}",
                note="trace location comes from adapter.trajectories(split), never a "
                     "hardcoded trace schema")
        c.check(run.diagnose(records, "root-cause", native)["reflective_dataset"][0]
                ["Trajectory"] == native,
                "the adapter's native trace dir did not reach the reflective dataset")
        c.check(run.trace_dir(None, "val") is None
                and run.trace_dir(str(Path(d) / "nope"), "val") is None,
                "trace_dir must degrade to None with no project / no adapter",
                note="a missing trace pointer never blocks a diagnosis")

        # the two same-root-cause failures cluster together under one signature.
        c.check(len(result["clusters"]) == 1
                and result["clusters"][0]["tasks"] == ["a", "b"],
                f"clustering did not group same-cause failures: {result['clusters']}",
                note="same cause, different values -> one cluster")
        c.check(result["clusters"][0]["score_lost"] == 2.0,
                f"cluster is not ranked by score lost: {result['clusters']}",
                note="clusters carry the score they can recover")

        # --split reads a DIFFERENT split's rollouts, and val stays the default. Four
        # other algorithms call this phase without --split, so the default must not
        # move; and diagnosing TRAIN must not silently return val's failures.
        c.check(run._load_records(rd, "seed", "train") == [],
                "diagnose read val rollouts when asked for train",
                note="--split train reads rollouts/train, never rollouts/val")
        train_dir = rd.rollouts / "train"
        train_dir.mkdir(parents=True, exist_ok=True)
        (train_dir / "z__seed__t0.json").write_text(
            '{"input": {"expr": "8+8"}, "rollout": {"task_id": "z", "output": "3", '
            '"error": null}, "score": {"task_id": "z", "reward": 0.0, '
            '"feedback": "Expected 16 but got 3", "n": 1, "stderr": 0.0, '
            '"trial_rewards": [0.0], "raw": {"errored": false}}}', encoding="utf-8")
        tr = run.diagnose(run._load_records(rd, "seed", "train"))
        c.check([e["task_id"] for e in tr["reflective_dataset"]] == ["z"],
                f"train diagnosis returned the wrong tasks: {tr['reflective_dataset']}",
                note="train is diagnosable — the honest learning surface when the gate "
                     "scores val")
        c.check([e["task_id"] for e in run.diagnose(
                    run._load_records(rd, "seed"))["reflective_dataset"]] == ["a", "b"],
                "the default split is no longer val — that would change every "
                "algorithm that calls diagnose without --split")

        # REGRESSION 1 — one root cause, three phrasings. A lexical prefix key splits
        # this into three clusters and the optimizer then writes three narrow patches.
        one = _clusters_of(run, {
            "p": "Task failed: the agent did not confirm the change with the user",
            "q": "Agent omitted required confirmation step",
            "r": "Missing confirmation before the write",
        })
        c.check(len(one) == 1 and one[0]["tasks"] == ["p", "q", "r"],
                f"one root cause fragmented into {len(one)} clusters: {one}",
                note="one cause under 3 phrasings -> 1 cluster (no fragmentation)")

        # REGRESSION 2 — three causes behind one long scorer preamble. A first-N-token
        # key collapses them into one cluster and the whole signal is lost.
        pre = "Grading failed for this trajectory because the expected outcome was not met: "
        many = _clusters_of(run, {
            "p": pre + "missing confirmation before the write",
            "q": pre + "wrong payment id supplied to the refund",
            "r": pre + "refused a valid booking request",
        })
        c.check(len(many) >= 3,
                f"3 distinct causes collapsed into {len(many)} cluster(s) behind a "
                f"shared preamble: {many}",
                note="shared scorer preamble is stripped -> distinct causes stay distinct")

        # Determinism: same input twice, byte-identical clusters.
        import json as _json
        again = _clusters_of(run, {
            "p": "Task failed: the agent did not confirm the change with the user",
            "q": "Agent omitted required confirmation step",
            "r": "Missing confirmation before the write",
        })
        c.check(_json.dumps(one, sort_keys=True) == _json.dumps(again, sort_keys=True),
                "clustering is not deterministic across runs",
                note="identical input -> identical output")

    return c.emit()


if __name__ == "__main__":
    sys.exit(main())

"""Deterministic failure clustering — the mechanical half of diagnose.

A cluster is ONE root cause. Its identity is a **(site, expectation)** pair: where
the trajectory went wrong (the tool / field / step the trace names) and which
expectation was missed. Two failures belong to the same cluster when both halves
agree, however differently the scorer phrased it and whatever task-specific values
it quoted.

Deriving that key from a feedback string is deterministic, so it lives here instead
of being re-improvised in prose on every iteration:

1. strip the token prefix every failure's feedback shares (a scorer boilerplate
   preamble otherwise makes every failure look identical);
2. reduce to content words — drop quoted literals / numbers / punctuation
   (task-specific, not causal), drop stopwords and OUTCOME-GENERIC words which say
   *that* it failed and never *why*, ALSO drop any stem that recurs in most of the
   BATCH's own feedbacks (corpus-relative, on top of the fixed English list — a
   benchmark's own recurring vocabulary is boilerplate too, just not English
   boilerplate), and crudely stem the rest so confirm/confirmed/confirmation collapse;
3. the surviving token set is the key — identifiers are the site, verbs the
   expectation;
4. two keys are the same cluster when they OVERLAP: |A∩B| / min(|A|,|B|) >= 0.5,
   merged transitively. Overlap rather than equality is what keeps one root cause
   from splitting into three clusters under three phrasings.

Everything is sorted, so the same input always yields byte-identical output.

One failure class is detected MECHANICALLY from the rollout instead of from the feedback
string — see ``narrated_without_action``. It is a known LLM-agent failure mode rather than
any benchmark's own, and the feedback a scorer writes for it is indistinguishable from the
feedback for a genuinely wrong write, so the lexical key above cannot separate the two.
"""

from __future__ import annotations

import re

# Words that name THAT something failed, never WHY. Keeping them lets two unrelated
# causes look similar just because both were reported as a failure.
_GENERIC = frozenset("""
fail failed failure error errored wrong incorrect invalid missing miss expected
expect expects got produce produced output outputs task tasks agent step steps
required require unexpected instead actual result results response correct bad
should would did does done value values return returned reward score scored
scoring trajectory grading grade graded because only just also however issue
problem problems reason cause caused unable cannot able
""".split())

_STOP = frozenset("""
the a an and or but of to in on for with that this these those it its is are was
were be been being at by from as has have had do than then when which while will
can could all any into out no not none more most some such very over under after
before during about again there here their them they you your our his her one two
""".split())

# Longest first so "confirmation" -> "confirm", not "confirmatio".
_SUFFIXES = ("ations", "ation", "ements", "ement", "ingly", "ings", "ing", "edly",
             "ness", "ions", "ion", "ive", "ed", "es", "ly", "s")

_MIN_STEM = 4

OVERLAP_MIN = 0.5

# A token that recurs in more than this fraction of the batch's feedbacks is treated as
# THIS BENCHMARK's own boilerplate, on top of (never instead of) the generic-English list
# above. `_GENERIC`/`_STOP` are a fixed vocabulary of ENGLISH filler; they cannot know that
# a given benchmark's scorer always says "action check state write" regardless of root
# cause. A fixed corpus can only be told apart from noise on that corpus, so the threshold
# is corpus-relative rather than a hardcoded word list — mechanical and benchmark-agnostic
# by construction.
CORPUS_STOP_FRAC = 0.65


#: The name of the mechanically-detected cluster (see ``narrated_without_action``).
NARRATED_WITHOUT_ACTION = "narrated_without_action"

#: A final message CLAIMING a state change happened. Generic English completion frames
#: ("has been updated", "I have cancelled it", "was successfully processed") — the verbs are
#: ordinary state-change English, not any benchmark's vocabulary.
_COMPLETION_RE = re.compile(
    r"\b(?:has|have|had|been|was|were|is|are|i've|ive)\b[^.!?\n]{0,60}?\b(?:successfully\s+)?"
    r"(?:updated|cancell?ed|canceled|booked|rebooked|changed|processed|created|deleted|"
    r"removed|submitted|scheduled|rescheduled|applied|completed|refunded|transferred|"
    r"modified|saved|sent|issued|placed|added|registered|assigned|closed|reset)\b",
    re.IGNORECASE)

#: Leading verb of a tool name that only READS. Anything else is treated as possibly
#: mutating, so the classification errs towards NOT flagging. Adapters that know better can
#: say so per call (``{"mutates": false}``), which wins over this heuristic.
# ponytail: name-prefix heuristic; adapters can carry an explicit `mutates` flag instead.
_READ_VERBS = frozenset("""
get list search find read lookup fetch view show query count describe inspect
calculate compute check validate verify think plan note summarize compare
""".split())


def _call_names(rollout: dict) -> list[str]:
    """Every tool name the rollout reports, from the two places core's shape puts them.

    ``Rollout.tool_calls`` is the declared field; a runner that stores an OpenAI-style
    message list in ``Rollout.trace`` carries them per message instead. Both are generic
    core/wire shapes — no runner-specific parsing.
    """
    out: list[str] = []

    def add(call) -> None:
        if isinstance(call, str):
            out.append(call)
        elif isinstance(call, dict):
            fn = call.get("function")
            name = call.get("name") or (fn.get("name") if isinstance(fn, dict) else None)
            if name:
                out.append(str(name))
                if call.get("mutates") is True:
                    out.append("!mutates")     # explicit adapter signal, see _mutates
    for call in rollout.get("tool_calls") or []:
        add(call)
    trace = rollout.get("trace")
    if isinstance(trace, list):
        for msg in trace:
            if not isinstance(msg, dict):
                continue
            for call in msg.get("tool_calls") or []:
                add(call)
    return out


def _mutates(name: str) -> bool:
    if name == "!mutates":
        return True
    head = re.split(r"[^a-z0-9]+", name.strip().lower(), maxsplit=1)[0]
    return head not in _READ_VERBS


def _final_text(rollout: dict) -> str:
    out = rollout.get("output")
    if isinstance(out, str) and out.strip():
        return out
    if out is not None and not isinstance(out, (list, dict)):
        return str(out)
    trace = rollout.get("trace")
    if isinstance(trace, list):
        for msg in reversed(trace):
            if isinstance(msg, dict) and msg.get("content"):
                return str(msg["content"])
    return "" if out is None else str(out)


def narrated_without_action(rollout: dict) -> bool:
    """Did the agent NARRATE a state change it never executed?

    A well-documented LLM-agent failure mode: the model treats its own completion signal
    (usually the user's "yes, go ahead") as satisfying the task and substitutes a narration
    of the change for the call that performs it. Mechanically: the final message claims a
    state change happened, and no tool call in the whole trace could have made one.

    Detected here rather than clustered from feedback because a scorer describes this
    exactly as it describes a wrong write, so the two land in one cluster and the optimizer
    ships an argument fix for a call that was never made.

    Requires at least one observable tool call before flagging anything: with an empty
    tool-call record there is no way to tell "the agent called nothing" from "this adapter
    does not report calls", and guessing would flag every failure on such an adapter.
    """
    names = _call_names(rollout or {})
    if not names or any(_mutates(n) for n in names):
        return False
    return bool(_COMPLETION_RE.search(_final_text(rollout or {})))


def _stem(tok: str) -> str:
    for suf in _SUFFIXES:
        if tok.endswith(suf) and len(tok) - len(suf) >= _MIN_STEM:
            return tok[: -len(suf)]
    return tok


def _words(feedback: str) -> list[str]:
    s = (feedback or "").lower()
    s = re.sub(r"['\"`].*?['\"`]", " ", s)   # quoted literals: task-specific
    s = re.sub(r"[0-9]+", " ", s)            # numbers: task-specific
    s = re.sub(r"[^a-z_ ]+", " ", s)         # punctuation (keep _: identifiers)
    return [t for t in s.split() if len(t) > 2]


def common_prefix(feedbacks: list[str]) -> list[str]:
    """The longest leading token run shared by EVERY feedback (the boilerplate)."""
    seqs = [_words(f) for f in feedbacks if (f or "").strip()]
    if len(seqs) < 2:
        return []
    out: list[str] = []
    for i in range(min(len(s) for s in seqs)):
        tok = seqs[0][i]
        if all(s[i] == tok for s in seqs):
            out.append(tok)
        else:
            break
    # If the shared run covers a whole feedback, the failures are not distinguishable
    # by their prefix at all (identical wording) — stripping would delete the signal
    # rather than boilerplate. Leave it; the generic-word filter still discriminates.
    if out and any(len(s) == len(out) for s in seqs):
        return []
    return out


def _stemmed(feedback: str, prefix: list[str] | None = None) -> list[str]:
    """Content-word stems for one feedback, with the shared boilerplate prefix removed."""
    toks = _words(feedback)
    pre = prefix or []
    if pre and toks[: len(pre)] == pre:
        toks = toks[len(pre):]
    return [_stem(t) for t in toks]


def corpus_stopwords(stemmed_per_item: list[list[str]],
                     threshold: float = CORPUS_STOP_FRAC) -> frozenset[str]:
    """Stems that appear in more than ``threshold`` of the batch's feedbacks.

    Document frequency, not raw count — a token used many times in ONE feedback must not
    count as "common to the batch". Needs at least 2 feedbacks to mean anything."""
    n = len(stemmed_per_item)
    if n < 2:
        return frozenset()
    doc_freq: dict[str, int] = {}
    for toks in stemmed_per_item:
        for t in set(toks):
            doc_freq[t] = doc_freq.get(t, 0) + 1
    return frozenset(t for t, c in doc_freq.items() if c / n > threshold)


def _filter_stems(stems: list[str], extra_stop: frozenset[str] | None = None) -> frozenset[str]:
    """Drop generic/stopword/corpus-boilerplate stems, cascading back if that empties the
    key — a cluster signature must never go blank just because a whole batch's failures
    happen to share their content words too (a corpus of 2 near-identical failures, say)."""
    extra = extra_stop or frozenset()
    keep = [s for s in stems if s not in _STOP and s not in _GENERIC and s not in extra]
    if not keep:                       # the corpus filter alone emptied it: back off
        keep = [s for s in stems if s not in _STOP and s not in _GENERIC]
    if not keep:                       # all-generic feedback: fall back to what we have
        keep = [s for s in stems if s not in _STOP] or stems
    return frozenset(keep)


def key_tokens(feedback: str, prefix: list[str] | None = None,
              extra_stop: frozenset[str] | None = None) -> frozenset[str]:
    """The (site, expectation) key: content-word stems, boilerplate removed."""
    return _filter_stems(_stemmed(feedback, prefix), extra_stop)


def overlap(a: frozenset[str], b: frozenset[str]) -> float:
    if not a or not b:
        return 0.0
    return len(a & b) / min(len(a), len(b))


def cluster(items: list[tuple[str, str, float]]) -> list[dict]:
    """Group ``(task_id, feedback, score_lost)`` triples by root cause.

    Clusters are sorted by score lost, then task count, then signature — a total
    order, so the output is byte-identical on repeated runs over the same input.
    """
    prefix = common_prefix([f for _, f, _ in items])
    stemmed = [_stemmed(f, prefix) for _, f, _ in items]
    # Corpus-relative, IN ADDITION to the generic-English list: a benchmark's own recurring
    # vocabulary ("action check state write") isn't English boilerplate, so no fixed list
    # catches it, but it is exactly as uninformative once it recurs in most of THIS batch's
    # failures. Without this, that vocabulary survives into every key and merges unrelated
    # failures into one mega-cluster with no discriminating signal.
    extra_stop = corpus_stopwords(stemmed)
    keys = [_filter_stems(s, extra_stop) for s in stemmed]

    # Union-find over the overlap relation (transitive: A~B and B~C => one cluster).
    # ponytail: O(n^2) pair scan — fine for a val split; index by token if it grows.
    parent = list(range(len(items)))

    def find(i: int) -> int:
        while parent[i] != i:
            parent[i] = parent[parent[i]]
            i = parent[i]
        return i

    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            if overlap(keys[i], keys[j]) >= OVERLAP_MIN:
                ri, rj = find(i), find(j)
                if ri != rj:
                    parent[max(ri, rj)] = min(ri, rj)

    groups: dict[int, list[int]] = {}
    for i in range(len(items)):
        groups.setdefault(find(i), []).append(i)

    out = []
    for root in sorted(groups):
        idxs = groups[root]
        counts: dict[str, int] = {}
        for i in idxs:
            for t in keys[i]:
                counts[t] = counts.get(t, 0) + 1
        label = " ".join(t for t, _ in
                         sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:4])
        out.append({
            "signature": label or "unknown",
            "tasks": sorted(items[i][0] for i in idxs),
            "score_lost": round(sum(items[i][2] for i in idxs), 4),
            # Judgement, not derivable here — diagnose's reader fills these in.
            "tag": None,
            "blast_radius": None,
        })
    out.sort(key=lambda c: (-c["score_lost"], -len(c["tasks"]), c["signature"]))
    return out

"""diagnose — turn rollouts/scores into an actionable learning signal.

Reads a candidate's persisted rollouts for one split and emits the reflective
dataset (per failing task ``{task_id, Inputs, Generated Outputs, Feedback,
Trajectory}`` — GEPA's shape) plus failure clusters ranked by score lost, plus
``kept_good``. The algorithm/optimizer consume this to know WHAT to change and WHY.

Trace access is schema-agnostic by construction. The rollout record (a cap-evolve
core format) supplies the score and the feedback; the full trace lives wherever the
RUNNER puts it, so its location comes from ``adapter.trajectories(split)`` and is
attached as a PATH only — this script never parses a runner's trace format. With no
``--project``, or when the adapter has no native trajectory store, the pointer falls
back to the rollout record's own file, which core wrote and therefore owns.

Clustering is deterministic and lives in ``cluster.py``; see its docstring for the
(site, expectation) signature. ``--cluster first-words`` keeps the old lexical key
available for comparison.
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path

import _bootstrap  # noqa: F401

import cluster as _cluster

from cap_evolve import RunDir


def _load_records(run_dir: RunDir, tag: str, split: str = "val") -> list[dict]:
    """Every persisted rollout for ``tag`` on ``split``, each tagged with its path.

    ``split`` exists because TRAIN is the honest surface to diagnose from: val is what
    the gate scores, so reading the learning signal off val and then gating on val
    fits the split you are being judged on. Hardcoding "val" here made train-based
    diagnosis unreachable for every caller.
    """
    out = []
    vdir = run_dir.rollouts / split
    if not vdir.exists():
        return out
    for f in sorted(vdir.glob(f"*__{tag}__t*.json")):
        rec = json.loads(f.read_text(encoding="utf-8"))
        rec["__file"] = str(f)
        out.append(rec)
    return out


def trace_dir(project: str | None, split: str) -> str | None:
    """The runner's native trajectory directory, via the adapter — never guessed.

    Returns ``None`` when there is no project to load or the adapter declares no
    native store (the documented default); callers then fall back to the per-rollout
    record. Any adapter failure is non-fatal: a missing trace pointer must not stop a
    diagnosis that the rollouts alone can still produce.
    """
    if not project:
        return None
    try:
        from cap_evolve.check import load_adapter
        d = load_adapter(Path(project)).trajectories(split)
        return str(d) if d else None
    except Exception:  # noqa: BLE001
        return None


def first_n_words_signature(feedback: str, n: int = 6) -> str:
    """Legacy lexical clustering key (opt-in via ``--cluster first-words``)."""
    return " ".join((feedback or "").split()[:n]) or "unknown"


def diagnose(records: list[dict], mode: str = "root-cause",
             traces: str | None = None) -> dict:
    reflective = []
    items: list[tuple[str, str, float]] = []
    narrated: list[tuple[str, str, float]] = []
    kept = []
    for rec in records:
        sc = rec.get("score", {})
        ro = rec.get("rollout", {})
        reward = sc.get("reward", 0) or 0
        if reward >= 1.0:
            kept.append(sc.get("task_id"))
            continue
        fb = sc.get("feedback", "") or ""
        # Mechanical, trace-derived, and benchmark-agnostic — see cluster.narrated_without_action.
        flagged = _cluster.narrated_without_action(ro)
        reflective.append({
            "task_id": sc.get("task_id"),
            # The actual task INPUT (carried through the rollout file), NOT the id.
            "Inputs": rec.get("input"),
            "Generated Outputs": ro.get("output"),
            "Feedback": fb,
            # Where the FULL trace is. A path, not parsed content: the format is the
            # runner's business and the adapter's to expose.
            "Trajectory": traces or rec.get("__file"),
            _cluster.NARRATED_WITHOUT_ACTION: flagged,
        })
        row = (sc.get("task_id"), fb, max(0.0, 1.0 - float(reward)))
        (narrated if flagged else items).append(row)

    if mode == "first-words":
        groups = defaultdict(list)
        lost = defaultdict(float)
        for tid, fb, sl in items:
            k = first_n_words_signature(fb)
            groups[k].append(tid)
            lost[k] += sl
        clusters = [{"signature": k, "tasks": sorted(v),
                     "score_lost": round(lost[k], 4), "tag": None, "blast_radius": None}
                    for k, v in groups.items()]
        clusters.sort(key=lambda c: (-c["score_lost"], -len(c["tasks"]), c["signature"]))
    else:
        clusters = _cluster.cluster(items)

    # Its own named cluster, never folded into a lexical one: the scorer's wording for
    # "narrated a change it never made" is the same wording it uses for a wrong write, so
    # merging them sends the optimizer to fix the arguments of a call that never happened.
    if narrated:
        clusters.append({
            "signature": _cluster.NARRATED_WITHOUT_ACTION,
            "tasks": sorted(t for t, _, _ in narrated),
            "score_lost": round(sum(sl for _, _, sl in narrated), 4),
            "detector": "mechanical: completion language in the final message, no mutating "
                        "tool call anywhere in the trace",
            "reading": "the agent treated its own completion signal (typically the user's "
                       "confirmation) as the action and narrated the change instead of "
                       "executing it. A prose reminder to call the tool does not fix this; "
                       "the fix is structural — make 'confirmed' and 'executed' the same "
                       "call, so no code path can reach one without the other.",
            "tag": None,
            "blast_radius": None,
        })
        clusters.sort(key=lambda c: (-c["score_lost"], -len(c["tasks"]), c["signature"]))

    return {
        "reflective_dataset": reflective,
        "clusters": clusters,
        "kept_good": kept,
    }


def main(argv=None) -> int:
    p = argparse.ArgumentParser(prog="diagnose")
    p.add_argument("--run-dir", required=True)
    p.add_argument("--tag", default="seed", help="candidate tag whose rollouts to read")
    p.add_argument("--project", default=None,
                   help="project dir — resolves the runner's native trace dir via "
                        "adapter.trajectories(split); optional")
    p.add_argument("--split", default="val", choices=["train", "val"],
                   help="which split's rollouts to diagnose (train is the honest "
                        "learning surface; val is what the gate scores)")
    p.add_argument("--cluster", default="root-cause",
                   choices=["root-cause", "first-words"],
                   help="failure-clustering method (root-cause: site+expectation key)")
    args = p.parse_args(argv)
    run_dir = RunDir.open(Path(args.run_dir))
    result = diagnose(_load_records(run_dir, args.tag, args.split),
                      args.cluster, trace_dir(args.project, args.split))
    result["split"] = args.split
    result["tag"] = args.tag
    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Thin shim: locate cap_evolve, then defer to cap_evolve._bootstrap.

Skill scripts ``import _bootstrap`` first. The real path-resolution logic lives
ONCE in ``cap_evolve._bootstrap`` (so it can't drift across skills); this shim
only has to find that package, which means a minimal upward walk for ``core/`` —
the single bit of bootstrapping that genuinely must run before cap_evolve is
importable. Everything else delegates.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path


def _seed_path() -> None:
    """Minimal: put a dir containing the cap_evolve package on sys.path.

    ``CAPEVOLVE_CORE`` is honoured BEFORE any ambient import. An editable install of a
    *different* cap-evolve checkout registers a ``sys.meta_path`` finder, which outranks
    both ``sys.path`` and ``PYTHONPATH`` — so "cap_evolve imports fine" is not evidence that
    it imports the checkout you are standing in. Deferring to the ambient package here made
    an explicit override unreachable, and the symptom was a stale core silently answering
    for this one (``ModuleNotFoundError: cap_evolve.constraints`` from a checkout that
    predates that module). An explicit env var wins.
    """
    env = os.environ.get("CAPEVOLVE_CORE")
    want = Path(env).resolve() if env else None
    if want and (want / "cap_evolve" / "__init__.py").exists():
        loaded = sys.modules.get("cap_evolve")
        already = getattr(loaded, "__file__", None)
        if already and Path(already).resolve().parent.parent == want:
            return                      # right checkout already imported: touch nothing
        p = str(want)
        if p in sys.path:
            sys.path.remove(p)
        sys.path.insert(0, p)
        if loaded is not None:
            # Evicting a module makes a re-import yield a DIFFERENT object, so anything
            # already holding a reference fails an `is` check. Only ever do it when the
            # loaded package really is the wrong checkout — otherwise this "fix" becomes
            # the bug (it broke two identity assertions in core/tests exactly once).
            for name in [m for m in sys.modules
                         if m == "cap_evolve" or m.startswith("cap_evolve.")]:
                sys.modules.pop(name, None)
        for finder in list(sys.meta_path):
            if "cap_evolve" in getattr(finder, "MAPPING", {}):
                sys.meta_path.remove(finder)
        return
    # A checkout's own core outranks an ambient install. Without this, a skill script run
    # from checkout X silently executed against checkout Y's cap_evolve (an editable install
    # registers a sys.meta_path finder, which outranks sys.path), and the only symptom was
    # missing modules — or, worse, a green result measured against the wrong tree.
    here = Path(__file__).resolve()
    own = next((p / "core" for p in here.parents
                if (p / "core" / "cap_evolve" / "__init__.py").exists()), None)
    if own is not None:
        os.environ.setdefault("CAPEVOLVE_CORE", str(own))
        return _seed_path()
    try:
        import cap_evolve  # noqa: F401
        return
    except Exception:
        pass
    cands = []
    for parent in here.parents:
        cands.append(parent / "core")
        cands.append(parent)
    for c in cands:
        if (c / "cap_evolve" / "__init__.py").exists():
            p = str(c)
            if p not in sys.path:
                sys.path.insert(0, p)
            return


_seed_path()
from cap_evolve._bootstrap import ensure_core  # noqa: E402

# Anchor the upward walk at THIS skill script's location (not the core module's).
ensure_core(Path(__file__).resolve())