# DETECT orchestrator Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use /ship (recommended) or /executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build `orchestrator/detect.py` — the production scan harness that selects `baseline` + matched specialist detectors per attack-surface target, dispatches `gate.py` k=1 per (target, detector), unions + semantic-merges findings, and (over the corpus, `--bench`) finally measures k=1 production recall — plus the one additive `gate.py --emit` it depends on.

**Architecture:** `detect.py` reuses, does not rewrite: `mapper` (enumerate/prioritize/coverage_map/kind_of), `gate.py` (subprocessed per detector, now emitting the `prevent/contract.py` finding shape via additive `--emit`), `semantic_merge.merge_groups` (one merge engine, reused cross-detector), `prevent/registry.load` (detector discovery), `bench` (cell loading + `is_flagged`). All modules path-loaded (`importlib.util.spec_from_file_location`, NO `__init__.py`), stdlib-only, no new deps. no-false-clean / no-false-coverage is load-bearing throughout: a target/cell is "clean" ONLY when every selected detector ran ok with zero findings; any degradation → COVERAGE-INCOMPLETE, never a silent recall-0.

**Tech Stack:** Python 3 (stdlib only), `claude -p` via the existing `gate.py`/`llm_runner` legs, `bun` oracle inside `gate.py`. Tests: pytest (`rtk proxy python3 -m pytest`), deterministic-first; LLM-live is rate-only evidence, NEVER a ship gate.

**SoT:** `docs/specs/2026-06-18-detect-orchestrator-design.md`. Read it; do not re-derive its decisions.

---

## Pre-flight (read before Wave 1)

- **Worktree.** This plan writes production code → never on master. `/ship` creates `.worktrees/<branch>/`; all edits happen there. Pass `WORKTREE_PATH` to every implementer.
- **Vocab.** Testing-native ONLY: detector / domain / corpus / cell / canary / band / oracle / roll / recall. The noun `adapter` is licensed. NEVER module/layer/tier/seam/port-adapter/plugin.
- **Path-load idiom** (copy verbatim from `mapper.py:13–17`):
  ```python
  def _load(name, path):
      spec = importlib.util.spec_from_file_location(name, path)
      mod = importlib.util.module_from_spec(spec)
      spec.loader.exec_module(mod)
      return mod
  ```
- **Verified facts (already read — do not re-assume):**
  - `gate.py`: `groups` computed at 284–287, `oracle_results` at 267, report built 290–322. `oracle_status(out) -> (silent, unreliable, label)`. `--detector <id>` loads `domains/security/detectors/<id>/detector.json` and requires `kind=="llm"` (236–238). Refactors 1 & 2 already landed (`from resolver import …`, `from llm_runner import run_llm`).
  - `prevent/contract.py`: `finding(rule_id, level, cls, message, file, line=0, symbol="")` → `{ruleId, level, class, message, file, line, symbol}`; raises `ValueError` if `level not in ("error","warning","note")`. `emit(...)` writes to STDOUT — DETECT/gate must NOT call it; build the dict + `json.dump` to a file.
  - `prevent/registry.py`: `load(domains_root) -> (detectors, skipped)`, globs `<domains_root>/domains/*/detectors/*/detector.json`; malformed → `skipped=[(path, reason)]`; each manifest gets `m["_dir"]`.
  - `orchestrator/semantic_merge.py`: `merge_groups(groups, config_dir=None, model="sonnet", effort="medium") -> (merged, degraded)`; groups are `{"title","sev","rolls":set}`; degraded=True returns input unchanged.
  - `orchestrator/mapper.py`: `_load`, `Entry(path, kind, why)` (frozen dataclass; `path` absolute), `enumerate_surface(root)->[Entry]`, `prioritize(entries)`, `actual_surface(root)->[(abspath,kind)]`, `coverage_map(enumerated, actual, scanned)` (scanned={path:report}), `_file_routed_kind(rel)`, `KIND_SIGNALS` (dict kind→(regex, reliable)), `FILE_ROUTE_RULES`. `dispatch(entry,out_dir,gate_args)` shells `gate.py`.
  - `bench.py`: `load_cells()` (each cell dict has `id`, `class`, `canonical_symbol`, `_dir`, usually `file`), `is_flagged(canonical, findings)` lower-cases each finding STRING and substring-matches `canonical_symbol`/`class`.
  - Corpus cell layout: `domains/security/corpus/<id>/{vuln.ts, safe.ts, canonical.json}`. Detector manifests: baseline/auth/access-control/finance are `kind:"llm"` with `applies_to`; oracle/deps/headers have NO `kind:"llm"` (correctly skipped by the router).
- **Commit allow-list (this project):** `orchestrator/*.py`, `tests/test_*.py`, `tests/golden/*`, `docs/`. tests/ ARE committed here. Commit messages: terse caveman, NEVER co-author.
- **Run tests with:** `rtk proxy python3 -m pytest`.

---

## Wave Plan

| Wave | Tasks | Files touched | Safe to parallelize? |
|------|-------|---------------|----------------------|
| 1 | Task 1, Task 2 | `orchestrator/gate.py`+`tests/test_gate_emit.py`+`tests/golden/`; `orchestrator/mapper.py`+`tests/test_mapper_kind_of.py` | ✅ disjoint files |
| 2 | Task 3 | `orchestrator/detect.py`, `tests/test_detect_select.py` | single (new file) |
| 3 | Task 4 | `orchestrator/detect.py`, `tests/test_detect_dispatch.py` | single (same file as W2) |
| 4 | Task 5 | `orchestrator/detect.py`, `tests/test_detect_target.py` | single (same file) |
| 5 | Task 6 | `orchestrator/detect.py`, `tests/test_detect_run.py` | single (same file) |
| 6 | Task 7 | `orchestrator/detect.py`, `tests/test_detect_bench.py` | single (same file) |
| 7 | Task 8 | `orchestrator/detect.py`, `tests/test_detect_cli.py` | single (same file) |
| 8 | Task 9 | `docs/validation/2026-06-18-detect-k1-recall-smoke.md`, `tests/test_detect_llm_smoke.py` | single |

Waves 2–7 all extend `detect.py` and are strictly sequential (each function uses the previous). Wave 1's two tasks touch disjoint files → parallel-safe. **Pilot-before-fanout:** Wave 1 lands and the full suite stays green BEFORE Wave 2 starts.

---

## File Structure

- `orchestrator/gate.py` (modify) — extract `build_report(...)`, add `build_emit_dict(...)` + `_load_contract()` + `LEVEL` map + `--emit` flag. Behavior-preserving for the no-emit path.
- `orchestrator/mapper.py` (modify) — add additive `kind_of(rel, body=None)` (factors the per-file inference `enumerate_surface` already does; existing functions untouched).
- `orchestrator/detect.py` (create) — the orchestrator: `_load` + module handles, `select_detectors`, `dispatch_one`/`_degraded`, `_emit_to_groups`/`detect_target`/`_write_target_report`, `run`, `run_bench`/`_cell_kind`, `main`.
- `tests/test_gate_emit.py`, `tests/golden/gate_report_single_file.md`, `tests/test_mapper_kind_of.py`, `tests/test_detect_*.py` (create).

---

## NOT covered / deferred (own follow-up specs — do NOT build here)

- **Opus-judgment triage WIDENING** (the router only WIDENS deterministically; an LLM widening pass is additive later).
- **Repo-level deterministic detectors** (deps/headers S11 once-per-repo) — not part of per-target DETECT.
- **Per-subprocess oracle de-duplication** — `gate.py` re-runs the oracle inside every detector subprocess on the same target (correctness-fine; cross-detector merge dedups the duplicate oracle findings). Perf-only optimization, deferred.
- **Per-finding `line`/`symbol` precision** — MVP uses file + message.
- **`mapper` rewrite into the orchestrator** — rejected for MVP; mapper's enumerate half is reused as-is.
- **`--map` as an INPUT cache** — MVP `--map` is the coverage-map OUTPUT path (matches `mapper.main`). Reuse-as-input is deferred: the 3-bucket coverage map drops per-entry `kind`, so it cannot rehydrate enumeration; a richer enumerate-cache is a separate follow-up.
- **Fast-follow (REQUIRED before any 99% comparison):** re-baseline the hand-judged ~86% floor THROUGH `run_bench`, and literal-ize the phrase `canonical_symbol`s (S5/S6/S7/S10) so the whole corpus is autoscorable. Tracked in the spec's Open Questions.

---

## Task 1: `gate.py --emit` (contract-unification) + report extraction

**Wave:** 1
**Blocks:** Task 4 (dispatch_one parses this emit)
**Blocked by:** —

**Files:**
- Modify: `orchestrator/gate.py` (add `LEVEL`, `_load_contract`, `build_report`, `build_emit_dict`; rewire `main()`; add `--emit`)
- Create: `tests/test_gate_emit.py`
- Create: `tests/golden/gate_report_single_file.md`

