diff --git a/dokimi/changelog.d/20261024_120000_napetrov_trigger-fidelity.md b/dokimi/changelog.d/20261024_120000_napetrov_trigger-fidelity.md
new file mode 100644
index 000000000..061b10536
--- /dev/null
+++ b/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.
diff --git a/dokimi/docs/evaluating-treatments.md b/dokimi/docs/evaluating-treatments.md
index 493bd782f..ee3a71cf2 100644
--- a/dokimi/docs/evaluating-treatments.md
+++ b/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
diff --git a/dokimi/dokimi/commands/catalog.py b/dokimi/dokimi/commands/catalog.py
index 06ac62262..94991d031 100644
--- a/dokimi/dokimi/commands/catalog.py
+++ b/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)
diff --git a/dokimi/dokimi/commands/catalog_probe.py b/dokimi/dokimi/commands/catalog_probe.py
new file mode 100644
index 000000000..b149613b5
--- /dev/null
+++ b/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)
diff --git a/dokimi/dokimi/harnesses/payload_audit.py b/dokimi/dokimi/harnesses/payload_audit.py
new file mode 100644
index 000000000..d83c1e4e3
--- /dev/null
+++ b/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]
diff --git a/dokimi/dokimi/harnesses/trigger_probe.py b/dokimi/dokimi/harnesses/trigger_probe.py
new file mode 100644
index 000000000..328078be1
--- /dev/null
+++ b/dokimi/dokimi/harnesses/trigger_probe.py
@@ -0,0 +1,261 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Real-harness skill triggering: one fresh ``claude -p`` per query, no notice.
+
+The ``skill-catalog:`` arm is a tool-calling emulation: one tool per skill, every
+description in full, no system prompt, no other tools. Claude Code instead sends one
+``Skill`` tool, a listing it cuts to a budget, its own system prompt and a dozen other
+tools, and the model may look around before it loads anything. This module measures
+that harness as shipped:
+
+* each query runs in a new git repository under a temp dir (no ancestor ``CLAUDE.md``,
+  no checkout to walk), with the catalog committed in ``.claude/skills/``, where
+  Claude Code looks for project skills, so ``git status`` shows nothing unusual;
+* ``HOME`` and ``CLAUDE_CONFIG_DIR`` are fresh and the env is the sandbox allowlist,
+  so the operator's skills, settings and ``CLAUDE.md`` are not loaded (the model can
+  still read any host path it is allowed to, the trigger set included);
+* the prompt is the query verbatim: no treatment notice, no task framing;
+* in ``stock`` mode (the default) the bundled skills and tools stay on, as a user
+  has them; ``sealed`` switches them off like the docker-claude arms do.
+
+The skills invoked are read from the ``stream-json`` transcript (``Skill`` tool uses);
+the first assistant turn that invokes any is the choice, as for the emulated arm.
+Records come out in the answers shape :func:`dokimi.eval.arm_summary.summarize_skill_selection`
+reads, so both arms are scored by the same code.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import signal
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence
+
+from dokimi import process
+from dokimi.harnesses.agent_sandbox import (
+    _SEAL_ENV, _SEAL_SETTINGS, AgentWorkspace, agent_env, operator_source_env,
+)
+from dokimi.harnesses.claude_json import result_event
+
+#: Fixed identity and dates for the staging commit, so the git context the harness
+#: shows the model is the same on every run.
+_GIT_IDENT = {
+    "GIT_AUTHOR_NAME": "user", "GIT_AUTHOR_EMAIL": "user@example.com",
+    "GIT_COMMITTER_NAME": "user", "GIT_COMMITTER_EMAIL": "user@example.com",
+    "GIT_AUTHOR_DATE": "2026-01-01T00:00:00Z", "GIT_COMMITTER_DATE": "2026-01-01T00:00:00Z",
+}
+_COMMIT_MESSAGE = "Initial commit"
+
+
+@dataclass
+class ProbeConfig:
+    """How each query is run. ``listing_budget_fraction`` ``None`` keeps the harness default."""
+
+    model: Optional[str] = None
+    max_turns: int = 3
+    timeout_sec: float = 300.0
+    stock: bool = True
+    listing_budget_fraction: Optional[float] = None
+    claude: str = "claude"
+    extra_args: List[str] = field(default_factory=list)
+
+
+#: Prefix of each run's temp root. Claude Code shows the model its working directory,
+#: so the name must not say what the run is (``AgentWorkspace.create`` uses ``dokimi-run-``).
+WORKSPACE_PREFIX = "project-"
+
+
+def new_workspace() -> AgentWorkspace:
+    """A scratch ``{work,home,config}`` root whose path names no experiment."""
+    root = Path(tempfile.mkdtemp(prefix=WORKSPACE_PREFIX))
+    for sub in ("work", "home", "config"):
+        (root / sub).mkdir()
+    return AgentWorkspace(root=root)
+
+
+def stage_workspace(catalog: Path, ws: AgentWorkspace, runner: process.Runner = process.run) -> Path:
+    """Copy *catalog* to ``work/.claude/skills`` and commit it; return the work dir."""
+    work = ws.work
+    shutil.copytree(catalog, work / ".claude" / "skills", symlinks=False)
+    env = process.Env.only(("PATH",), {**_GIT_IDENT, "HOME": str(ws.root / "home"),
+                                       "GIT_CONFIG_NOSYSTEM": "1", "GIT_TERMINAL_PROMPT": "0"})
+    for args in (("init", "-q", "-b", "main"), ("add", "-A"),
+                 ("commit", "-q", "--no-gpg-sign", "-m", _COMMIT_MESSAGE)):
+        done = process.git(*args, cwd=work, env=env, runner=runner)
+        if done.returncode != 0:
+            raise RuntimeError(f"git {args[0]} failed staging the catalog: {process.git_failure(done)}")
+    return work
+
+
+def probe_env(ws: AgentWorkspace, cfg: ProbeConfig, source: Optional[Dict[str, str]] = None) -> Dict[str, str]:
+    """The allowlisted env, scratch HOME/config; the bundled-skill seal only when not ``stock``."""
+    env = agent_env(operator_source_env() if source is None else source)
+    if cfg.stock:
+        for k in _SEAL_ENV:
+            env.pop(k, None)
+    env["HOME"] = str(ws.root / "home")
+    env["CLAUDE_CONFIG_DIR"] = str(ws.root / "config")
+    env["TMPDIR"] = str(ws.root / "home")
+    return env
+
+
+def probe_argv(query: str, cfg: ProbeConfig) -> List[str]:
+    """``claude`` argv: the query verbatim, stream-json out, and nothing that steers the model."""
+    argv = [cfg.claude, "-p", query, "--output-format", "stream-json", "--verbose",
+            "--max-turns", str(cfg.max_turns), "--no-session-persistence"]
+    if cfg.model:
+        argv += ["--model", cfg.model]
+    # One --settings: sealed carries the arms' plugin seal, merged with any listing budget.
+    settings: Dict[str, Any] = {} if cfg.stock else json.loads(json.dumps(_SEAL_SETTINGS))
+    if cfg.listing_budget_fraction is not None:
+        settings["skillListingBudgetFraction"] = cfg.listing_budget_fraction
+    if settings:
+        argv += ["--settings", json.dumps(settings)]
+    return argv + list(cfg.extra_args)
+
+
+def _events(stream: str) -> Iterable[Dict[str, Any]]:
+    for line in stream.splitlines():
+        line = line.strip()
+        if not line.startswith("{"):
+            continue
+        try:
+            obj = json.loads(line)
+        except ValueError:
+            continue
+        if isinstance(obj, dict):
+            yield obj
+
+
+def _skill_name(raw: Any) -> str:
+    # A plugin skill is invoked as `plugin:skill`; the catalog names the skill alone.
+    return str(raw or "").rsplit(":", 1)[-1]
+
+
+def parse_stream(stream: str) -> Dict[str, Any]:
+    """Skills listed and invoked, per the transcript; ``first_turn_loaded`` is the choice."""
+    listed: List[str] = []
+    version = None
+    # Claude Code emits one assistant event per content block; the blocks of one turn
+    # share message.id. Subagent events (parent_tool_use_id set) are not the choice.
+    by_turn: Dict[str, List[str]] = {}
+    other_tools: List[str] = []
+    for n, ev in enumerate(_events(stream)):
+        if ev.get("type") == "system" and ev.get("subtype") == "init":
+            listed = [str(s) for s in ev.get("skills") or []]
+            version = ev.get("claude_code_version")
+        if ev.get("type") != "assistant" or ev.get("parent_tool_use_id"):
+            continue
+        message = ev.get("message") or {}
+        uses = [c for c in message.get("content") or [] if isinstance(c, dict) and c.get("type") == "tool_use"]
+        skills = [_skill_name((c.get("input") or {}).get("skill")) for c in uses if c.get("name") == "Skill"]
+        if not by_turn:  # tools after the choice are not "before" it
+            other_tools += [str(c.get("name")) for c in uses if c.get("name") != "Skill"]
+        if skills:
+            turn = by_turn.setdefault(str(message.get("id") or f"#{n}"), [])
+            turn += [s for s in dict.fromkeys(skills) if s not in turn]
+    turns = list(by_turn.values())
+    first = turns[0] if turns else []
+    return {
+        "offered": True,
+        "listed": listed,
+        "selected": first[0] if first else None,
+        "first_turn_loaded": first,
+        "loaded": [s for t in turns for s in t],
+        "tools_before_choice": other_tools,
+        "claude_code_version": version,
+    }
+
+
+def run_query(query: str, catalog: Path, cfg: ProbeConfig, *, runner: process.Runner = process.run,
+              source_env: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
+    """One fresh harness run of *query*; the parsed skill block plus cost and any error."""
+    ws = new_workspace()
+    try:
+        work = stage_workspace(catalog, ws, runner=runner)
+        argv = probe_argv(query, cfg)
+        done = runner(argv, cwd=work, timeout=cfg.timeout_sec,
+                      env=process.Env.only((), probe_env(ws, cfg, source_env)), input="",
+                      capture_limit=None)
+        stream = done.stdout
+        skill = parse_stream(stream)
+        result = result_event(stream) or {}
+        error = None
+        # A Skill call seen is the choice, whatever came after it (a hang, a timeout, an
+        # error). A result event read decides the rest, even if the CLI then hung.
+        if skill["first_turn_loaded"]:
+            pass
+        elif not result:
+            error = "timeout" if done.timed_out else f"no result event (exit {done.returncode})"
+        # Out of turns without a Skill call is an observation (nothing triggered), not a failure.
+        elif result.get("is_error") and result.get("subtype") != "error_max_turns":
+            error = str(result.get("subtype") or "error")
+        return {"skill": skill, "cost_usd": result.get("total_cost_usd"),
+                "num_turns": result.get("num_turns"), "result_subtype": result.get("subtype"), "error": error}
+    finally:
+        kill_strays(ws.root)
+        ws.cleanup()
+
+
+def kill_strays(root: Path) -> List[int]:
+    """SIGKILL every process whose cwd is under *root*; return their pids (Linux only).
+
+    A command the model backgrounds in its Bash tool runs in a session of its own, so
+    the timeout's process-group kill misses it and it outlives the run and its workspace.
+    """
+    if not Path("/proc").is_dir():
+        return []
+    prefix = str(Path(root).resolve())
+    killed = []
+    for entry in os.listdir("/proc"):
+        if not entry.isdigit() or int(entry) == os.getpid():
+            continue
+        try:
+            cwd = os.readlink(f"/proc/{entry}/cwd")
+        except OSError:  # exited meanwhile, or not ours to read
+            continue
+        if cwd == prefix or cwd.startswith(prefix + os.sep):
+            try:
+                os.kill(int(entry), signal.SIGKILL)
+                killed.append(int(entry))
+            except OSError:
+                pass
+    return killed
+
+
+def probe_trigger_set(queries: Sequence[Any], catalog: Path, cfg: ProbeConfig, *, trials: int = 1,
+                      arm_name: str = "claude_code", runner: process.Runner = process.run,
+                      on_record: Optional[Callable[[Dict[str, Any]], None]] = None) -> List[Dict[str, Any]]:
+    """Run every labelled query *trials* times; answers-shaped records for the selection summary.
+
+    *queries* are :class:`dokimi.external.trigger_set.TriggerQuery`. Each trial is its own
+    arm (``<arm_name>#<k>``), so the summary reports trials side by side.
+    """
+    out: List[Dict[str, Any]] = []
+    for q in queries:
+        if not getattr(q, "labelled", True):
+            continue
+        arms = {}
+        for k in range(trials):
+            try:
+                r = run_query(q.text, catalog, cfg, runner=runner)
+            except (OSError, RuntimeError) as exc:
+                # One failed run (staging, spawn) is that trial's error, not the lost set.
+                r = {"skill": parse_stream(""), "cost_usd": None, "num_turns": None, "result_subtype": None,
+                     "error": f"{type(exc).__name__}: {exc}"}
+            arms[f"{arm_name}#{k + 1}"] = {
+                "answer": None,
+                "error": r["error"],
+                "tool_usage": {"skill": r["skill"]},
+                "metrics": {"cost_usd": r["cost_usd"], "num_turns": r["num_turns"], "result_subtype": r.get("result_subtype")},
+            }
+        rec = {"question_id": q.id, "question_text": q.text, "expected_skill": q.expected_skill, "arms": arms}
+        out.append(rec)
+        if on_record:
+            on_record(rec)
+    return out
diff --git a/dokimi/tests/test_payload_audit.py b/dokimi/tests/test_payload_audit.py
new file mode 100644
index 000000000..1d2d284b2
--- /dev/null
+++ b/dokimi/tests/test_payload_audit.py
@@ -0,0 +1,129 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Leak audit: markers in what an agent is sent, applied to the skill-catalog and docker-claude paths."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import dokimi.eval.agent_runner as ar_mod
+from dokimi.eval.agent_runner import run_agent_loop
+from dokimi.harnesses import load_task
+from dokimi.harnesses.docker_solver import DockerSolverHarness
+from dokimi.harnesses.payload_audit import added_framing, audit_payload, markers_for, new_leaks
+from dokimi.harnesses.skill_treatment import TREATMENT_NOTICE, Treatment
+from dokimi.skills import load_skill
+from dokimi.treatments.arms import SkillCatalogTreatment
+
+
+def _skill(root: Path, name: str, description: str, body: str) -> Path:
+    d = root / name
+    d.mkdir(parents=True)
+    (d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: {description}\n---\n{body}\n",
+                                encoding="utf-8")
+    return d / "SKILL.md"
+
+
+def test_planted_markers_are_found_with_their_path():
+    payload = {"messages": [{"role": "user", "content": "Q-17: how do I sum?"}],
+               "meta": {"expected_skill": "x"}, "cwd": "/srv/dokimi-run-ab12/work"}
+    leaks = audit_payload(payload, markers_for(question_ids=["Q-17"]))
+    found = {(lk.marker, lk.path) for lk in leaks}
+    assert ("Q-17", "$.messages[0].content") in found
+    assert ("expected_skill", "$.meta.expected_skill#key") in found
+    # A hyphenated path still names the tool: `dokimi-run-` is a leak.
+    assert ("dokimi", "$.cwd") in found
+
+
+def test_marker_inside_a_longer_word_is_not_a_leak():
+    assert audit_payload({"t": "evaluate the baseline_x reevaluation"}, ["eval"]) == []
+    assert [lk.marker for lk in audit_payload({"t": "see eval_queries.json"}, ["eval"])] == ["eval"]
+
+
+def test_markers_match_any_case_only_between_non_alphanumerics_and_inside_tuples():
+    assert [lk.marker for lk in audit_payload({"t": "Run by DOKIMI"}, ["dokimi"])] == ["dokimi"]
+    assert audit_payload({"t": "xdokimi and dokimix"}, ["dokimi"]) == []
+    assert [(lk.path, lk.marker) for lk in audit_payload({"t": ("ok", "the judge")}, ["judge"])] == [
+        ("$.t[1]", "judge")]
+
+
+def test_allowed_texts_blank_longest_first():
+    # Shortest first would blank "skill" inside the longer allowance, leaving "eval" visible.
+    assert audit_payload({"t": "eval the skill"}, ["eval"], allowed=["skill", "eval the skill"]) == []
+
+
+def test_operator_content_is_blanked_but_dokimi_text_around_it_is_not():
+    desc = "Compare against a baseline run before tuning."
+    payload = {"tools": [{"description": f"Load skill. {desc}"}],
+               "system": "You are in the treatment arm."}
+    leaks = audit_payload(payload, allowed=[desc])
+    assert [(lk.marker, lk.path) for lk in leaks] == [("treatment", "$.system")]
+    # Without the allowance the skill's own word is reported too.
+    assert {lk.marker for lk in audit_payload(payload)} == {"baseline", "treatment"}
+
+
+def test_new_leaks_ignore_what_the_stock_harness_already_says():
+    reference = {"system": "Test the golden path and judge edge cases."}
+    measured = {"system": "Test the golden path and judge edge cases.", "user": "graded by the judge"}
+    assert [(lk.path, lk.marker) for lk in new_leaks(measured, reference)] == [("$.user", "judge")]
+
+
+def test_skill_catalog_request_carries_no_dokimi_artifacts(tmp_path, monkeypatch):
+    # The request the skill-catalog arm actually sends (messages + tool schemas), captured
+    # at the LLM chokepoint, holds the query and the skills' own text and nothing else of ours.
+    paths = [_skill(tmp_path / "cat", "sum-fast", "Use when summing arrays; beats the baseline loop.",
+                    "# Sum\nUse a reduction."),
+             _skill(tmp_path / "cat", "io-read", "Use when reading files.", "# IO\nRead in chunks.")]
+    skills = [load_skill(p) for p in paths]
+    arm = SkillCatalogTreatment(skills)
+    question = "How do I sum a big array quickly?"
+    cfg = arm.prepare(question, "lib")
+    sent = []
+
+    class _Msg:
+        content, tool_calls = "Use a reduction.", None
+
+    class _Resp:
+        choices = [type("C", (), {"message": _Msg()})()]
+        usage = type("U", (), {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2})()
+
+    def fake(messages, model, tools=None, **kw):
+        sent.append({"messages": [dict(m) for m in messages], "tools": tools})
+        return _Resp()
+
+    monkeypatch.setattr(ar_mod, "chat_completion", fake)
+    run_agent_loop(question, cfg.tools, model="m", system_prompt=cfg.system_prompt)
+    allowed = [question] + [s.description for s in skills] + [s.name for s in skills]
+    markers = markers_for(question_ids=["q-001"], arm_names=[arm.name], paths=[str(tmp_path)])
+    assert sent and audit_payload(sent, markers, allowed) == []
+    # The audit is live on this payload: the arm's own name, planted, is caught.
+    sent[0]["messages"].append({"role": "user", "content": f"arm {arm.name}"})
+    assert {lk.marker for lk in audit_payload(sent, markers, allowed)} == {"skill_catalog", arm.name}
+
+
+def _task(tmp_path: Path) -> Path:
+    task = tmp_path / "content" / "tasks" / "demo-task"
+    (task / "tests").mkdir(parents=True)
+    (task / "solution").mkdir()
+    (task / "environment").mkdir()
+    (task / "instruction.md").write_text("Make sum() faster.", encoding="utf-8")
+    (task / "task.toml").write_text("[environment]\n", encoding="utf-8")
+    return task
+
+
+def test_docker_claude_discover_prompt_is_task_framing_only(tmp_path):
+    task = load_task(_task(tmp_path))
+    treatment = Treatment(name="sum-fast", plugin="sum-fast")
+    out = tmp_path / "out"
+    out.mkdir()
+    prompts = {}
+    for notice in (True, False):
+        h = DockerSolverHarness(harness_id="docker-claude", mode="claude", skill_notice=notice)
+        argv = h._claude_args(task, treatment, "/plugins/sum-fast", "haiku", out)
+        prompt = argv[argv.index("-p") + 1]
+        prompts[notice] = added_framing(prompt, task.instruction)
+        assert audit_payload(prompt, markers_for(paths=[str(task.path)])) == []
+    # Discover keeps dokimi's task framing but drops the notice that names the skill.
+    assert "sum-fast" not in prompts[False] and prompts[False].startswith("Solve this coding task.")
+    assert prompts[True].startswith(TREATMENT_NOTICE.format(skill_id=treatment.skill_id).strip())
diff --git a/dokimi/tests/test_trigger_probe.py b/dokimi/tests/test_trigger_probe.py
new file mode 100644
index 000000000..041789890
--- /dev/null
+++ b/dokimi/tests/test_trigger_probe.py
@@ -0,0 +1,383 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Real-harness trigger probe: staging, argv/env hygiene, transcript parsing, scoring."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from dokimi import process
+from dokimi.cli import build_parser
+from dokimi.eval.arm_summary import summarize_skill_selection
+from dokimi.external.trigger_set import TriggerQuery
+from dokimi.harnesses import trigger_probe as tp
+from dokimi.harnesses.agent_sandbox import solver_cli_flags
+from dokimi.harnesses.payload_audit import audit_payload, markers_for
+
+
+def _catalog(tmp_path: Path) -> Path:
+    for name in ("sum-fast", "io-read"):
+        d = tmp_path / "cat" / name
+        d.mkdir(parents=True)
+        (d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: Use for {name}.\n---\nbody\n",
+                                    encoding="utf-8")
+    return tmp_path / "cat"
+
+
+def _stream(*skills_per_turn, other: str = "", version: str = "2.1.304") -> str:
+    lines = [{"type": "system", "subtype": "init", "skills": ["sum-fast", "io-read", "dataviz"],
+              "claude_code_version": version}]
+    if other:
+        lines.append({"type": "assistant", "message": {"content": [
+            {"type": "tool_use", "name": other, "input": {}}]}})
+    for turn in skills_per_turn:
+        lines.append({"type": "assistant", "message": {"content": [
+            {"type": "text", "text": "ok"}] + [
+            {"type": "tool_use", "name": "Skill", "input": {"skill": s}} for s in turn]}})
+    lines.append({"type": "result", "subtype": "success", "is_error": False,
+                  "total_cost_usd": 0.01, "num_turns": 2})
+    return "\n".join(json.dumps(x) for x in lines) + "\n"
+
+
+def test_parse_stream_takes_the_first_skill_turn_as_the_choice():
+    s = tp.parse_stream(_stream(["plug:sum-fast", "io-read"], ["io-read"], other="Bash"))
+    assert s["first_turn_loaded"] == ["sum-fast", "io-read"]
+    assert s["selected"] == "sum-fast"
+    assert s["loaded"] == ["sum-fast", "io-read", "io-read"]
+    assert s["tools_before_choice"] == ["Bash"]
+    assert s["listed"] == ["sum-fast", "io-read", "dataviz"] and s["claude_code_version"] == "2.1.304"
+    assert tp.parse_stream(_stream())["selected"] is None
+
+
+def test_tools_after_the_choice_are_not_counted_before_it():
+    events = [{"type": "assistant", "message": {"id": "m1", "content": [
+                  {"type": "tool_use", "name": "Skill", "input": {"skill": "sum-fast"}}]}},
+              {"type": "assistant", "message": {"id": "m2", "content": [
+                  {"type": "tool_use", "name": "Bash", "input": {}}]}}]
+    s = tp.parse_stream("\n".join(json.dumps(e) for e in events))
+    assert s["first_turn_loaded"] == ["sum-fast"] and s["tools_before_choice"] == []
+
+
+def test_parse_stream_groups_per_block_events_by_message_id():
+    # Claude Code streams each content block as its own event; one turn shares message.id.
+    def block(mid, block, parent=None):
+        return {"type": "assistant", "parent_tool_use_id": parent, "message": {"id": mid, "content": [block]}}
+
+    def skill(s):
+        return {"type": "tool_use", "name": "Skill", "input": {"skill": s}}
+
+    events = [block("m1", {"type": "thinking", "thinking": "..."}), block("m1", skill("sum-fast")),
+              block("m1", skill("io-read")), {"type": "user", "message": {"content": []}},
+              block("m2", skill("dataviz"), parent="toolu_1"), block("m3", skill("sum-fast")),
+              {"type": "result", "subtype": "success", "is_error": False}]
+    s = tp.parse_stream("\n".join(json.dumps(e) for e in events))
+    assert s["first_turn_loaded"] == ["sum-fast", "io-read"]
+    assert s["loaded"] == ["sum-fast", "io-read", "sum-fast"]
+    rec = {"question_id": "a", "expected_skill": "sum-fast",
+           "arms": {"cc": {"answer": None, "error": None, "tool_usage": {"skill": s}}}}
+    summary = summarize_skill_selection([rec])["cc"]
+    assert summary["recall"] == 0.0 and summary["multi_load_rate"] == 1.0
+
+
+def test_argv_is_the_query_verbatim_and_pins_only_what_is_asked():
+    cfg = tp.ProbeConfig(model="haiku", max_turns=2)
+    argv = tp.probe_argv("How do I sum?", cfg)
+    assert argv[argv.index("-p") + 1] == "How do I sum?"
+    assert "--settings" not in argv and "--tools" not in argv and "--plugin-dir" not in argv
+    pinned = tp.probe_argv("q", tp.ProbeConfig(listing_budget_fraction=0.03))
+    assert json.loads(pinned[pinned.index("--settings") + 1]) == {"skillListingBudgetFraction": 0.03}
+
+
+def test_sealed_argv_switches_off_the_builtin_plugins_as_the_arms_do():
+    seal = json.loads(solver_cli_flags()[solver_cli_flags().index("--settings") + 1])
+    for frac, extra in ((None, {}), (0.05, {"skillListingBudgetFraction": 0.05})):
+        argv = tp.probe_argv("q", tp.ProbeConfig(stock=False, listing_budget_fraction=frac))
+        assert argv.count("--settings") == 1
+        assert json.loads(argv[argv.index("--settings") + 1]) == {**seal, **extra}
+
+
+def test_env_keeps_operator_config_out_and_bundled_skills_on_in_stock_mode(tmp_path):
+    ws = tp.new_workspace()
+    try:
+        source = {"PATH": "/bin", "AWS_REGION": "us-east-2", "CLAUDECODE": "1",
+                  "CLAUDE_CONFIG_DIR": "/home/op/.claude", "GITHUB_TOKEN": "x"}
+        stock = tp.probe_env(ws, tp.ProbeConfig(), source)
+        sealed = tp.probe_env(ws, tp.ProbeConfig(stock=False), source)
+        assert stock["CLAUDE_CONFIG_DIR"] == str(ws.root / "config") and stock["HOME"] == str(ws.root / "home")
+        assert "GITHUB_TOKEN" not in stock and "CLAUDECODE" not in stock
+        assert "CLAUDE_CODE_DISABLE_BUNDLED_SKILLS" not in stock
+        assert sealed["CLAUDE_CODE_DISABLE_BUNDLED_SKILLS"] == "1"
+    finally:
+        ws.cleanup()
+
+
+def test_run_query_stages_a_clean_repo_whose_path_names_no_experiment(tmp_path):
+    seen = {}
+
+    def handler(call: process.Call):
+        if call.argv[0] == "git":
+            return None if call.argv[1] != "status" else process.reply("")
+        seen["cwd"] = call.cwd
+        seen["skills"] = sorted(p.name for p in (call.cwd / ".claude" / "skills").iterdir())
+        seen["env"] = call.env
+        return process.reply(_stream(["io-read"]))
+
+    fake = process.FakeRunner(handler)
+    # Real git for staging (it is the clean-repo claim under test), fake claude.
+    def runner(argv, **kw):
+        return process.run(argv, **kw) if argv[0] == "git" else fake(argv, **kw)
+
+    r = tp.run_query("read a csv", _catalog(tmp_path), tp.ProbeConfig(), runner=runner,
+                     source_env={"PATH": "/usr/bin:/bin"})
+    assert r["skill"]["first_turn_loaded"] == ["io-read"] and r["error"] is None and r["cost_usd"] == 0.01
+    assert seen["skills"] == ["io-read", "sum-fast"]
+    assert audit_payload(str(seen["cwd"]), markers_for()) == []
+    assert not seen["cwd"].exists()  # cleaned up
+
+
+def test_out_of_turns_without_a_skill_is_no_trigger_not_a_failure(tmp_path):
+    stream = _stream().replace('"subtype": "success", "is_error": false',
+                               '"subtype": "error_max_turns", "is_error": true')
+
+    def runner(argv, **kw):
+        return process.run(argv, **kw) if argv[0] == "git" else process.reply(stream)
+
+    r = tp.run_query("q", _catalog(tmp_path), tp.ProbeConfig(), runner=runner, source_env={"PATH": "/usr/bin:/bin"})
+    assert r["error"] is None and r["skill"]["first_turn_loaded"] == []
+    broken = tp.run_query("q", _catalog(tmp_path / "b"), tp.ProbeConfig(),
+                          runner=lambda a, **kw: process.run(a, **kw) if a[0] == "git" else process.reply("", returncode=1),
+                          source_env={"PATH": "/usr/bin:/bin"})
+    assert broken["error"] == "no result event (exit 1)"
+
+
+def test_a_hang_after_the_choice_is_scored_not_dropped(tmp_path):
+    # A stock-tool Bash call can outlive the turn; the CLI is then killed at --timeout.
+    with_result = _stream(["sum-fast"])
+    no_result = with_result.rsplit('{"type": "result"', 1)[0]
+    cases = {"after-result": (with_result, None, ["sum-fast"]), "no-result": (no_result, None, ["sum-fast"]),
+             "no-skill-after-result": (_stream(), None, []), "nothing": ("", "timeout", [])}
+    for name, (stream, error, loaded) in cases.items():
+        def runner(argv, stream=stream, **kw):
+            return process.run(argv, **kw) if argv[0] == "git" else process.reply(stream, timed_out=True)
+
+        r = tp.run_query("q", _catalog(tmp_path / name), tp.ProbeConfig(), runner=runner,
+                         source_env={"PATH": "/usr/bin:/bin"})
+        assert r["error"] == error, name
+        assert r["skill"]["first_turn_loaded"] == loaded, name
+
+
+def test_records_score_with_the_same_summary_as_the_emulated_arm(tmp_path):
+    answers = {"sum it": ["sum-fast"], "read it": ["sum-fast"], "hello": []}
+
+    def runner(argv, **kw):
+        if argv[0] == "git":
+            return process.run(argv, **kw)
+        q = argv[list(argv).index("-p") + 1]
+        return process.reply(_stream(answers[q]) if answers[q] else _stream())
+
+    queries = [TriggerQuery("a", "sum it", "sum-fast"), TriggerQuery("b", "read it", "io-read"),
+               TriggerQuery("c", "hello", None), TriggerQuery("d", "unlabelled", None, labelled=False)]
+    recs = tp.probe_trigger_set(queries, _catalog(tmp_path), tp.ProbeConfig(), trials=2, runner=runner)
+    assert [r["question_id"] for r in recs] == ["a", "b", "c"]
+    s = summarize_skill_selection(recs)
+    assert set(s) == {"claude_code#1", "claude_code#2"}
+    assert s["claude_code#1"]["recall"] == 0.5 and s["claude_code#1"]["misroute_rate"] == 0.5
+    assert s["claude_code#1"]["false_trigger_rate"] == 0.0
+
+
+def test_one_failed_run_is_that_trials_error_not_the_lost_set(tmp_path):
+    def runner(argv, **kw):
+        if argv[0] == "git":
+            return process.run(argv, **kw)
+        if argv[list(argv).index("-p") + 1] == "read it":
+            raise FileNotFoundError(2, "No such file or directory", argv[0])
+        return process.reply(_stream(["sum-fast"]))
+
+    queries = [TriggerQuery("a", "sum it", "sum-fast"), TriggerQuery("b", "read it", "io-read")]
+    recs = tp.probe_trigger_set(queries, _catalog(tmp_path), tp.ProbeConfig(), runner=runner)
+    assert [r["question_id"] for r in recs] == ["a", "b"]
+    assert recs[1]["arms"]["claude_code#1"]["error"].startswith("FileNotFoundError")
+    s = summarize_skill_selection(recs)["claude_code#1"]
+    assert s["recall"] == 1.0 and s["n_failed"] == 1
+
+
+@pytest.mark.skipif(os.name == "nt", reason="a shell-script stand-in for claude")
+def test_cli_runs_a_relative_claude_from_the_scratch_cwd(tmp_path, monkeypatch, capsys):
+    (tmp_path / "stream.jsonl").write_text(_stream(["sum-fast"]), encoding="utf-8")
+    fake = tmp_path / "bin" / "claude"
+    fake.parent.mkdir()
+    fake.write_text("#!/bin/sh\nexec cat \"$STREAM\"\n".replace("$STREAM", str(tmp_path / "stream.jsonl")),
+                    encoding="utf-8")
+    fake.chmod(0o755)
+    (tmp_path / "ts.json").write_text(json.dumps({"questions": [
+        {"id": "a", "question": "sum it", "expected_skill": "sum-fast"}]}), encoding="utf-8")
+    _catalog(tmp_path)
+    monkeypatch.chdir(tmp_path)
+    args = build_parser().parse_args(["catalog", "trigger-probe", "--catalog", "cat", "--trigger-set",
+                                      "ts.json", "--claude", "bin/claude", "--out", "probe.json"])
+    args.func(args)
+    out = json.loads((tmp_path / "probe.json").read_text(encoding="utf-8"))
+    arm = out["answers"][0]["arms"]["claude_code#1"]
+    assert arm["error"] is None and arm["tool_usage"]["skill"]["first_turn_loaded"] == ["sum-fast"]
+
+
+def test_run_query_passes_argv_env_timeout_and_empty_stdin(tmp_path):
+    seen = {}
+
+    def runner(argv, **kw):
+        if argv[0] == "git":
+            return process.run(argv, **kw)
+        seen.update(kw, argv=list(argv))
+        return process.reply(_stream(["sum-fast"]))
+
+    cfg = tp.ProbeConfig(model="haiku", max_turns=7, timeout_sec=12.5, stock=False,
+                         listing_budget_fraction=0.02, claude="/opt/claude")
+    tp.run_query("sum it", _catalog(tmp_path), cfg, runner=runner, source_env={"PATH": "/bin"})
+    argv = seen["argv"]
+    assert argv[:3] == ["/opt/claude", "-p", "sum it"]
+    for flag, value in (("--output-format", "stream-json"), ("--max-turns", "7"), ("--model", "haiku")):
+        assert argv[argv.index(flag) + 1] == value
+    assert "--verbose" in argv and "--no-session-persistence" in argv
+    assert json.loads(argv[argv.index("--settings") + 1])["skillListingBudgetFraction"] == 0.02
+    env = seen["env"].resolve()
+    root = Path(env["HOME"]).parent
+    assert env["TMPDIR"] == env["HOME"] == str(root / "home") and env["CLAUDE_CONFIG_DIR"] == str(root / "config")
+    assert env["CLAUDE_CODE_DISABLE_BUNDLED_SKILLS"] == "1"
+    assert seen["timeout"] == 12.5 and seen["input"] == "" and seen["cwd"] == root / "work"
+
+
+def test_staging_copies_symlinked_skills_as_files(tmp_path):
+    cat = _catalog(tmp_path)
+    (tmp_path / "ext").mkdir()
+    (tmp_path / "ext" / "ref.md").write_text("ref", encoding="utf-8")
+    (cat / "io-read" / "ref.md").symlink_to(tmp_path / "ext" / "ref.md")
+    ws = tp.new_workspace()
+    try:
+        copied = tp.stage_workspace(cat, ws) / ".claude" / "skills" / "io-read" / "ref.md"
+        assert not copied.is_symlink() and copied.read_text(encoding="utf-8") == "ref"
+    finally:
+        ws.cleanup()
+
+
+@pytest.mark.skipif(not Path("/proc/self/cwd").exists(), reason="reads /proc")
+def test_kill_strays_ends_a_backgrounded_process_left_in_the_workspace(tmp_path):
+    import subprocess
+
+    (tmp_path / "work").mkdir()
+    # Its own session, as the Bash tool's background job is: a group kill does not reach it.
+    proc = subprocess.Popen(["sleep", "60"], cwd=tmp_path / "work", start_new_session=True)
+    try:
+        assert tp.kill_strays(tmp_path) == [proc.pid]
+        assert proc.wait(timeout=5) == -9
+    finally:
+        proc.kill()
+
+
+@pytest.mark.skipif(not Path("/proc/self/cwd").exists(), reason="reads /proc")
+def test_run_query_kills_what_the_run_left_in_its_workspace_only(tmp_path):
+    import shutil
+    import subprocess
+
+    procs = {}
+
+    def runner(argv, **kw):
+        if argv[0] == "git":
+            return process.run(argv, **kw)
+        sibling = Path(str(kw["cwd"].parent) + "d")  # a concurrent run's root, same prefix
+        sibling.mkdir()
+        procs["sibling_dir"] = sibling
+        # Own sessions, as the Bash tool's background jobs: the timeout's group kill misses them.
+        for name, cwd in (("left", kw["cwd"]), ("sibling", sibling)):
+            procs[name] = subprocess.Popen(["sleep", "60"], cwd=cwd, start_new_session=True)
+        return process.reply(_stream(["sum-fast"]))
+
+    try:
+        r = tp.run_query("q", _catalog(tmp_path), tp.ProbeConfig(), runner=runner,
+                         source_env={"PATH": "/usr/bin:/bin"})
+        assert r["error"] is None
+        assert procs["left"].wait(timeout=5) == -9
+        assert procs["sibling"].poll() is None
+    finally:
+        for name in ("left", "sibling"):
+            if name in procs:
+                procs[name].kill()
+                procs[name].wait()
+        if "sibling_dir" in procs:
+            shutil.rmtree(procs["sibling_dir"], ignore_errors=True)
+
+
+def _cli(tmp_path, monkeypatch, *extra, init_skills=("sum-fast", "io-read"), expected="sum-fast"):
+    from dokimi.commands import catalog_probe
+
+    seen = {}
+
+    def fake_probe(queries, catalog, cfg, *, trials=1, **kw):
+        seen.update(cfg=cfg, trials=trials, catalog=catalog)
+        seen.setdefault("queries", []).extend(q.id for q in queries)
+        skill = tp.parse_stream(_stream(["sum-fast"]).replace(
+            '["sum-fast", "io-read", "dataviz"]', json.dumps(list(init_skills))))
+        return [{"question_id": q.id, "question_text": q.text, "expected_skill": q.expected_skill,
+                 "arms": {"claude_code#1": {"answer": None, "error": None, "tool_usage": {"skill": skill},
+                                            "metrics": {"cost_usd": 0.01, "num_turns": 3,
+                                                        "result_subtype": "error_max_turns"}}}} for q in queries]
+
+    class Pool(catalog_probe.ThreadPoolExecutor):
+        def __init__(self, n):
+            seen["concurrency"] = n
+            super().__init__(n)
+
+    monkeypatch.setattr(tp, "probe_trigger_set", fake_probe)
+    monkeypatch.setattr(catalog_probe, "ThreadPoolExecutor", Pool)
+    (tmp_path / "ts.json").write_text(json.dumps({"questions": [
+        {"id": q, "question": f"q {q}", "expected_skill": expected} for q in ("a", "b", "c")]}),
+        encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+    if not (tmp_path / "cat").exists():
+        _catalog(tmp_path)
+    args = build_parser().parse_args(["catalog", "trigger-probe", "--catalog", "cat", "--trigger-set",
+                                      "ts.json", "--claude", "sh", *extra])
+    args.func(args)
+    return seen
+
+
+def test_cli_passes_every_flag_to_the_probe(tmp_path, monkeypatch):
+    seen = _cli(tmp_path, monkeypatch, "--model", "haiku", "--limit", "2", "--trials", "3",
+                "--max-turns", "6", "--timeout", "9", "--listing-budget-fraction", "0.04",
+                "--concurrency", "5", "--sealed", "--out", "probe.json")
+    cfg = seen["cfg"]
+    assert (cfg.model, cfg.max_turns, cfg.timeout_sec, cfg.stock, cfg.listing_budget_fraction) == (
+        "haiku", 6, 9.0, False, 0.04)
+    assert os.path.isabs(cfg.claude) and seen["trials"] == 3 and seen["concurrency"] == 5
+    assert sorted(seen["queries"]) == ["a", "b"]
+    out = json.loads((tmp_path / "probe.json").read_text(encoding="utf-8"))
+    assert not (tmp_path / "probe.json.partial").exists()
+    assert out["mode"] == "sealed" and out["max_turns_hit"] == 2 and out["not_listed"] == []
+
+
+def test_cli_refuses_a_bad_out_or_target_before_any_run(tmp_path, monkeypatch):
+    from dokimi.utils import InputError
+
+    with pytest.raises(InputError, match="not a writable file path"):
+        _cli(tmp_path, monkeypatch, "--out", "nodir/probe.json")
+    with pytest.raises(InputError, match="not a skill in --catalog"):
+        _cli(tmp_path, monkeypatch, "--target", "sum-fsat", "--out", "probe.json")
+
+
+def test_cli_refuses_a_label_that_names_no_catalog_skill(tmp_path, monkeypatch):
+    from dokimi.utils import InputError
+
+    with pytest.raises(InputError, match="names no skill in --catalog: sum-fsat"):
+        _cli(tmp_path, monkeypatch, "--out", "probe.json", expected="sum-fsat")
+    assert not (tmp_path / "probe.json.partial").exists()
+    _cli(tmp_path, monkeypatch, "--out", "probe.json", expected=["sum-fast", "io-read"])
+
+
+def test_cli_fails_when_claude_code_did_not_list_a_catalog_skill(tmp_path, monkeypatch):
+    with pytest.raises(SystemExit, match="did not list: io-read"):
+        _cli(tmp_path, monkeypatch, "--out", "probe.json", init_skills=("sum-fast", "dataviz"))
+    assert json.loads((tmp_path / "probe.json").read_text(encoding="utf-8"))["not_listed"] == ["io-read"]