#!/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"))
verdict = _load("sg_verdict", os.path.join(_ROOT, "prevent", "verdict.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


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}")


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


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


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"]]
        # Verdict owns the credit/miss/punt decision (prevent/verdict.py). CREDIT = right-reason flag (a hit);
        # MISS = autoscorable but not flagged (a REAL recall-0); PUNT = not autoscorable -> hand-judge.
        exp = verdict.Expectation(symbol=c.get("canonical_symbol") or "", why=c.get("why") or "",
                                  polarity=verdict.POSITIVE, autoscore_src=_read_text(vuln))
        v = verdict.match(titles, exp)
        if v.outcome == verdict.CREDIT:
            findings_out[cid] = titles
            scored.append(cid)
            hits.append(cid)
        elif v.outcome == verdict.MISS:
            findings_out[cid] = titles
            scored.append(cid)
        else:  # PUNT -> hand-judge (autoscore-unscorable)
            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


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())