- [ ] **Step 1: Write the failing test (emit-dict shape + report golden)**

Create `tests/test_gate_emit.py`:

```python
"""test_gate_emit.py — deterministic: gate.py --emit contract serialization + report byte-identity.
No LLM: build_emit_dict / build_report are pure over synthetic groups/oracle. Path-load per convention."""
import importlib.util, os

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GOLDEN = os.path.join(ROOT, "tests", "golden", "gate_report_single_file.md")


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


gate = _load("sg_gate", os.path.join(ROOT, "orchestrator", "gate.py"))

# synthetic, deterministic inputs (no LLM)
GROUPS = [
    {"title": "reset token single-use not enforced", "sev": "high", "rolls": {0}},
    {"title": "missing rate limit on reset", "sev": "low", "rolls": {0}},
]
ORACLE_SILENT = [("/abs/target.ts", "(oracle silent)")]
TARGET = "/abs/target.ts"


def test_emit_dict_is_contract_shape_with_additive_keys():
    d = gate.build_emit_dict(TARGET, "baseline", ["S1", "S2"], 1, GROUPS, ORACLE_SILENT, {}, False)
    assert d["detector"] == "baseline"
    assert d["status"] == "ok"                       # no unresolved → ok
    assert set(d["coverage"]) == {"scanned", "unresolved"}
    f0 = d["findings"][0]
    # frozen contract keys present
    for kpresent in ("ruleId", "level", "class", "message", "file", "line", "symbol"):
        assert kpresent in f0
    assert f0["level"] == "error"                    # high → error
    assert f0["class"] == "S1,S2"                    # covers joined
    assert f0["message"] == "reset token single-use not enforced"
    # additive keys
    assert f0["sev"] == "high" and f0["rolls"] == 1 and f0["of"] == 1
    assert d["findings"][1]["level"] == "note"       # low → note


def test_emit_dict_degrades_on_dropped_dep():
    d = gate.build_emit_dict(TARGET, "baseline", ["S1"], 1, [], ORACLE_SILENT,
                             {"/abs/dep.ts": "over max-scope"}, False)
    assert d["status"] == "degraded"                 # non-empty unresolved → degraded (no-false-clean)
    assert any("dep.ts" in u for u in d["coverage"]["unresolved"])


def test_report_is_byte_identical_to_golden():
    rep = gate.build_report(TARGET, [], {}, ORACLE_SILENT, GROUPS, 1, "sonnet", "medium", False, False)
    assert rep == open(GOLDEN, encoding="utf-8").read()
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_gate_emit.py -q`
Expected: FAIL — `AttributeError: module 'sg_gate' has no attribute 'build_emit_dict'` (and golden file missing).

- [ ] **Step 3: Add `LEVEL`, `_load_contract`, `build_report`, `build_emit_dict` to `gate.py`**

Insert after the `SEV_RANK` definition (`gate.py:132`):

```python
LEVEL = {"critical": "error", "high": "error", "medium": "warning", "low": "note", "unrated": "note"}


def _load_contract():
    """Path-load prevent/contract.py LAZILY (only when --emit) so the no-emit path stays dep-free."""
    import importlib.util
    p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "prevent", "contract.py")
    spec = importlib.util.spec_from_file_location("sg_contract", p)
    m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m


def build_report(target, deps, dropped, oracle_results, groups, k, model, effort, no_merge, merge_degraded):
    """The gate's markdown report (extracted verbatim from main() so --emit cannot alter it; golden-locked)."""
    lines = ["# Security-gate report", "", "## Target reviewed", f"- `{os.path.relpath(target)}`"]
    if deps:
        lines.append("")
        lines.append("## Dependency bodies inlined as cross-file context (#36)")
        for dep_f, sym, importer in deps:
            lines.append(f"- `{os.path.relpath(dep_f)}` — import `{sym}` from {importer}")
    if dropped:
        lines.append("")
        lines.append("## Dropped — NOT inlined, NOT clean (raise --max-scope to cover)")
        for dep_f, reason in dropped.items():
            lines.append(f"- `{os.path.relpath(dep_f)}` — {reason}")
    lines.append("")
    lines.append("**Oracle (complete-mediation) — run per in-scope file (target + inlined deps):**")
    for f, out in oracle_results:
        silent, unreliable, status = oracle_status(out)
        role = "target" if f == target else "dep"
        lines.append(f"- `{os.path.relpath(f)}` ({role}): {status}")
        if not silent or unreliable:
            lines.append("")
            lines.append("```\n" + out + "\n```")
            lines.append("")
    merge_note = " (raw lexical union, --no-merge)" if no_merge else " + semantic-merge"
    scope_note = f", {len(deps)} dep bodies inlined" if deps else ", single-file"
    lines.append(f"**LLM leg (v2@{model}/{effort.upper()}, {k}-roll union{merge_note}{scope_note}):**")
    if merge_degraded:
        lines.append("- ⚠️ semantic-merge DEGRADED — findings shown un-deduped; "
                     "paraphrases of one bug may appear as separate low-roll entries (recall undersold).")
    for g in groups:
        m = g.get("members", 1)
        mtag = f", {m} paraphrases merged" if m > 1 else ""
        lines.append(f"- [{g['sev']}] {g['title']}  _(found by {len(g['rolls'])}/{k} rolls{mtag})_")
    lines.append("")
    return "\n".join(lines)


def build_emit_dict(target, detector_id, covers, k, groups, oracle_results, dropped, merge_degraded):
    """Serialize gate findings into the prevent/contract.py shape (a FILE, never contract.emit()'s stdout).
    Additive keys sev/rolls/of ride each finding (contract consumers ignore unknown keys). no-false-clean:
    dropped deps + merge-degraded + unreliable oracle ⇒ coverage.unresolved ⇒ status='degraded'."""
    contract = _load_contract()
    cls = ",".join(covers) if covers else "security"
    findings, unresolved = [], []
    for g in groups:
        f = contract.finding(rule_id=detector_id, level=LEVEL.get(g["sev"], "note"), cls=cls,
                             message=g["title"], file=os.path.relpath(target), line=0, symbol="")
        f["sev"] = g["sev"]; f["rolls"] = len(g["rolls"]); f["of"] = k
        findings.append(f)
    for dep_f, reason in dropped.items():
        unresolved.append(f"{os.path.relpath(dep_f)}: {reason}")
    if merge_degraded:
        unresolved.append("semantic-merge degraded")
    scanned = []
    for f, out in oracle_results:
        rel = os.path.relpath(f)
        if rel not in scanned:
            scanned.append(rel)
        silent, unreliable, label = oracle_status(out)
        if unreliable:
            unresolved.append(f"{rel}: oracle unreliable (unresolved predicate imports)")
        if not silent:
            of = contract.finding(rule_id="oracle", level="error", cls="S9",
                                  message=f"oracle FLAGS in {rel}: {label}", file=rel, line=0, symbol="")
            of["sev"] = "critical"; of["rolls"] = 1; of["of"] = k
            findings.append(of)
    status = "degraded" if unresolved else "ok"
    return {"detector": detector_id, "status": status, "findings": findings,
            "coverage": {"scanned": scanned, "unresolved": unresolved}}
```

- [ ] **Step 4: Rewire `main()` to use the extracted functions + add `--emit`**

In `gate.py:main()`: add the flag right after `--report` (`gate.py:230`):
```python
    ap.add_argument("--emit", default=None,
                    help="write the prevent/contract.py finding JSON here (additive; report path unchanged)")
```

Replace the `--detector` resolution block (`gate.py:233–238`) so it also captures id + covers:
```python
    detector_id, covers = "baseline", []
    if a.detector:  # modular detector selection: resolve <id>/detector.json -> its prompt (overrides --template)
        man_path = os.path.join(_root, "domains", "security", "detectors", a.detector, "detector.json")
        man = json.load(open(man_path, encoding="utf-8"))
        if man.get("kind") != "llm":
            ap.error(f"detector '{a.detector}' is kind={man.get('kind')!r}, not an llm prompt detector")
        a.template = os.path.join(os.path.dirname(man_path), man["prompt"])
        detector_id, covers = a.detector, man.get("covers", [])
```

Replace the inline report build (`gate.py:290–323`, from `# report` through `print(report)`) with:
```python
    report = build_report(target, deps, dropped, oracle_results, groups, a.k, a.model, a.effort,
                          a.no_merge, merge_degraded)
    print(report)
```
(Keep the existing `if a.report:` write block immediately after.)

Then, immediately after the `if a.report:` block, add:
```python
    if a.emit:
        with open(a.emit, "w", encoding="utf-8") as fh:
            json.dump(build_emit_dict(target, detector_id, covers, a.k, groups, oracle_results,
                                      dropped, merge_degraded), fh)
        print(f"\n[emit] {a.emit}", file=sys.stderr)
```

- [ ] **Step 5: Seed the golden from an INDEPENDENT copy of the pre-extraction report logic (anti-circular)**

