feat(catalog): trigger-probe in real Claude Code; payload leak audit

dokimi/changelog.d/20261024_120000_napetrov_trigger-fidelity.md

@@ -0,0 +1,10 @@
+### Added
+
+- **`catalog trigger-probe`** — runs each trigger-set query in a fresh, unannounced
+  `claude -p` session with the catalog committed as project skills, a scratch `HOME`
+  and config, and the query sent verbatim; scores the `Skill` calls with the same
+  selection summary as the `skill-catalog:` arm, one column per trial.
+- **`dokimi.harnesses.payload_audit`** — `audit_payload` / `new_leaks` report dokimi
+  artifacts (arm names, label keys, question ids, grading words, task paths) in a
+  captured request, argv or env, with operator content blanked and the stock harness's
+  own text subtracted.

dokimi/docs/evaluating-treatments.md

@@ -582,6 +582,23 @@ uv run reach eval --skills <skills-checkout>/skills --queries trigger_set.json \
 uv run reach diff --vary description control.json treatment.json   # A/B vs noise floor
 ```
 
+**Real harness, in dokimi: `catalog trigger-probe`.** Runs each query in a fresh
+`claude -p`: a new git repository with the catalog committed in `.claude/skills/`, a
+scratch `HOME` and `CLAUDE_CONFIG_DIR` (your own skills, settings and `CLAUDE.md` are not
+loaded; the model can still read any host path it has access to, the trigger set and its
+labels included), the query sent verbatim with no notice or task framing, and bundled skills
+left on as a user has them (`--sealed` turns them off). The skill chosen is read from the
+transcript's `Skill` calls, and the output's `skill_selection_summary` is scored by the
+same code as the `skill-catalog:` arm, one column per trial. `max_turns_hit` counts runs
+that ended on `--max-turns` (scored as no load), and the command exits 1 when Claude Code
+did not list a catalog skill (`not_listed`). Records are kept in `<out>.partial` while the
+probe runs:
+
+```bash
+python cli.py catalog trigger-probe --catalog <skills-checkout>/skills \
+  --trigger-set trigger_set.json --model <model> --trials 3 --out probe.json
+```
+
 Two things decide whether such a number means anything:
 
 - **Listing budget.** Claude Code shows skill descriptions only up to a budget (a

dokimi/dokimi/commands/catalog.py

@@ -364,3 +364,7 @@ def register(sub, positive_int) -> None:
                       help="Exit 1 if any query names a skill, keeps exam framing, or has an unresolved @field label")
     ts_p.add_argument("--out", required=True, help="Output JSON path")
     ts_p.set_defaults(func=cmd_catalog_trigger_set)
+
+    from dokimi.commands import catalog_probe
+
+    catalog_probe.register(catalog_sub, positive_int)

dokimi/dokimi/commands/catalog_probe.py

@@ -0,0 +1,149 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""``catalog trigger-probe``: run a trigger set through real Claude Code, one fresh run per query."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+
+def cmd_catalog_trigger_probe(args: argparse.Namespace) -> None:
+    from dokimi.eval.arm_summary import summarize_skill_selection
+    from dokimi.harnesses import trigger_probe as tp
+    from dokimi.skills import discover_skills
+    from dokimi.utils import InputError
+
+    catalog = Path(args.catalog)
+    if not catalog.is_dir():
+        raise InputError(f"--catalog is not a directory: {catalog}")
+    try:
+        names = {s.name for s in discover_skills(catalog)}
+    except (OSError, ValueError) as exc:
+        raise InputError(f"--catalog: {exc}") from exc
+    if args.target and args.target not in names:
+        raise InputError(f"--target {args.target!r} is not a skill in --catalog "
+                         f"(have: {', '.join(sorted(names))})")
+    # Checked before any paid run, so a bad --out costs nothing.
+    out_path = Path(args.out)
+    partial = out_path.with_name(out_path.name + ".partial")
+    if out_path.is_dir() or not out_path.parent.is_dir() or not os.access(out_path.parent, os.W_OK):
+        raise InputError(f"--out: not a writable file path: {out_path}")
+    if not args.timeout > 0:
+        raise InputError(f"--timeout must be > 0, got {args.timeout}")
+    # Absolute: each run's cwd is a scratch dir, where a relative path names nothing.
+    claude = shutil.which(args.claude)
+    if claude is None:
+        raise SystemExit(f"{args.claude!r} is not on PATH")
+    queries = _labelled_queries(args, names)
+    cfg = tp.ProbeConfig(model=args.model, max_turns=args.max_turns, timeout_sec=args.timeout,
+                         stock=not args.sealed, listing_budget_fraction=args.listing_budget_fraction,
+                         claude=os.path.abspath(claude))
+
+    done: list = []
+    lock = threading.Lock()
+
+    def one(q):
+        rec = tp.probe_trigger_set([q], catalog, cfg, trials=args.trials)[0]
+        with lock:  # what has run so far survives a later failure or a kill
+            done.append(rec)
+            _write(partial, {"partial": True, "answers": done})
+        print(f"{q.id}: " + ", ".join(
+            f"{a}={'+'.join(v['tool_usage']['skill']['first_turn_loaded']) or '-'}"
+            + (f" ({v['error']})" if v["error"] else "") for a, v in rec["arms"].items()),
+            file=sys.stderr)
+        return rec
+
+    with ThreadPoolExecutor(max(1, args.concurrency)) as pool:
+        records = list(pool.map(one, queries))
+    runs = [v for r in records for v in r["arms"].values()]
+    listed = {s.rsplit(":", 1)[-1] for v in runs for s in v["tool_usage"]["skill"]["listed"]}
+    # A skill the harness never offered (a frontmatter name that is not its directory's,
+    # a nested layout) reads as a 0% recall; say so instead of scoring it.
+    not_listed = sorted(names - listed) if listed else []
+    cost = sum(v["metrics"]["cost_usd"] or 0 for r in records for v in r["arms"].values())
+    out = {
+        "harness": "claude-code",
+        "mode": "sealed" if args.sealed else "stock",
+        "model": args.model,
+        "catalog": str(catalog),
+        "trials": args.trials,
+        "max_turns": args.max_turns,
+        "listing_budget_fraction": args.listing_budget_fraction,
+        "claude_code_version": next((v["tool_usage"]["skill"]["claude_code_version"] for r in records
+                                     for v in r["arms"].values()
+                                     if v["tool_usage"]["skill"]["claude_code_version"]), None),
+        "cost_usd": round(cost, 4),
+        "max_turns_hit": sum(v["metrics"].get("result_subtype") == "error_max_turns" for v in runs),
+        "not_listed": not_listed,
+        "answers": records,
+        "skill_selection_summary": summarize_skill_selection(records),
+    }
+    _write(out_path, out)
+    partial.unlink(missing_ok=True)
+    print(json.dumps({"cost_usd": out["cost_usd"], "claude_code_version": out["claude_code_version"],
+                      "max_turns_hit": out["max_turns_hit"], "not_listed": not_listed,
+                      "summary": {a: {k: s[k] for k in ("recall", "false_trigger_rate", "misroute_rate", "n_failed")}
+                                  for a, s in out["skill_selection_summary"].items()}}, indent=2))
+    print(f"wrote {args.out}")
+    if not_listed:
+        raise SystemExit(f"catalog skills Claude Code did not list: {', '.join(not_listed)}")
+
+
+def _labelled_queries(args: argparse.Namespace, names: set) -> list:
+    """The labelled queries to run; a label naming no *names* skill is an input error."""
+    from dokimi.external import trigger_set as ts
+    from dokimi.utils import InputError
+
+    try:
+        data = json.loads(Path(args.trigger_set).read_text(encoding="utf-8"))
+        _fmt, queries = ts.read_trigger_set(data, target=args.target)
+    except (OSError, ValueError) as exc:
+        raise SystemExit(f"--trigger-set: {exc}") from exc
+    queries = [q for q in queries if q.labelled][: args.limit or None]
+    # As the skill-catalog: arm refuses it: a label no run can load scores as a miss.
+    unknown = sorted({s for q in queries for s in q.acceptable} - names)
+    if unknown:
+        raise InputError(f"--trigger-set: expected_skill names no skill in --catalog: "
+                         f"{', '.join(unknown)}")
+    return queries
+
+
+def _write(path: Path, data: dict) -> None:
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+    os.replace(tmp, path)
+
+
+def register(catalog_sub, positive_int) -> None:
+    """Add ``catalog trigger-probe`` to the ``catalog`` group."""
+    p = catalog_sub.add_parser(
+        "trigger-probe",
+        help="Run each trigger-set query in a fresh, unannounced Claude Code session with the "
+             "catalog installed as project skills; score which skill it invokes",
+    )
+    p.add_argument("--catalog", required=True, help="Directory of skills, installed as .claude/skills")
+    p.add_argument("--trigger-set", required=True, dest="trigger_set",
+                   help="skill-reach QuerySet, skill-creator set (with --target) or questions with expected_skill")
+    p.add_argument("--target", help="Target skill of a skill-creator set")
+    p.add_argument("--model", help="--model passed to claude (default: the harness default)")
+    p.add_argument("--trials", type=positive_int, default=1, help="Fresh runs per query (default: 1)")
+    p.add_argument("--max-turns", type=positive_int, default=3, dest="max_turns",
+                   help="claude --max-turns (default: 3); runs that hit it are counted in max_turns_hit")
+    p.add_argument("--timeout", type=float, default=300.0, help="Seconds per run (default: 300)")
+    p.add_argument("--concurrency", type=positive_int, default=4, help="Queries run at once (default: 4)")
+    p.add_argument("--limit", type=positive_int, help="Only the first N labelled queries")
+    p.add_argument("--listing-budget-fraction", type=float, dest="listing_budget_fraction",
+                   help="skillListingBudgetFraction for the run (default: the harness's own, 0.01)")
+    p.add_argument("--sealed", action="store_true",
+                   help="Switch off bundled skills as the docker-claude arms do (default: stock harness)")
+    p.add_argument("--claude", default="claude", help="claude executable (default: claude)")
+    p.add_argument("--out", required=True, help="Output JSON path")
+    p.set_defaults(func=cmd_catalog_trigger_probe)

dokimi/dokimi/harnesses/payload_audit.py

@@ -0,0 +1,134 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Leak audit: does what an agent is sent carry dokimi's own artifacts?
+
+A measured run is meant to reproduce what a user's agent sees. Anything dokimi adds
+that names the experiment (an arm name, the ``expected_skill`` key, a question id,
+judge or rubric text, a task path that says what is graded) can change what the model
+does, and makes the number measure dokimi rather than the harness.
+
+:func:`audit_payload` walks a payload (a captured request body, an argv, an env map:
+any JSON-like value) and reports every marker it finds. Text the operator supplied
+on purpose (the query, the skills' own descriptions and bodies) is passed as
+``allowed`` and blanked out before matching: a skill whose description says
+"baseline" is content, not a leak. What is left is what dokimi wrote.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any, Iterable, Iterator, List, Optional, Sequence, Tuple
+
+#: Words and keys that name an experiment rather than a user's task. Matched whole-word,
+#: case-insensitive, after the allowed texts are blanked.
+DEFAULT_MARKERS: Tuple[str, ...] = (
+    "dokimi", "expected_skill", "acceptable_skills", "should_trigger", "trigger_set",
+    "skill_catalog", "skill-catalog", "skill_selection", "treatment", "baseline",
+    "with_docs", "without_docs", "judge", "rubric", "golden", "grader", "verifier",
+    "reward", "oracle", "eval", "evals", "evaluation", "benchmark", "arm_kind",
+    "abskill", "test_outputs",
+)
+
+
+@dataclass(frozen=True)
+class Leak:
+    """One marker found: where (a JSON path), which marker, and the text around it."""
+
+    path: str
+    marker: str
+    excerpt: str
+
+    def as_dict(self) -> dict:
+        return {"path": self.path, "marker": self.marker, "excerpt": self.excerpt}
+
+
+def markers_for(*, question_ids: Iterable[str] = (), arm_names: Iterable[str] = (),
+                paths: Iterable[str] = (), extra: Iterable[str] = (),
+                base: Sequence[str] = DEFAULT_MARKERS) -> List[str]:
+    """The default markers plus this run's own identifiers.
+
+    *question_ids* and *arm_names* come from the run; *paths* are host paths that
+    reveal intent (a task directory, an output dir). Empty values are dropped.
+    """
+    out = list(base)
+    for group in (question_ids, arm_names, paths, extra):
+        out += [str(v) for v in group if str(v).strip()]
+    return out
+
+
+def _strings(value: Any, path: str = "$") -> Iterator[Tuple[str, str]]:
+    """Every string in *value* (keys included) with its JSON path."""
+    if isinstance(value, str):
+        yield path, value
+    elif isinstance(value, dict):
+        for k, v in value.items():
+            yield f"{path}.{k}#key", str(k)
+            yield from _strings(v, f"{path}.{k}")
+    elif isinstance(value, (list, tuple)):
+        for i, v in enumerate(value):
+            yield from _strings(v, f"{path}[{i}]")
+
+
+def _pattern(marker: str) -> "re.Pattern[str]":
+    # Identifier-ish markers match between non-alphanumerics, so `dokimi-run-x` and
+    # `eval_queries` are caught while `evaluate` is not; anything else is a substring.
+    if re.fullmatch(r"[\w-]+", marker):
+        return re.compile(rf"(?<![A-Za-z0-9]){re.escape(marker)}(?![A-Za-z0-9])", re.IGNORECASE)
+    return re.compile(re.escape(marker), re.IGNORECASE)
+
+
+def _blank(text: str, allowed: Sequence[str]) -> str:
+    for a in allowed:
+        if a and a in text:
+            text = text.replace(a, " " * len(a))
+    return text
+
+
+def audit_payload(payload: Any, markers: Optional[Iterable[str]] = None,
+                  allowed: Iterable[str] = ()) -> List[Leak]:
+    """Markers found in *payload* outside the *allowed* texts, one :class:`Leak` each.
+
+    *allowed* is longest-first blanked, so a skill body containing its description
+    blanks as a whole. An empty result means the payload carries none of the markers.
+    """
+    pats = [(m, _pattern(m)) for m in (DEFAULT_MARKERS if markers is None else markers) if m]
+    keep = sorted({a for a in allowed if a and a.strip()}, key=len, reverse=True)
+    leaks: List[Leak] = []
+    for path, text in _strings(payload):
+        visible = _blank(text, keep)
+        for marker, pat in pats:
+            for m in pat.finditer(visible):
+                lo, hi = max(0, m.start() - 40), min(len(text), m.end() + 40)
+                leaks.append(Leak(path, marker, text[lo:hi].replace("\n", " ")))
+    return leaks
+
+
+def added_framing(prompt: str, query: str) -> str:
+    """What a harness put in the prompt around the user's *query* (``""`` if verbatim).
+
+    The framing is not a leak by the marker test, but it is an influence: an
+    out-of-the-box run sends the query alone.
+    """
+    if prompt == query:
+        return ""
+    if query and query in prompt:
+        head, _, tail = prompt.partition(query)
+        return (head + tail).strip()
+    return prompt
+
+
+def new_leaks(payload: Any, reference: Any, markers: Optional[Iterable[str]] = None,
+              allowed: Iterable[str] = ()) -> List[Leak]:
+    """Leaks in *payload* that a *reference* capture does not have.
+
+    The reference is the same harness run out of the box (no dokimi in the loop): its
+    own system prompt says "judge edge cases" and "golden path", which is the
+    harness, not the experiment. A marker counts as new when its excerpt is not
+    among the reference's.
+    """
+    marks = list(DEFAULT_MARKERS if markers is None else markers)
+    allowed = list(allowed)
+    seen = {(lk.marker, lk.excerpt) for lk in audit_payload(reference, marks, allowed)}
+    return [lk for lk in audit_payload(payload, marks, allowed) if (lk.marker, lk.excerpt) not in seen]