Do NOT generate the golden from `build_report` — that would be `build_report(X) == build_report(X)`, a tautology that catches only non-determinism and certifies NOTHING about whether the extraction matched the original inline code. Since `tests/test_gate.py` does not test the report format, this golden is the ONLY guard on `build_report`; it must come from a copy of the report logic that is independent of the function under test. The `_ref` body below is the **verbatim pre-extraction inline block** (`gate.py:290–322` before Step 4 deleted it: `a.k→k`, `a.model→model`, `a.effort→effort`, `a.no_merge→no_merge`, group loop var renamed `grp` to avoid shadowing the module handle `g`). A typo introduced while extracting `build_report` in Step 3 now breaks `test_report_is_byte_identical_to_golden` — a real behavior-preservation lock.

```bash
cd <WORKTREE_PATH> && rtk proxy python3 - <<'PY'
import importlib.util, os
s = importlib.util.spec_from_file_location("g", "orchestrator/gate.py")
g = importlib.util.module_from_spec(s); s.loader.exec_module(g)  # for g.oracle_status only — NOT build_report

def _ref(target, deps, dropped, oracle_results, groups, k, model, effort, no_merge, merge_degraded):
    """Verbatim pre-extraction inline report block — independent reference, never calls build_report."""
    lines = ["# Security-gate report", "", "## Target reviewed", f"- `{os.path.relpath(target)}`"]
    if deps:
        lines.append("")
        lines.append("## Dependency bodies inlined as cross-file context (#36)")
        for dep_f, sym, importer in deps:
            lines.append(f"- `{os.path.relpath(dep_f)}` — import `{sym}` from {importer}")
    if dropped:
        lines.append("")
        lines.append("## Dropped — NOT inlined, NOT clean (raise --max-scope to cover)")
        for dep_f, reason in dropped.items():
            lines.append(f"- `{os.path.relpath(dep_f)}` — {reason}")
    lines.append("")
    lines.append("**Oracle (complete-mediation) — run per in-scope file (target + inlined deps):**")
    for f, out in oracle_results:
        silent, unreliable, status = g.oracle_status(out)
        role = "target" if f == target else "dep"
        lines.append(f"- `{os.path.relpath(f)}` ({role}): {status}")
        if not silent or unreliable:
            lines.append("")
            lines.append("```\n" + out + "\n```")
            lines.append("")
    merge_note = " (raw lexical union, --no-merge)" if no_merge else " + semantic-merge"
    scope_note = f", {len(deps)} dep bodies inlined" if deps else ", single-file"
    lines.append(f"**LLM leg (v2@{model}/{effort.upper()}, {k}-roll union{merge_note}{scope_note}):**")
    if merge_degraded:
        lines.append("- ⚠️ semantic-merge DEGRADED — findings shown un-deduped; "
                     "paraphrases of one bug may appear as separate low-roll entries (recall undersold).")
    for grp in groups:
        m = grp.get("members", 1)
        mtag = f", {m} paraphrases merged" if m > 1 else ""
        lines.append(f"- [{grp['sev']}] {grp['title']}  _(found by {len(grp['rolls'])}/{k} rolls{mtag})_")
    lines.append("")
    return "\n".join(lines)

# synthetic inputs — BYTE-IDENTICAL to tests/test_gate_emit.py (GROUPS/ORACLE_SILENT/TARGET)
GROUPS = [
    {"title": "reset token single-use not enforced", "sev": "high", "rolls": {0}},
    {"title": "missing rate limit on reset", "sev": "low", "rolls": {0}},
]
ORACLE_SILENT = [("/abs/target.ts", "(oracle silent)")]
TARGET = "/abs/target.ts"
os.makedirs("tests/golden", exist_ok=True)
open("tests/golden/gate_report_single_file.md", "w", encoding="utf-8").write(
    _ref(TARGET, [], {}, ORACLE_SILENT, GROUPS, 1, "sonnet", "medium", False, False))
print("golden written")
PY
```
Read `tests/golden/gate_report_single_file.md` and confirm it reads like the current `# Security-gate report` (Target reviewed → Oracle → LLM leg). `os.path.relpath` resolves against CWD, so this generation step AND the pytest run must both execute from `<WORKTREE_PATH>` (pytest already does). This file is the extraction-fidelity + additive-flag regression lock.

- [ ] **Step 6: Run the new test + the FULL existing suite**

Run: `rtk proxy python3 -m pytest tests/test_gate_emit.py -q`
Expected: PASS (3 tests).
Run: `rtk proxy python3 -m pytest -q`
Expected: PASS — the extraction is behavior-preserving; `tests/test_gate.py` (build_bundle) and all others stay green.

- [ ] **Step 7: Commit**

```bash
git add orchestrator/gate.py tests/test_gate_emit.py tests/golden/gate_report_single_file.md
git commit -m "gate: additive --emit contract shape + extract build_report (golden-locked)"
```

---

## Task 2: `mapper.kind_of(rel, body=None)` — single-file production kind-inference

**Wave:** 1
**Blocks:** Task 6 (run_bench routes blind via this)
**Blocked by:** —

**Files:**
- Modify: `orchestrator/mapper.py` (add `kind_of`)
- Create: `tests/test_mapper_kind_of.py`

- [ ] **Step 1: Write the failing test (fidelity vs enumerate_surface + body fallback)**

Create `tests/test_mapper_kind_of.py`:

```python
"""test_mapper_kind_of.py — kind_of() must agree with enumerate_surface() per file (no-drift), and infer a
call-registered kind from body content when not file-routed. Deterministic, no LLM. Path-load per convention."""
import importlib.util, os, tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


mapper = _load("sg_mapper", os.path.join(ROOT, "orchestrator", "mapper.py"))


def _route_tree():
    """A file-routed tree: a route file under a recognized route location + a non-route helper.
    NOTE: the route location must match a live FILE_ROUTE_RULES rule — http-file-route requires the
    CONSECUTIVE ("src","pages") segment pair (or app/**/route.ts). `src/routes/` is NOT a rule here."""
    d = tempfile.mkdtemp()
    routes = os.path.join(d, "apps/api/src/pages/auth")
    os.makedirs(routes, exist_ok=True)
    open(os.path.join(routes, "refresh.ts"), "w").write(
        "export async function POST(req){ return new Response('ok') }\n")
    helper = os.path.join(d, "packages/util/src")
    os.makedirs(helper, exist_ok=True)
    open(os.path.join(helper, "plain.ts"), "w").write("export const add = (a,b)=>a+b\n")
    return d


def test_kind_of_agrees_with_enumerate_surface_per_file():
    d = _route_tree()
    enr = {e.path: e.kind for e in mapper.enumerate_surface(d)}
    assert enr, "fixture must enumerate at least one entry"
    for path, kind in enr.items():
        rel = os.path.relpath(path, d)
        body = open(path, encoding="utf-8").read()
        assert mapper.kind_of(rel, body=body) == kind, f"kind_of drifted from enumerate on {rel}"


def test_kind_of_file_routed_needs_no_body():
    # a path under a route location resolves by LOCATION alone (body=None ok)
    assert mapper.kind_of("apps/api/src/pages/auth/refresh.ts") is not None


def test_kind_of_returns_none_for_unroutable_without_body():
    assert mapper.kind_of("packages/util/src/plain.ts") is None
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_mapper_kind_of.py -q`
Expected: FAIL — `AttributeError: module 'sg_mapper' has no attribute 'kind_of'`.

- [ ] **Step 3: Add `kind_of` to `mapper.py`**

Insert after `_file_routed_kind` (`mapper.py:166`):

```python
def kind_of(rel, body=None):
    """Production kind-inference for ONE repo-relative path, blind to any ground-truth label. Same two-step
    inference enumerate_surface uses per file: file-routed kinds by LOCATION (FILE_ROUTE_RULES via
    _file_routed_kind) first; else the first content-matching call-registered kind (KIND_SIGNALS insertion
    order, file-routed kinds skipped) when `body` is given. Returns the kind str or None. Factored so
    single-file callers (DETECT run_bench) route the SAME way the repo scan does; test_mapper_kind_of locks
    no-drift vs enumerate_surface."""
    rel = rel.replace(os.sep, "/")
    fk = _file_routed_kind(rel)
    if fk is not None:
        return fk
    if body is None:
        return None
    file_routed = set(FILE_ROUTE_RULES)
    for kind, (rx, _reliable) in KIND_SIGNALS.items():
        if kind in file_routed:
            continue
        if rx.search(body):
            return kind
    return None
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_mapper_kind_of.py -q`
Expected: PASS (3 tests).
Run: `rtk proxy python3 -m pytest tests/test_mapper.py -q`
Expected: PASS — existing mapper behavior unchanged (additive helper only).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/mapper.py tests/test_mapper_kind_of.py
git commit -m "mapper: additive kind_of() single-file inference (no-drift vs enumerate)"
```

---

## Task 3: `detect.py` scaffold + `select_detectors` (the `applies_to` router)

**Wave:** 2
**Blocks:** Task 4, 5, 6, 7, 8
**Blocked by:** Task 1, Task 2

**Files:**
- Create: `orchestrator/detect.py`
- Create: `tests/test_detect_select.py`

- [ ] **Step 1: Write the failing routing tests**

Create `tests/test_detect_select.py`:

```python
"""test_detect_select.py — select_detectors routing units (pure, deterministic, no LLM). Fixture manifests."""
import importlib.util, os, tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))

BASELINE = {"id": "baseline", "kind": "llm", "applies_to": {"kinds": [], "signal": "", "always": True}}
AUTH = {"id": "auth", "kind": "llm",
        "applies_to": {"kinds": ["http-file-route"], "signal": r"\b(token|session)\b", "always": False}}
FINANCE = {"id": "finance", "kind": "llm",
           "applies_to": {"kinds": ["http-defn-call"], "signal": r"\b(payout|refund)\b", "always": False}}
ORACLE = {"id": "oracle", "exec": ["bun", "x"], "scope": "per-file", "class": "S9"}  # not kind==llm
NOAPPLIES = {"id": "weird", "kind": "llm"}  # no applies_to → never routed


def _tmpfile(text):
    fd, p = tempfile.mkstemp(suffix=".ts")
    with os.fdopen(fd, "w") as fh:
        fh.write(text)
    return p


def _ids(sel):
    return [d["id"] for d in sel]


def test_baseline_always_included_even_with_no_match():
    p = _tmpfile("const x = 1\n")
    sel = detect.select_detectors(p, "unknown-kind", [BASELINE, AUTH, FINANCE])
    assert _ids(sel) == ["baseline"]


def test_kind_match_includes_specialist():
    p = _tmpfile("const x = 1\n")  # no signal hit; routed by KIND
    sel = detect.select_detectors(p, "http-file-route", [BASELINE, AUTH])
    assert set(_ids(sel)) == {"baseline", "auth"}


def test_signal_match_includes_specialist_on_content():
    p = _tmpfile("if (refund > 0) doPayout()\n")  # finance kind is http-defn-call; here matched by SIGNAL
    sel = detect.select_detectors(p, "graphql-resolver", [BASELINE, FINANCE])
    assert set(_ids(sel)) == {"baseline", "finance"}


def test_always_true_specialist_included():
    always = {"id": "everywhere", "kind": "llm", "applies_to": {"always": True}}
    p = _tmpfile("const x = 1\n")
    sel = detect.select_detectors(p, "x", [BASELINE, always])
    assert set(_ids(sel)) == {"baseline", "everywhere"}


def test_no_applies_to_not_routed():
    p = _tmpfile("const token = 1\n")
    sel = detect.select_detectors(p, "x", [BASELINE, NOAPPLIES])
    assert _ids(sel) == ["baseline"]


def test_non_llm_detector_skipped():
    p = _tmpfile("const x = 1\n")
    sel = detect.select_detectors(p, "x", [BASELINE, ORACLE])
    assert _ids(sel) == ["baseline"]


def test_never_narrower_routes_baseline_plus_all_matches():
    p = _tmpfile("session refund payout token\n")
    sel = detect.select_detectors(p, "http-file-route", [BASELINE, AUTH, FINANCE])
    assert set(_ids(sel)) == {"baseline", "auth", "finance"}


def test_baseline_absent_returns_none():
    p = _tmpfile("const x = 1\n")
    assert detect.select_detectors(p, "x", [AUTH, FINANCE]) is None
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_select.py -q`
Expected: FAIL — `detect.py` does not exist / no `select_detectors`.

- [ ] **Step 3: Create `detect.py` scaffold + `select_detectors`**

Create `orchestrator/detect.py`:

```python
#!/usr/bin/env python3
"""detect.py — the DETECT orchestrator: production scan harness + k=1 recall measurement.
SoT: docs/specs/2026-06-18-detect-orchestrator-design.md.

Per attack-surface target: select baseline + matched specialist detectors (applies_to), dispatch gate.py k=1
per (target, detector) as a SUBPROCESS, union findings across detectors, semantic-merge paraphrases, emit a
per-target report + a run-level coverage map. The SAME core over the corpus (--bench) measures k=1 production
recall. no-false-clean throughout: clean ONLY when every selected detector ran ok with zero findings.

Reuses (path-loaded, project convention — spec_from_file_location, NO __init__.py, stdlib-only): mapper,
semantic_merge, prevent/registry, bench. gate.py is invoked as a SUBPROCESS, never path-loaded."""
import argparse, json, os, re, sys, subprocess, importlib.util
import concurrent.futures as cf

_HERE = os.path.dirname(os.path.abspath(__file__))
_ROOT = os.path.dirname(_HERE)
GATE = os.path.join(_HERE, "gate.py")


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


mapper = _load("sg_mapper", os.path.join(_HERE, "mapper.py"))
semantic_merge = _load("sg_semantic_merge", os.path.join(_HERE, "semantic_merge.py"))
registry = _load("sg_registry", os.path.join(_ROOT, "prevent", "registry.py"))
bench = _load("sg_bench", os.path.join(_ROOT, "bench.py"))

REVERSE_LEVEL = {"error": "high", "warning": "medium", "note": "low"}


def _read_text(p):
    try:
        return open(p, encoding="utf-8", errors="replace").read()
    except OSError:
        return ""


def select_detectors(target_path, kind, detectors):
    """baseline (always, by GUARDED id lookup) + every kind=='llm' specialist whose applies_to matches the
    target. APPLIES = applies_to.always OR kind in applies_to.kinds OR applies_to.signal regex hits the file
    content. Returns the selected list, or None if NO baseline is registered — the caller turns None into a
    RUN-LEVEL hard stop (a missing baseline is reduced coverage on EVERYTHING, never a per-target footnote).
    Never narrower than the manifest match; a detector with no applies_to is simply not routed (honest)."""
    baseline = next((d for d in detectors if d.get("id") == "baseline"), None)
    if baseline is None:
        return None
    sel, seen = [baseline], {"baseline"}
    body = None
    for d in detectors:
        if d.get("id") in seen or d.get("kind") != "llm":
            continue
        ap = d.get("applies_to")
        if not ap:
            continue
        ok = bool(ap.get("always")) or (kind in ap.get("kinds", []))
        if not ok and ap.get("signal"):
            if body is None:
                body = _read_text(target_path)
            ok = re.search(ap["signal"], body) is not None
        if ok:
            sel.append(d); seen.add(d.get("id"))
    return sel
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_detect_select.py -q`
Expected: PASS (8 tests).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_select.py
git commit -m "detect: scaffold + select_detectors applies_to router (baseline-guarded)"
```

---

## Task 4: `detect.py` `dispatch_one` + `_degraded`

**Wave:** 3
**Blocks:** Task 5
**Blocked by:** Task 1, Task 3

**Files:**
- Modify: `orchestrator/detect.py` (append `_degraded`, `dispatch_one`)
- Create: `tests/test_detect_dispatch.py`

- [ ] **Step 1: Write the failing tests (gate stubbed via a fake GATE script)**

Create `tests/test_detect_dispatch.py`:

```python
"""test_detect_dispatch.py — dispatch_one parses a contract emit on success; a failed/garbled gate → synthetic
degraded dict, NEVER an empty-findings clean. We stub the gate by pointing detect.GATE at a tiny fake script
that writes a known --emit file (or exits non-zero), so no real gate.py/LLM runs."""
import importlib.util, os, tempfile, json, textwrap

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))

DET = {"id": "baseline", "kind": "llm"}


def _fake_gate(body):
    fd, p = tempfile.mkstemp(suffix=".py")
    with os.fdopen(fd, "w") as fh:
        fh.write(textwrap.dedent(body))
    return p


def test_dispatch_one_parses_contract_emit(monkeypatch):
    # fake gate: find the --emit path in argv, write a contract dict there, exit 0
    fake = _fake_gate("""
        import sys, json
        a = sys.argv
        emit = a[a.index("--emit") + 1]
        json.dump({"detector": "baseline", "status": "ok",
                   "findings": [{"ruleId": "baseline", "level": "error", "class": "S1",
                                 "message": "reset token single-use not enforced", "file": "t.ts",
                                 "line": 0, "symbol": "", "sev": "high", "rolls": 1, "of": 1}],
                   "coverage": {"scanned": ["t.ts"], "unresolved": []}}, open(emit, "w"))
    """)
    monkeypatch.setattr(detect, "GATE", fake)
    out = tempfile.mkdtemp()
    r = detect.dispatch_one("/abs/t.ts", DET, 1, None, "sonnet", "medium", out)
    assert r["status"] == "ok"
    assert r["findings"][0]["sev"] == "high"


def test_dispatch_one_gate_failure_is_degraded_not_clean(monkeypatch):
    fake = _fake_gate("import sys\nsys.exit(3)\n")  # non-zero, no emit written
    monkeypatch.setattr(detect, "GATE", fake)
    out = tempfile.mkdtemp()
    r = detect.dispatch_one("/abs/t.ts", DET, 1, None, "sonnet", "medium", out)
    assert r["status"] == "error"
    assert r["findings"] == []
    assert r["coverage"]["unresolved"]  # surfaced, never silent


def test_dispatch_one_unparseable_emit_is_degraded(monkeypatch):
    fake = _fake_gate("""
        import sys
        a = sys.argv
        open(a[a.index("--emit") + 1], "w").write("{not json")
    """)
    monkeypatch.setattr(detect, "GATE", fake)
    out = tempfile.mkdtemp()
    r = detect.dispatch_one("/abs/t.ts", DET, 1, None, "sonnet", "medium", out)
    assert r["status"] == "error" and r["findings"] == []
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_dispatch.py -q`
Expected: FAIL — no `dispatch_one`.

- [ ] **Step 3: Append `_degraded` + `dispatch_one` to `detect.py`**

```python
def _degraded(detector_id, reason):
    """A failed/garbled detector run — surfaced as COVERAGE-INCOMPLETE, NEVER an empty-findings clean."""
    return {"detector": detector_id, "status": "error", "findings": [],
            "coverage": {"scanned": [], "unresolved": [reason]}}


def dispatch_one(target, detector, k, cfg, model, effort, out_dir):
    """Subprocess gate.py --detector <id> --k <k> --emit <json> --report <md> (exactly as mapper.dispatch shells
    gate.py). Parse the contract-shape --emit JSON on success. rc!=0 OR missing/unparseable emit → synthetic
    degraded dict (mirrors mapper.dispatch returning None = 'never clean')."""
    os.makedirs(out_dir, exist_ok=True)
    did = detector.get("id")
    stem = f"{os.path.basename(target)}.{did}"
    emit = os.path.join(out_dir, stem + ".emit.json")
    rep = os.path.join(out_dir, stem + ".report.md")
    cmd = [sys.executable, GATE, target, "--detector", did, "--k", str(k),
           "--emit", emit, "--report", rep, "--model", model, "--effort", effort]
    if cfg:
        cmd += ["--config-dir", cfg]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0 or not os.path.isfile(emit):
        return _degraded(did, f"gate failed (rc={r.returncode}): {r.stderr.strip()[:200]}")
    try:
        with open(emit, encoding="utf-8") as fh:
            return json.load(fh)
    except (OSError, ValueError) as e:
        return _degraded(did, f"unparseable emit: {e}")
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_detect_dispatch.py -q`
Expected: PASS (3 tests).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_dispatch.py
git commit -m "detect: dispatch_one subprocess gate + degraded-never-clean"
```

---

## Task 5: `detect.py` `detect_target` (cross-detector union → merge → status)

**Wave:** 4
**Blocks:** Task 6
**Blocked by:** Task 4

**Files:**
- Modify: `orchestrator/detect.py` (append `_emit_to_groups`, `detect_target`, `_write_target_report`)
- Create: `tests/test_detect_target.py`

- [ ] **Step 1: Write the failing no-false-clean tests (dispatch_one stubbed)**

Create `tests/test_detect_target.py`:

```python
"""test_detect_target.py — detect_target status invariants (no-false-clean). dispatch_one + merge stubbed so no
gate/LLM runs. clean is reachable ONLY when every selected detector ran ok with zero findings."""
import importlib.util, os, tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))
BASELINE = {"id": "baseline", "kind": "llm", "applies_to": {"always": True}}


def _tmp_target():
    fd, p = tempfile.mkstemp(suffix=".ts")
    os.close(fd)
    return p


def _stub_dispatch(results):
    it = iter(results)
    return lambda *a, **k: next(it)


def test_all_ok_zero_findings_is_clean(monkeypatch):
    monkeypatch.setattr(detect, "dispatch_one",
                        _stub_dispatch([{"detector": "baseline", "status": "ok", "findings": [],
                                         "coverage": {"scanned": ["t"], "unresolved": []}}]))
    r = detect.detect_target(_tmp_target(), "x", [BASELINE], out_dir=tempfile.mkdtemp())
    assert r["status"] == "clean"


def test_detector_error_is_degraded_not_clean(monkeypatch):
    monkeypatch.setattr(detect, "dispatch_one",
                        _stub_dispatch([{"detector": "baseline", "status": "error", "findings": [],
                                         "coverage": {"scanned": [], "unresolved": ["gate failed"]}}]))
    r = detect.detect_target(_tmp_target(), "x", [BASELINE], out_dir=tempfile.mkdtemp())
    assert r["status"] == "degraded"


def test_findings_present_is_findings(monkeypatch):
    monkeypatch.setattr(detect, "dispatch_one",
                        _stub_dispatch([{"detector": "baseline", "status": "ok",
                                         "findings": [{"message": "bug", "sev": "high", "rolls": 1, "of": 1}],
                                         "coverage": {"scanned": ["t"], "unresolved": []}}]))
    r = detect.detect_target(_tmp_target(), "x", [BASELINE], out_dir=tempfile.mkdtemp())
    assert r["status"] == "findings"
    assert r["findings"][0]["title"] == "bug"


def test_merge_degraded_is_degraded(monkeypatch):
    monkeypatch.setattr(detect, "dispatch_one",
                        _stub_dispatch([{"detector": "baseline", "status": "ok",
                                         "findings": [{"message": "a", "sev": "high", "rolls": 1, "of": 1},
                                                      {"message": "b", "sev": "low", "rolls": 1, "of": 1}],
                                         "coverage": {"scanned": ["t"], "unresolved": []}}]))
    monkeypatch.setattr(detect.semantic_merge, "merge_groups", lambda g, **k: (list(g), True))
    r = detect.detect_target(_tmp_target(), "x", [BASELINE], out_dir=tempfile.mkdtemp())
    assert r["status"] == "degraded"
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_target.py -q`
Expected: FAIL — no `detect_target`.

- [ ] **Step 3: Append `_emit_to_groups`, `detect_target`, `_write_target_report`**

```python
def _emit_to_groups(emits):
    """Reconstruct semantic_merge group dicts {title,sev,rolls:set} from every detector's contract findings.
    sev falls back from the contract level when an emit (e.g. oracle) carries no additive 'sev'."""
    groups = []
    for e in emits:
        for f in e.get("findings", []):
            sev = f.get("sev") or REVERSE_LEVEL.get(f.get("level"), "unrated")
            n = f.get("rolls", 1) or 1
            groups.append({"title": f.get("message", ""), "sev": sev, "rolls": set(range(n))})
    return groups


def detect_target(target, kind, detectors, k=1, cfg=None, model="sonnet", effort="medium",
                  out_dir="detect-reports"):
    """select → dispatch each selected detector (bounded ThreadPoolExecutor) → cross-detector union →
    semantic_merge.merge_groups → per-target merged report. status: 'degraded' if ANY detector degraded/error
    OR merge degraded; else 'clean' iff zero merged findings; else 'findings'. clean ONLY when every selected
    detector ran ok with zero findings (no-false-clean)."""
    sel = select_detectors(target, kind, detectors)
    if sel is None:  # defensive — run-level guard should have aborted before any target
        return {"target": target, "kind": kind, "detectors": [], "status": "degraded",
                "findings": [], "merged_report_path": None}
    with cf.ThreadPoolExecutor(max_workers=max(1, len(sel))) as ex:
        emits = list(ex.map(lambda d: dispatch_one(target, d, k, cfg, model, effort, out_dir), sel))
    degraded = any(e.get("status") != "ok" for e in emits)
    groups = _emit_to_groups(emits)
    if len(groups) >= 2:
        merged, merge_degraded = semantic_merge.merge_groups(groups, config_dir=cfg, model=model, effort=effort)
    else:
        merged, merge_degraded = groups, False
    degraded = degraded or merge_degraded
    status = "degraded" if degraded else ("clean" if not merged else "findings")
    rep = _write_target_report(target, kind, sel, emits, merged, status, merge_degraded, out_dir)
    return {"target": target, "kind": kind, "detectors": [d.get("id") for d in sel],
            "status": status, "findings": merged, "merged_report_path": rep}


def _write_target_report(target, kind, sel, emits, merged, status, merge_degraded, out_dir):
    os.makedirs(out_dir, exist_ok=True)
    lines = [f"# DETECT — {os.path.relpath(target)}", "",
             f"- kind: `{kind}`", f"- detectors: {', '.join(d.get('id') for d in sel)}",
             f"- status: **{status}**", ""]
    incomplete = [e for e in emits if e.get("status") != "ok"]
    if incomplete:
        lines.append("## COVERAGE-INCOMPLETE (not a clean)")
        for e in incomplete:
            lines.append(f"- `{e.get('detector')}`: {', '.join(e.get('coverage', {}).get('unresolved', []))}")
        lines.append("")
    if merge_degraded:
        lines.append("- ⚠️ cross-detector semantic-merge DEGRADED — findings shown un-deduped.")
        lines.append("")
    lines.append("## Merged findings")
    for g in merged:
        lines.append(f"- [{g['sev']}] {g['title']}")
    lines.append("")
    rep = os.path.join(out_dir, os.path.basename(target) + ".detect.md")
    open(rep, "w", encoding="utf-8").write("\n".join(lines))
    return rep
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_detect_target.py -q`
Expected: PASS (4 tests).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_target.py
git commit -m "detect: detect_target cross-detector union+merge+status (clean only all-ok-zero)"
```

---

## Task 6: `detect.py` `run` (production scan + coverage map)

**Wave:** 5
**Blocks:** Task 8
**Blocked by:** Task 5

**Files:**
- Modify: `orchestrator/detect.py` (append `run`)
- Create: `tests/test_detect_run.py`

- [ ] **Step 1: Write the failing coverage-map test (detect_target stubbed)**

Create `tests/test_detect_run.py`:

```python
"""test_detect_run.py — run() assembles the 3-bucket coverage map (disjoint) + per-target status + skipped +
coverage_incomplete, and baseline-absent is a RUN-LEVEL hard stop. detect_target stubbed so no gate/LLM runs."""
import importlib.util, os, tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))
BASELINE = {"id": "baseline", "kind": "llm", "applies_to": {"always": True}}


def _route_repo():
    """Two route files so prioritize/enumerate yield >=2 entries; max_targets=1 forces a budget-dropped bucket."""
    d = tempfile.mkdtemp()
    r = os.path.join(d, "apps/api/src/pages/auth")  # consecutive src/pages -> http-file-route (live FILE_ROUTE_RULES)
    os.makedirs(r, exist_ok=True)
    open(os.path.join(r, "login.ts"), "w").write("export async function POST(req){return new Response('a')}\n")
    open(os.path.join(r, "refresh.ts"), "w").write("export async function POST(req){return new Response('b')}\n")
    return d


def test_run_baseline_absent_is_hard_stop():
    out = detect.run(_route_repo(), [{"id": "auth", "kind": "llm"}], [], out_dir=tempfile.mkdtemp())
    assert out["status"] == "COVERAGE-INCOMPLETE"
    assert "baseline" in out["error"]


def test_run_buckets_disjoint_and_status_present(monkeypatch):
    monkeypatch.setattr(detect, "detect_target",
                        lambda target, kind, dets, *a, **k: {"target": target, "kind": kind,
                                                             "detectors": ["baseline"], "status": "clean",
                                                             "findings": [], "merged_report_path": "/r.md"})
    d = _route_repo()
    cmap = detect.run(d, [BASELINE], [("bad/detector.json", "boom")], out_dir=tempfile.mkdtemp(), max_targets=1)
    scanned = {e["path"] for e in cmap["enumerated_scanned"]}
    dropped = set(cmap["enumerated_budget_dropped"])
    notenum = {e["path"] for e in cmap["not_enumerated"]}
    assert not (scanned & dropped) and not (scanned & notenum) and not (dropped & notenum)  # disjoint
    assert dropped, "max_targets=1 over 2 entries must leave a budget-dropped entry"
    assert cmap["per_target_status"] and cmap["per_target_status"][0]["status"] == "clean"
    assert cmap["skipped_detectors"][0]["reason"] == "boom"
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_run.py -q`
Expected: FAIL — no `run`.

- [ ] **Step 3: Append `run` to `detect.py`**

```python
def run(root, detectors, skipped, k=1, cfg=None, model="sonnet", effort="medium",
        out_dir="detect-reports", max_targets=None):
    """Production scan over a real repo. Returns the run-level coverage map: mapper's 3 disjoint buckets +
    per_target_status + skipped_detectors + coverage_incomplete. BASELINE-ABSENT = run-level hard stop BEFORE
    scanning any target (a missing baseline is reduced coverage on EVERYTHING)."""
    if not any(d.get("id") == "baseline" for d in detectors):
        return {"status": "COVERAGE-INCOMPLETE",
                "error": "no baseline detector registered — every target's floor is gone; run aborted",
                "skipped_detectors": [{"manifest": m, "reason": r} for m, r in skipped]}
    entries = mapper.prioritize(mapper.enumerate_surface(root))
    act = mapper.actual_surface(root)
    budget = entries if max_targets is None else entries[:max_targets]
    scanned, per_target, incomplete = {}, [], []
    for e in budget:
        res = detect_target(e.path, e.kind, detectors, k, cfg, model, effort, out_dir)
        if res["merged_report_path"]:
            scanned[e.path] = res["merged_report_path"]
        per_target.append({"path": os.path.relpath(e.path, root), "kind": e.kind, "status": res["status"]})
        if res["status"] == "degraded":
            incomplete.append(os.path.relpath(e.path, root))
    cmap = mapper.coverage_map(entries, act, scanned)
    cmap["per_target_status"] = per_target
    cmap["skipped_detectors"] = [{"manifest": m, "reason": r} for m, r in skipped]
    cmap["coverage_incomplete"] = incomplete
    return cmap
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_detect_run.py -q`
Expected: PASS (2 tests).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_run.py
git commit -m "detect: run() production scan + 3-bucket coverage map + baseline hard-stop"
```

---

## Task 7: `detect.py` `run_bench` (k=1 recall + autoscore no-false-clean + blind routing)

**Wave:** 6
**Blocks:** Task 8
**Blocked by:** Task 5, Task 2

**Files:**
- Modify: `orchestrator/detect.py` (append `_cell_kind`, `run_bench`)
- Create: `tests/test_detect_bench.py`

- [ ] **Step 1: Write the failing tests (autoscore classification + routing-blindness)**

Create `tests/test_detect_bench.py`:

```python
"""test_detect_bench.py — run_bench no-false-clean + teaching-to-test guard. detect_target stubbed (no gate/LLM).
- autoscore: literal-symbol cell whose titles MISS the canonical → scorable recall-0 (true miss), NOT incomplete.
- phrase cell (symbol absent from vuln source) whose titles MISS → coverage_incomplete['autoscore-unscorable'].
- routing-blindness: kind handed to detect_target comes from mapper.kind_of, NEVER from canonical['class']."""
import importlib.util, os, tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))
BASELINE = {"id": "baseline", "kind": "llm", "applies_to": {"always": True}}


def _cell(cid, klass, symbol, vuln_src):
    d = tempfile.mkdtemp()
    open(os.path.join(d, "vuln.ts"), "w").write(vuln_src)
    return {"id": cid, "class": klass, "canonical_symbol": symbol, "_dir": d, "file": "apps/api/x.ts"}


def test_run_bench_baseline_absent_hard_stop():
    fo, meta = detect.run_bench([], [{"id": "auth", "kind": "llm"}], [], out_dir=tempfile.mkdtemp())
    assert meta["status"] == "COVERAGE-INCOMPLETE" and fo == {}


def test_autoscore_literal_symbol_miss_is_scorable_recall0(monkeypatch):
    # canonical symbol IS a literal token in the vuln source → autoscorable; titles miss → real recall-0
    cell = _cell("S3-x", "S3", "getSchedulingConnectionById",
                 "export function getSchedulingConnectionById(id){return db.x(id)}\n")
    monkeypatch.setattr(detect, "detect_target",
                        lambda t, k, d, *a, **kw: {"status": "clean", "findings": [{"title": "unrelated note"}]})
    fo, meta = detect.run_bench([cell], [BASELINE], [], out_dir=tempfile.mkdtemp())
    assert "S3-x" in fo                              # scorable: bench will score it recall-0 (true miss)
    assert "S3-x" not in meta["coverage_incomplete"]
    assert meta["recall"] == 0.0 and meta["scored"] == 1


def test_phrase_canonical_miss_is_unscorable_not_recall0(monkeypatch):
    # canonical symbol is a PHRASE not present in vuln source → not autoscorable; titles miss → unscorable
    cell = _cell("S5-y", "S5", "escape email interpolation",
                 "export function render(t){return `<b>${t}</b>`}\n")
    monkeypatch.setattr(detect, "detect_target",
                        lambda t, k, d, *a, **kw: {"status": "clean", "findings": [{"title": "unrelated note"}]})
    fo, meta = detect.run_bench([cell], [BASELINE], [], out_dir=tempfile.mkdtemp())
    assert "S5-y" not in fo                          # NOT fed to bench (would be a false recall-0)
    assert meta["coverage_incomplete"]["S5-y"] == "autoscore-unscorable"


def test_scan_degraded_cell_is_incomplete(monkeypatch):
    cell = _cell("S3-z", "S3", "touchSession", "export function touchSession(id){return db.u(id)}\n")
    monkeypatch.setattr(detect, "detect_target",
                        lambda t, k, d, *a, **kw: {"status": "degraded", "findings": []})
    fo, meta = detect.run_bench([cell], [BASELINE], [], out_dir=tempfile.mkdtemp())
    assert "S3-z" not in fo and meta["coverage_incomplete"]["S3-z"] == "scan-degraded"


def test_routing_kind_from_mapper_not_canonical_class(monkeypatch):
    cell = _cell("S1-r", "S1", "reset token single-use", "export function reset(){}\n")
    captured = {}
    monkeypatch.setattr(detect.mapper, "kind_of", lambda rel, body=None: "edge-function")
    def _spy(target, kind, dets, *a, **kw):
        captured["kind"] = kind
        return {"status": "clean", "findings": []}
    monkeypatch.setattr(detect, "detect_target", _spy)
    detect.run_bench([cell], [BASELINE], [], out_dir=tempfile.mkdtemp())
    assert captured["kind"] == "edge-function"       # mapper-derived
    assert captured["kind"] != cell["class"]         # NOT the canonical class (teaching-to-test guard)
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_bench.py -q`
Expected: FAIL — no `run_bench`.

- [ ] **Step 3: Append `_cell_kind` + `run_bench` to `detect.py`**

```python
def _cell_kind(cell, vuln_path):
    """Derive kind the SAME way production does — mapper.kind_of on the cell's production path (canonical['file'])
    or the vuln basename, BLIND to canonical['class']. Routing by the known class would be teaching-to-test."""
    rel = cell.get("file") or os.path.basename(vuln_path)
    return mapper.kind_of(rel, body=_read_text(vuln_path))


def run_bench(cells, detectors, skipped, k=1, cfg=None, model="sonnet", effort="medium", out_dir="detect-bench"):
    """k=1 recall over corpus cells (the headline number). Returns (findings_out, meta). findings_out =
    {cell_id: [merged group title strings]} for SCORABLE cells ONLY, so bench.py --findings never sees a cell it
    would wrongly score recall-0. BASELINE-ABSENT = run-level hard stop. A cell is:
      - scan-degraded (detect_target degraded) → coverage_incomplete['scan-degraded'], NOT scored;
      - autoscorable (canonical_symbol is a contiguous case-insensitive substring of the vuln source) →
        scored (a substring miss is a REAL recall-0);
      - else PHRASE canonical: scored only if it flagged; a miss → coverage_incomplete['autoscore-unscorable']
        (hand-judged), NEVER recall-0.
    Cells without a vuln.ts (deterministic-detector cells) are out of LLM-recall scope → skipped_non_llm."""
    if not any(d.get("id") == "baseline" for d in detectors):
        return {}, {"status": "COVERAGE-INCOMPLETE",
                    "error": "no baseline detector registered — run aborted",
                    "skipped_detectors": [{"manifest": m, "reason": r} for m, r in skipped]}
    findings_out, incomplete, skipped_non_llm, scored, hits = {}, {}, [], [], []
    for c in cells:
        cid = c["id"]
        vuln = os.path.join(c.get("_dir", ""), "vuln.ts")
        if not os.path.isfile(vuln):
            skipped_non_llm.append(cid)
            continue
        kind = _cell_kind(c, vuln)
        res = detect_target(vuln, kind, detectors, k, cfg, model, effort, out_dir)
        if res["status"] == "degraded":
            incomplete[cid] = "scan-degraded"
            continue
        titles = [g["title"] for g in res["findings"]]
        flagged = bench.is_flagged(c, titles)
        sym = (c.get("canonical_symbol") or "").lower()
        autoscorable = bool(sym) and sym in _read_text(vuln).lower()
        if flagged or autoscorable:
            findings_out[cid] = titles
            scored.append(cid)
            if flagged:
                hits.append(cid)
        else:
            incomplete[cid] = "autoscore-unscorable"
    meta = {"status": "ok", "k": k, "scored": len(scored),
            "recall": round(len(hits) / len(scored), 3) if scored else None,
            "hits": sorted(hits), "coverage_incomplete": incomplete,
            "skipped_non_llm": sorted(skipped_non_llm),
            "skipped_detectors": [{"manifest": m, "reason": r} for m, r in skipped]}
    return findings_out, meta
```

- [ ] **Step 4: Run test to verify it passes**

Run: `rtk proxy python3 -m pytest tests/test_detect_bench.py -q`
Expected: PASS (5 tests).

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_bench.py
git commit -m "detect: run_bench k=1 recall + autoscore no-false-clean + blind routing"
```

---

## Task 8: `detect.py` `main(argv)` CLI

**Wave:** 7
**Blocks:** —
**Blocked by:** Task 6, Task 7

**Files:**
- Modify: `orchestrator/detect.py` (append `main` + `__main__`)
- Create: `tests/test_detect_cli.py`

- [ ] **Step 1: Write the failing CLI tests (deterministic — baseline-absent hard stop end-to-end)**

Create `tests/test_detect_cli.py`:

```python
"""test_detect_cli.py — main(argv) wiring, deterministic. --bench against a detectors-root with NO baseline must
exit COVERAGE-INCOMPLETE (returncode 2). No-args (no root, no --bench) must error. No LLM."""
import importlib.util, os, tempfile, subprocess, sys, json

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DETECT = os.path.join(ROOT, "orchestrator", "detect.py")


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", DETECT)


def _empty_detectors_root():
    """A detectors-root whose domains/*/detectors/* has a single NON-baseline manifest → baseline-absent."""
    d = tempfile.mkdtemp()
    md = os.path.join(d, "domains", "security", "detectors", "auth")
    os.makedirs(md, exist_ok=True)
    json.dump({"id": "auth", "kind": "llm", "applies_to": {"always": False}},
              open(os.path.join(md, "detector.json"), "w"))
    return d


def test_bench_baseline_absent_exits_coverage_incomplete():
    droot = _empty_detectors_root()
    fo = os.path.join(tempfile.mkdtemp(), "f.json")
    r = subprocess.run([sys.executable, DETECT, "--bench", "--detectors-root", droot, "--findings-out", fo],
                       capture_output=True, text=True)
    assert r.returncode == 2, r.stderr
    assert "COVERAGE-INCOMPLETE" in r.stdout


def test_no_root_no_bench_errors():
    r = subprocess.run([sys.executable, DETECT], capture_output=True, text=True)
    assert r.returncode != 0  # argparse error


def test_main_returns_int():
    assert detect.main(["--bench", "--detectors-root", _empty_detectors_root()]) == 2
```

- [ ] **Step 2: Run it — verify it fails**

Run: `rtk proxy python3 -m pytest tests/test_detect_cli.py -q`
Expected: FAIL — no `main`.

- [ ] **Step 3: Append `main` + `__main__` to `detect.py`**

```python
def main(argv=None):
    ap = argparse.ArgumentParser(description="DETECT — production scan harness + k=1 recall (SoT: detect spec)")
    ap.add_argument("root", nargs="?", help="repo root to scan (omit with --bench)")
    ap.add_argument("--bench", action="store_true",
                    help="measure k=1 recall over the corpus instead of scanning a repo")
    ap.add_argument("--map", default=None, help="write the run-level coverage map JSON here (default stdout)")
    ap.add_argument("--k", type=int, default=1, help="rolls per (target,detector). 1=production; 3=flakiness audit")
    ap.add_argument("--model", default="sonnet")
    ap.add_argument("--effort", default="medium")
    ap.add_argument("--config-dir", default=None, help="CLAUDE_CONFIG_DIR for blind catch-test")
    ap.add_argument("--max-targets", type=int, default=None, help="budget cap (default: all enumerated)")
    ap.add_argument("--out-dir", default="detect-reports")
    ap.add_argument("--findings-out", default=None, help="bench-shape {cell_id:[str]} JSON (--bench mode)")
    ap.add_argument("--detectors-root", default=_ROOT,
                    help="root whose domains/*/detectors/*/detector.json is loaded")
    a = ap.parse_args(argv)
    detectors, skipped = registry.load(a.detectors_root)
    if a.bench:
        cells = bench.load_cells()
        findings_out, meta = run_bench(cells, detectors, skipped, a.k, a.config_dir, a.model, a.effort, a.out_dir)
        if a.findings_out:
            with open(a.findings_out, "w", encoding="utf-8") as fh:
                json.dump(findings_out, fh, indent=2)
        print(json.dumps(meta, indent=2))
        return 2 if meta.get("status") == "COVERAGE-INCOMPLETE" else 0
    if not a.root:
        ap.error("need a repo root (or --bench)")
    cmap = run(os.path.abspath(a.root), detectors, skipped, a.k, a.config_dir, a.model, a.effort,
               a.out_dir, a.max_targets)
    js = json.dumps(cmap, indent=2)
    if a.map:
        open(a.map, "w", encoding="utf-8").write(js)
    else:
        print(js)
    return 2 if cmap.get("status") == "COVERAGE-INCOMPLETE" else 0


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

- [ ] **Step 4: Run the new test + the FULL suite**

Run: `rtk proxy python3 -m pytest tests/test_detect_cli.py -q`
Expected: PASS (3 tests).
Run: `rtk proxy python3 -m pytest -q`
Expected: PASS — all detect + gate + mapper + pre-existing tests green.

- [ ] **Step 5: Commit**

```bash
git add orchestrator/detect.py tests/test_detect_cli.py
git commit -m "detect: main() CLI (--bench/--findings-out/--map/--detectors-root)"
```

---

## Task 9: LLM-live k=1 recall smoke (rate-only — NEVER a ship gate)

**Wave:** 8
**Blocks:** —
**Blocked by:** Task 8

**Files:**
- Create: `docs/validation/2026-06-18-detect-k1-recall-smoke.md`
- Create: `tests/test_detect_llm_smoke.py` (opt-in; skipped unless `SG_LLM_LIVE=1`)

> Per create-tests rule `llm-live-selftest-not-a-ship-gate`: the LLM-live run is rate-only evidence. It MUST NOT be in the default pytest gate. The smoke test is `skipif`-guarded; the authoritative artifact is the validation doc.

- [ ] **Step 1: Write the opt-in smoke test (skipped by default)**

Create `tests/test_detect_llm_smoke.py`:

```python
"""test_detect_llm_smoke.py — LLM-LIVE, rate-only, NOT a ship gate (skipped unless SG_LLM_LIVE=1).
Runs detect.run_bench on ONE autoscorable cell with the real registry, k=1, blind config-dir; asserts the
canonical is flagged. A single stochastic miss is NOT a regression — see the validation doc."""
import importlib.util, os, tempfile
import pytest

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
pytestmark = pytest.mark.skipif(not os.environ.get("SG_LLM_LIVE"),
                                reason="LLM-live rate-only smoke; set SG_LLM_LIVE=1 + blind creds to run")


def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod


detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py"))


def test_autoscorable_cell_flags_canonical_live():
    cfg = os.environ.get("SG_CONFIG_DIR")  # blind config-dir with fresh creds
    detectors, skipped = detect.registry.load(ROOT)
    cells = [c for c in detect.bench.load_cells() if c["id"] == "S3-calendar-tenant-isolation-idor"]
    assert cells, "expected the autoscorable S3 calendar cell in the corpus"
    fo, meta = detect.run_bench(cells, detectors, skipped, k=1, cfg=cfg, out_dir=tempfile.mkdtemp())
    # rate-only: record the number; a miss is data, not a hard failure
    print("LIVE k=1 recall:", meta["recall"], "findings:", fo)
    assert meta["status"] == "ok"
```

- [ ] **Step 2: Verify it is SKIPPED in the default suite**

Run: `rtk proxy python3 -m pytest tests/test_detect_llm_smoke.py -q`
Expected: 1 skipped (no `SG_LLM_LIVE`). Confirms it never gates a normal run.

- [ ] **Step 3: Write the validation doc with the exact live command + creds probe**

Create `docs/validation/2026-06-18-detect-k1-recall-smoke.md`:

```markdown
# DETECT k=1 recall smoke (rate-only) — 2026-06-18

audience: AI coding agents first. BLUF. This is RATE-ONLY evidence, NEVER a ship gate
(create-tests `llm-live-selftest-not-a-ship-gate`). Production recall = k=1 per (target,detector).

## Blind creds probe FIRST (an expired token makes claude -p return empty → ZERO findings → a false clean)
CLAUDE_CONFIG_DIR=/tmp/sg_cfg claude -p --model sonnet --dangerously-skip-permissions 'PONG'
# on 401: cp ~/.claude/.credentials.json /tmp/sg_cfg/.credentials.json

## Run (one autoscorable cell, k=1, blind)
SG_LLM_LIVE=1 SG_CONFIG_DIR=/tmp/sg_cfg rtk proxy python3 -m pytest tests/test_detect_llm_smoke.py -q -s
# or the whole corpus:
rtk proxy python3 orchestrator/detect.py --bench --config-dir /tmp/sg_cfg \
    --findings-out /tmp/detect_findings.json
rtk proxy python3 bench.py --findings /tmp/detect_findings.json

## Record (fill after running)
- date / commit:
- scored cells (autoscorable ∪ flagged):
- coverage_incomplete (autoscore-unscorable + scan-degraded):
- k=1 recall (MEASURED, run_bench instrument — NOT comparable to the hand-judged ~86% until the fast-follow
  re-baselines the floor through run_bench):
```

- [ ] **Step 4: Commit**

```bash
git add tests/test_detect_llm_smoke.py docs/validation/2026-06-18-detect-k1-recall-smoke.md
git commit -m "detect: rate-only LLM-live recall smoke (skipif-guarded) + validation doc"
```

---

## Self-Review

**1. Spec coverage:**
- Component A (`gate.py --emit`, LEVEL map, additive sev/rolls/of, build dict not contract.emit(), byte-identical report) → Task 1 ✅ (golden generated from an INDEPENDENT verbatim copy of the pre-extraction inline block, so it locks extraction fidelity not just determinism; lazy contract path-load).
- Component B (`select_detectors` applies_to router, guarded baseline id lookup, baseline-absent run-level hard stop, never narrower, registry.load + skipped) → Task 3 (router) + Task 6/Task 7 (run-level hard stop in `run`/`run_bench`) ✅.
- Component C (`dispatch_one` subprocess gate, degraded-never-clean) → Task 4 ✅.
- Component D (`detect_target` select→dispatch parallel→reconstruct groups→cross-merge→status; clean only all-ok-zero) → Task 5 ✅.
- Component E (`run` 3-bucket map + per-target status + skipped + incomplete + per-target findings; `run_bench` blind kind, autoscore classify, phrase-unscorable, scan-degraded; `main` argparse) → Tasks 6, 7, 8 ✅.
- Testing test-1..test-7 → test_detect_select / test_gate_emit / test_detect_target / test_detect_run / test_detect_bench (autoscore + routing-blindness) / test_detect_cli / test_detect_llm_smoke ✅.
- Deferred items listed in "NOT covered / deferred" ✅.
- `run`'s per-target findings JSON keyed by repo relpath: the coverage map's `enumerated_scanned` carries `{path, report}` (report path) + `per_target_status` carries relpath+status; per-target findings live in each detector's emit JSON + the per-target `.detect.md`. NOTE: a single consolidated repo-relpath→findings JSON is thinner than the spec's wording (the spec says "per-target findings JSON keyed by repo relpath"). **Gap closed:** Task 6 emits report paths, not a findings-by-relpath map. To match the spec precisely, `run` should also write a `{relpath: [titles]}` map. → Added below.

**Gap fix (apply in Task 6, Step 3):** extend `run` to also collect a findings map. Add `findings_by_path = {}` before the loop; inside the loop after `res`: `findings_by_path[os.path.relpath(e.path, root)] = [g["title"] for g in res["findings"]]`; and `cmap["findings_by_path"] = findings_by_path` after building cmap. Add an assertion to `test_detect_run.py::test_run_buckets_disjoint_and_status_present`: `assert "findings_by_path" in cmap`.

**2. Placeholder scan:** No TBD/TODO/"handle edge cases"/"similar to". Every code step is complete and runnable. ✅

**3. Type consistency:**
- group dict `{title, sev, rolls:set}` consistent across `_emit_to_groups` → `semantic_merge.merge_groups` → `_write_target_report`/`run_bench` (read `g["title"]`). ✅
- emit dict `{detector, status, findings, coverage:{scanned, unresolved}}` consistent: `build_emit_dict` produces it; `dispatch_one`/`_degraded` return it; `detect_target` reads `e["status"]`, `e["findings"]`. ✅
- `detect_target` returns `{target, kind, detectors, status, findings, merged_report_path}` — `run` reads `merged_report_path`/`status`/`findings`; `run_bench` reads `status`/`findings`. ✅
- `select_detectors` returns list or None; `run`/`run_bench` guard baseline at run level (None never reached in `detect_target` in normal flow; defensive branch present). ✅
- `kind_of(rel, body=None)` signature matches `_cell_kind` + Task 2 tests. ✅

**4. Wave plan check:** Every task has Wave/Blocks/Blocked-by. Wave table matches. Wave 1 = Task 1 (gate.py + test_gate_emit + golden) and Task 2 (mapper.py + test_mapper_kind_of) — disjoint files ✅. Waves 2–7 each touch `detect.py` and are separate sequential waves (no two same-file tasks share a wave) ✅. Dependencies: Task 4 uses Task 1's emit (later wave ✅); Task 5 uses Task 4 (✅); Task 6/7 use Task 5 (✅); Task 7 uses Task 2's `kind_of` (Wave 1 < Wave 6 ✅); Task 8 uses Task 6/7 (✅). No violations.

---

**Plan complete and saved to `docs/plans/2026-06-18-detect-orchestrator.md`. Proceeding with Subagent-Driven execution.**