27Bslash6 · July 22, 2026 02:23
diff --git a/lab251_pin_ledger.py b/lab251_pin_ledger.py
 """LAB-251 — cheap per-study pin-landing ledger for all 67 seeded studies.

 For each of the 67 registry studies: run the REAL curated seed + REAL selection path against an
 EMPTY organic pool (the strictest case — the pin must stand alone), and record whether it lands
 technique-direct, or is excluded with a stated reason (fetch_failed / retracted / citation_mismatch).

 This answers AC1 (how many reach the page) and AC2 (each present or excluded-with-reason) for all 67
 WITHOUT the throttle-bound organic-pool graph walk: for a vocab-gap technique the organic pool is
 ~entirely GRADE-indirect and dropped by the topical-relevance gate before the cap (verified on
 TECH_023: 552/553 dropped), so the pin lands alone regardless of pool. The empty-pool run is that
 worst case made explicit. Fetch calls (~2/study) still hit SS/OpenAlex, so re-runs are resume-safe.

 Run:  uv run python lab251_pin_ledger.py
 Output: pin_ledger.json
 """

 from __future__ import annotations

 import json
 import logging
 import signal
 import time
 from pathlib import Path


 class _Timeout(Exception):
    pass


 def _deadline(seconds: int):
    """Hard wall-clock cap on a throttle-prone fetch: SIGALRM aborts the 300s transient-retry block so
    a rate-limited study is deferred (retry-pending next wave) instead of stalling the whole run."""

    def _raise(signum, frame):
        raise _Timeout()

    class _Ctx:
        def __enter__(self):
            self.old = signal.signal(signal.SIGALRM, _raise)
            signal.alarm(seconds)

        def __exit__(self, *a):
            signal.alarm(0)
            signal.signal(signal.SIGALRM, self.old)

    return _Ctx()

 HERE = Path(__file__).resolve().parent
 logging.basicConfig(level=logging.ERROR)

 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import (  # noqa: E402
    load_curated_refs,
    pin_record_key,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.evidence_acquisition import (  # noqa: E402
    _STAGED_MAX_SELECTED_REFERENCES,
    _diversity_select_enabled,
    _merged_evidence_to_selected_reference,
    _packet_scope_aliases,
    _partition_admissible_records,
    _partition_topically_relevant_records,
    _quality_first_reorder,
    _select_with_quota,
    _source_context_terms,
    _topical_relevance_gate_enabled,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.input_packet_builder import (  # noqa: E402
    build_packet_for_tech_id,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import (  # noqa: E402
    apply_curated_references,
 )


 def select_from(records, scope_ctx, pins):
    reordered = _quality_first_reorder(records, scope_ctx=scope_ctx, pinned_keys=pins)
    admitted, _ = _partition_admissible_records(reordered)
    if _topical_relevance_gate_enabled():
        admitted, _ = _partition_topically_relevant_records(admitted, scope_ctx=scope_ctx, pinned_keys=pins)
    if _diversity_select_enabled():
        capped, _ = _select_with_quota(admitted)
    else:
        capped = admitted[: _STAGED_MAX_SELECTED_REFERENCES]
    refs = [
        _merged_evidence_to_selected_reference(
            rec,
            rank=i,
            technique=scope_ctx["technique"],
            aliases=scope_ctx["aliases"],
            modality=scope_ctx["modality"],
            mechanism_terms=scope_ctx["mechanism_terms"],
            category=scope_ctx["category"],
            is_pinned=pin_record_key(rec) in pins,
        )
        for i, rec in enumerate(capped, start=1)
    ]
    return capped, refs


 def main() -> None:
    import csv

    repo_data = (
        HERE.parent
        / "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data/missing-literature.csv"
    )
    reg_rows = list(csv.DictReader(open(repo_data, encoding="utf-8")))
    by_tech: dict[str, list[dict]] = {}
    for r in reg_rows:
        by_tech.setdefault(r["tech_id"], []).append(r)

    dest = HERE / "pin_ledger.json"
    ledger: dict = json.loads(dest.read_text()) if dest.is_file() else {}

    for tech_id in sorted(by_tech):
        if tech_id in ledger:
            print(f"{tech_id}: cached", flush=True)
            continue
        try:
            packet = build_packet_for_tech_id(tech_id)
            technique = packet.get("technique") or packet.get("display_label") or tech_id
            scope_ctx = {
                "technique": technique,
                "aliases": _packet_scope_aliases(packet),
                "modality": str(packet.get("modality") or ""),
                "mechanism_terms": _source_context_terms(packet).get("mechanism", []),
                "category": str(packet.get("category") or ""),
            }
            curated = load_curated_refs(tech_id)
            prov: dict = {}
            with _deadline(25):
                seeded = apply_curated_references([], curated, live=True, tech_id=tech_id, provenance=prov)
            pins = frozenset(prov.get("injected") or ()) | frozenset(prov.get("matched_present") or ())
            capped, refs = select_from(seeded, scope_ctx, pins)
            on_page = {pin_record_key(r) for r in capped} & pins
            scope_by_key = {}
            for i, rec in enumerate(capped):
                k = pin_record_key(rec)
                if k in pins:
                    scope_by_key[k] = refs[i].get("evidence_scope")
            ledger[tech_id] = {
                "technique": technique,
                "n_pins": len(curated.get("pinned") or []),
                "injected": prov.get("injected") or [],
                "fetch_failed": prov.get("fetch_failed") or [],
                "retracted_refused": prov.get("retracted_refused") or [],
                "citation_mismatch": prov.get("citation_mismatch") or [],
                "invalid": prov.get("invalid") or [],
                "on_page": sorted(on_page),
                "capped_out": sorted(pins - on_page),
                "scope_by_key": scope_by_key,
                "pins": prov.get("pins") or {},
            }
            e = ledger[tech_id]
            print(
                f"{tech_id}: pins={e['n_pins']} on_page={len(e['on_page'])} "
                f"tech_direct={sum(1 for v in scope_by_key.values() if v == 'technique')} "
                f"fetch_failed={len(e['fetch_failed'])} mismatch={len(e['citation_mismatch'])} "
                f"retracted={len(e['retracted_refused'])}",
                flush=True,
            )
        except _Timeout:
            # Throttled this wave — do NOT cache; a later wave retries once the per-IP quota recovers.
            print(f"{tech_id}: THROTTLED (deferred)", flush=True)
            dest.write_text(json.dumps(ledger, indent=2))
            time.sleep(3.0)
            continue
        except Exception as exc:
            ledger[tech_id] = {"error": f"{type(exc).__name__}: {exc}"}
            print(f"{tech_id}: ERROR {type(exc).__name__}: {exc}", flush=True)
        dest.write_text(json.dumps(ledger, indent=2))
        time.sleep(1.5)  # pace targeted SS/OpenAlex lookups under the anonymous per-IP rate limit


 if __name__ == "__main__":
    main()
diff --git a/lab251_report.md b/lab251_report.md
diff --git a/lab251_report.py b/lab251_report.py
 """LAB-251 — final report from the KEYED full-pool run (production .env: all providers + cachekit
 edge + Lever-3 Haiku design pass; perplexity_dr 401 = same 6/7 state as the original #315 audit).

 Every page: live organic acquisition ONCE, then the production selection path twice from that
 identical pool — BEFORE (no seeding) and AFTER (curated seeding, authoritative pins).
 """

 from __future__ import annotations

 import csv
 import json
 import re
 from pathlib import Path

 HERE = Path(__file__).resolve().parent
 OUT = HERE / "out"
 REPO_DATA = HERE.parent / "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data"


 def registry_key(raw: str) -> str | None:
    raw = (raw or "").strip().strip("`")
    if not raw:
        return None
    if raw.upper().startswith("PMID"):
        m = re.search(r"\d+", raw.split("/")[0])
        return f"pmid:{m.group()}" if m else None
    doi = raw.split(" ")[0] if raw.startswith("10.") else raw
    return doi.lower() if doi.startswith("10.") else None


 def load_buckets() -> dict[tuple[str, str], str]:
    text = (HERE / "geo315_report.md").read_text(encoding="utf-8")
    out: dict[tuple[str, str], str] = {}

    def rows(a: str, b: str) -> list[list[str]]:
        blk = text.split(a, 1)[1].split(b, 1)[0]
        return [[c.strip() for c in ln.strip("|").split("|")] for ln in blk.splitlines() if ln.startswith("| TECH_")]

    for c in rows("## found_but_dropped", "## acquisition_miss"):
        if (k := registry_key(c[2])):
            out[(c[0], k)] = "found_but_dropped"
    for c in rows("## acquisition_miss", "## on_page"):
        if (k := registry_key(c[2])):
            out[(c[0], k)] = c[3]
    for c in rows("## on_page", "## Caveats"):
        if (k := registry_key(c[1])):
            out[(c[0], k)] = "on_page_temporal"
    return out


 def main() -> None:
    buckets = load_buckets()
    reg_rows = list(csv.DictReader(open(REPO_DATA / "missing-literature.csv", encoding="utf-8")))
    full = {p.stem: json.loads(p.read_text()) for p in sorted(OUT.glob("TECH_*.json"))}

    def pct(counts: dict, scope: str) -> float:
        t = sum(counts.values())
        return 100.0 * counts.get(scope, 0) / t if t else 0.0

    page_lines = [
        "| tech | technique | pool | reg on-page before→after | selected before→after | tech-direct% b→a | pins inj/matched/failed | capped_out |",
        "|---|---|---:|---|---|---|---|---:|",
    ]
    regressions: list[str] = []
    tot_before = tot_after = tot_reg = 0
    tot_inj = tot_mat = tot_ff = tot_cap = 0
    scope_tot = {"before": {"technique": 0, "n": 0}, "after": {"technique": 0, "n": 0}}
    empty_before_pages = []
    for tid, d in full.items():
        b, a, sp = d["before"], d["after"], d["after"]["seed_prov"]
        tot_before += len(b["registry_on_page"])
        tot_after += len(a["registry_on_page"])
        tot_reg += len(d["registry_keys"])
        tot_inj += len(sp.get("injected") or [])
        tot_mat += len(sp.get("matched_present") or [])
        tot_ff += len(sp.get("fetch_failed") or [])
        tot_cap += len(a["capped_out"])
        for lane_name, lane in (("before", b), ("after", a)):
            scope_tot[lane_name]["technique"] += lane["scope_counts"].get("technique", 0)
            scope_tot[lane_name]["n"] += sum(lane["scope_counts"].values())
        if b["n_selected"] == 0:
            empty_before_pages.append(tid)
        lost = set(b["registry_on_page"]) - set(a["registry_on_page"])
        if lost or len(a["registry_on_page"]) < len(b["registry_on_page"]):
            regressions.append(f"{tid}: {sorted(lost)}")
        page_lines.append(
            f"| {tid} | {d['technique'][:24]} | {d['pool_size']}"
            f" | {len(b['registry_on_page'])}→{len(a['registry_on_page'])} of {len(d['registry_keys'])}"
            f" | {b['n_selected']}→{a['n_selected']}"
            f" | {pct(b['scope_counts'], 'technique'):.0f}%→{pct(a['scope_counts'], 'technique'):.0f}%"
            f" | {len(sp.get('injected') or [])}/{len(sp.get('matched_present') or [])}/{len(sp.get('fetch_failed') or [])}"
            f" | {len(a['capped_out'])} |"
        )

    led_lines = ["| tech | study | #315 bucket | outcome | scope |", "|---|---|---|---|---|"]
    on_page = tech_direct = excluded = 0
    vocab_total = vocab_on = 0
    for row in reg_rows:
        tid = row["tech_id"]
        key = registry_key(row["doi"]) or f"?:{row['doi'][:30]}"
        bucket = buckets.get((tid, key), "?")
        d = full.get(tid)
        outcome, scope = "PAGE NOT RUN", "—"
        if d:
            b, a, sp = d["before"], d["after"], d["after"]["seed_prov"]
            pin_scopes = {e["key"]: e["scope"] for e in a.get("selected_pinned_ranks", [])}
            if key in a["registry_on_page"]:
                on_page += 1
                scope = pin_scopes.get(key, "?")
                if scope == "technique":
                    tech_direct += 1
                lane = (
                    "injected"
                    if key in (sp.get("injected") or [])
                    else "matched-organic (pin authority)"
                    if key in (sp.get("matched_present") or [])
                    else "organic"
                )
                pre = "was already on-page before seeding; " if key in b["registry_on_page"] else ""
                outcome = f"ON PAGE ({pre}{lane})"
            elif key in (sp.get("fetch_failed") or []):
                excluded += 1
                outcome = "EXCLUDED: fetch_failed"
            elif key in (sp.get("citation_mismatch") or []):
                excluded += 1
                outcome = "EXCLUDED: citation mismatch (wrong-paper guard)"
            elif key in (sp.get("retracted_refused") or []):
                excluded += 1
                outcome = "EXCLUDED: retracted"
            elif key in a["capped_out"]:
                outcome = "CAPPED OUT (regression!)"
            else:
                outcome = "NOT MATCHED (check)"
        if bucket == "query_fixable_title_only":
            vocab_total += 1
            if outcome.startswith("ON PAGE"):
                vocab_on += 1
        led_lines.append(f"| {tid} | `{key}` | {bucket} | {outcome} | {scope} |")

    lines = [
        "# LAB-251 — live recall validation (KEYED production run): 67 seeded studies across 56 pages",
        "",
        "Config: production `.env` (all provider keys + cachekit edge + Lever-3 Haiku design pass;",
        "`perplexity_dr` 401-expired = same 6/7-provider state as the original #315 audit and the",
        "2026-06-30 batch). Per page: one live organic staged acquisition, then the production",
        "selection path twice from the identical pool — before (no seeding) vs after (curated seeding).",
        "",
        "## Headline",
        "",
        f"- **Pages: {len(full)}/56 run, 0 failures.** Registry studies on-page: **{tot_before}→{tot_after} of {tot_reg}**.",
        f"- Pin lanes: {tot_inj} injected + {tot_mat} matched-organic (pin authority); {tot_ff} fetch-failed; "
        f"**{tot_cap} capped out**; {excluded} excluded.",
        f"- **{tech_direct}/{tot_after} on-page studies classify technique-direct.**",
        f"- **Vocabulary-gap studies on-page (AC target ~45/47): {vocab_on}/{vocab_total}.**",
        f"- Aggregate technique-direct share of selected refs: "
        f"{100 * scope_tot['before']['technique'] / max(scope_tot['before']['n'], 1):.1f}%→"
        f"{100 * scope_tot['after']['technique'] / max(scope_tot['after']['n'], 1):.1f}%.",
        f"- **Regressions: {len(regressions)}**" + ("" if not regressions else " — " + "; ".join(regressions)),
        f"- Pages whose BEFORE selection is EMPTY (would EvidenceFloorError/hold without seeding): "
        f"{len(empty_before_pages)} — {', '.join(empty_before_pages) if empty_before_pages else '-'}",
        "",
        "## Per-page recall (before→after, same live pool per page)",
        "",
        *page_lines,
        "",
        "## 67-study ledger",
        "",
        *led_lines,
    ]
    (HERE / "lab251_report.md").write_text("\n".join(lines), encoding="utf-8")
    print("\n".join(lines[: lines.index("## Per-page recall (before→after, same live pool per page)")]))


 if __name__ == "__main__":
    main()
diff --git a/lab251_validate.py b/lab251_validate.py
 """LAB-251 — live recall validation of the curated-seeding path (67 pins / 56 pages).

 Per technique: acquire the organic pool ONCE (live, keyless Stage-1 providers), then run the
 production selection path twice from that identical pool — BEFORE (no seeding = the 6cc50bdc
 baseline behaviour on current spine) and AFTER (curated seeding, authoritative pins). Selection
 calls the SAME node functions in the SAME order as run_staged_acquisition; the only node step
 skipped is the live Haiku design pre-pass (GEO_ABSTRACT_DESIGN_CLASSIFY — needs an LLM key;
 same omission as the original geo315 audit, caveat 3).

 Run:  uv run --project <repo> python lab251_validate.py [--tech TECH_023] [--all]
 Outputs: out/{tech_id}.json (resume-safe: existing files are skipped), out/_failures.json
 """

 from __future__ import annotations

 import argparse
 import json
 import logging
 import os
 import time
 import traceback
 from pathlib import Path

 HERE = Path(__file__).resolve().parent
 OUT = HERE / "out"
 OUT.mkdir(exist_ok=True)

 logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")

 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import (  # noqa: E402
    apply_curated_references,
    pin_record_key,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.evidence_acquisition import (  # noqa: E402
    _STAGED_MAX_SELECTED_REFERENCES,
    _abstract_design_enabled,
    _classify_designs_via_llm,
    _diversity_select_enabled,
    _extract_outcome_dimension,
    _extract_outcome_dimensions,
    _merged_evidence_to_selected_reference,
    _packet_scope_aliases,
    _partition_admissible_records,
    _partition_topically_relevant_records,
    _quality_first_reorder,
    _select_with_quota,
    _source_context_terms,
    _topical_relevance_gate_enabled,
    load_input_packet,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.dedupe import (  # noqa: E402
    apply_cochrane_design_override,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.identity_norm import (  # noqa: E402
    normalize_doi,
    normalize_pmid,
 )
 from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.staged_orchestrator import (  # noqa: E402
    acquire_evidence_staged_with_taxonomy,
 )


 def registry_key(raw: str) -> str | None:
    """Identity key for a missing-literature.csv doi cell: normalized DOI, else pmid:{digits}.

    Mirrors curated_refs._key_for_value ordering (PMID first) but tolerates the CSV's
    annotated cells ("PMID 30936803 / PMCID PMC6438091", "10.1136/bmj.a884 (PMID ...)").
    """
    raw = (raw or "").strip()
    if not raw:
        return None
    if raw.upper().startswith("PMID"):
        pmid = normalize_pmid(raw.split("/")[0])
        return f"pmid:{pmid}" if pmid else None
    return normalize_doi(raw.split(" ")[0] if raw.startswith("10.") else raw)


 def record_keys(rec) -> set[str]:
    keys = set()
    if nd := normalize_doi(getattr(rec, "doi", None)):
        keys.add(nd)
    if pmid := normalize_pmid(getattr(rec, "pmid", None)):
        keys.add(f"pmid:{pmid}")
    return keys


 def run_lane(pool, curated, *, tech_id, technique, scope_ctx, registry_keys_for_tech):
    prov: dict = {}
    records = apply_curated_references(list(pool), curated, live=True, tech_id=tech_id, provenance=prov)
    pins = frozenset(prov.get("injected") or ()) | frozenset(prov.get("matched_present") or ())
    # Lever 3 (node parity, seed → LLM classify → cochrane): abstract-aware Haiku design pass on the
    # coarse residual. Skips any record already stamped (curated-pin assertions protected) and is
    # edge-cached, so the second lane re-reads identical verdicts; in-place stamps on shared pool
    # objects keep both lanes consistent even cache-cold. Soft-fails internally; requires live keys.
    if _abstract_design_enabled():
        _classify_designs_via_llm(records)
    apply_cochrane_design_override(records)
    reordered = _quality_first_reorder(records, scope_ctx=scope_ctx, pinned_keys=pins)
    admitted, gate_rejected = _partition_admissible_records(reordered)
    relevance_rejected: list = []
    if _topical_relevance_gate_enabled():
        admitted, relevance_rejected = _partition_topically_relevant_records(admitted, scope_ctx=scope_ctx, pinned_keys=pins)
    if _diversity_select_enabled():
        capped, _ = _select_with_quota(admitted)
    else:
        capped = admitted[: _STAGED_MAX_SELECTED_REFERENCES]

    selected = [
        _merged_evidence_to_selected_reference(
            rec,
            rank=i,
            technique=technique,
            aliases=scope_ctx["aliases"],
            modality=scope_ctx["modality"],
            mechanism_terms=scope_ctx["mechanism_terms"],
            category=scope_ctx["category"],
            is_pinned=pin_record_key(rec) in pins,
        )
        for i, rec in enumerate(capped, start=1)
    ]
    capped_keys = {k for r in capped for k in record_keys(r)}
    scope_counts: dict[str, int] = {}
    for ref in selected:
        scope = ref.get("evidence_scope") or "indirect"
        scope_counts[scope] = scope_counts.get(scope, 0) + 1
    pool_keys = {k for r in records for k in record_keys(r)}
    return {
        "n_selected": len(selected),
        "scope_counts": scope_counts,
        "registry_on_page": sorted(k for k in registry_keys_for_tech if k in capped_keys),
        "registry_in_pool": sorted(k for k in registry_keys_for_tech if k in pool_keys),
        "capped_out": sorted(pins - {k for r in capped if (k := pin_record_key(r))}),
        "gate_rejected": len(gate_rejected),
        "relevance_rejected": len(relevance_rejected),
        "seed_prov": {
            k: prov.get(k)
            for k in (
                "injected",
                "matched_present",
                "fetch_failed",
                "retracted_refused",
                "citation_mismatch",
                "drop_conflicts",
                "invalid",
                "asserted",
                "pins",
            )
            if prov.get(k)
        },
        "selected_pinned_ranks": [
            {"rank": i, "key": pin_record_key(rec), "scope": selected[i - 1].get("evidence_scope")}
            for i, rec in enumerate(capped, start=1)
            if pin_record_key(rec) in pins
        ],
    }


 def run_tech(tech_id: str, registry_keys_for_tech: set[str]) -> dict:
    t0 = time.time()
    outdir = OUT / "runs" / tech_id
    outdir.mkdir(parents=True, exist_ok=True)
    state: dict = {"evidence_mode": "fresh", "tech_id": tech_id, "live": True, "output_dir": str(outdir)}
    state = load_input_packet(state)
    packet = state["input_packet"]
    technique = packet.get("technique") or packet.get("display_label") or tech_id
    outcome = _extract_outcome_dimension(state)
    extras: list[str] = []
    if os.environ.get("GEO_MULTI_OUTCOME_ACQUISITION", "1") != "0":
        extras = _extract_outcome_dimensions(state)[1:]

    # All 7 providers enabled — the audit's default config. The 3 keyless ones (consensus/openai/
    # perplexity) contribute 0 records in this runtime, so the organic pool is production-minus-those-
    # keys; keeping them enabled ALSO paces the SS/OpenAlex index calls (consensus's transient retry
    # delay spaces the burst) which, without an S2 API key, is what keeps them under the anonymous
    # 429 rate limit. Disabling them sped each tech up but throttled the real index providers into
    # 120s timeouts — net slower and a degraded pool. Slow-and-paced is both faithful and reliable.
    result = acquire_evidence_staged_with_taxonomy(tech_id, outcome, extra_outcome_dimensions=extras)
    pool = result.records

    scope_ctx = {
        "technique": technique,
        "aliases": _packet_scope_aliases(packet),
        "modality": str(packet.get("modality") or ""),
        "mechanism_terms": _source_context_terms(packet).get("mechanism", []),
        "category": str(packet.get("category") or ""),
    }

    provider_status = {
        name: {"status": ps.status, "records": ps.record_count} for name, ps in (result.provider_status or {}).items()
    }

    before = run_lane(
        pool,
        {"pinned": [], "dropped": []},
        tech_id=tech_id,
        technique=technique,
        scope_ctx=scope_ctx,
        registry_keys_for_tech=registry_keys_for_tech,
    )
    after = run_lane(
        pool,
        packet.get("curated_references") or {},
        tech_id=tech_id,
        technique=technique,
        scope_ctx=scope_ctx,
        registry_keys_for_tech=registry_keys_for_tech,
    )
    return {
        "tech_id": tech_id,
        "technique": technique,
        "outcome_dimension": outcome,
        "extra_outcome_dimensions": extras,
        "pool_size": len(pool),
        "provider_status": provider_status,
        "registry_keys": sorted(registry_keys_for_tech),
        "before": before,
        "after": after,
        "elapsed_s": round(time.time() - t0, 1),
    }


 def load_registry() -> dict[str, set[str]]:
    import csv

    repo_data = (
        Path(__file__).resolve().parents[1]
        / "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data/missing-literature.csv"
    )
    reg: dict[str, set[str]] = {}
    for row in csv.DictReader(open(repo_data, encoding="utf-8")):
        key = registry_key(row["doi"])
        if key:
            reg.setdefault(row["tech_id"], set()).add(key)
    return reg


 def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--tech", action="append", default=None)
    ap.add_argument("--all", action="store_true")
    ap.add_argument("--shard", default=None, help="i/n — process techs where index %% n == i (parallel workers)")
    args = ap.parse_args()

    registry = load_registry()
    techs = args.tech or (sorted(registry) if args.all else [])
    if not techs:
        ap.error("pass --tech TECH_xxx or --all")
    if args.shard:
        i, n = (int(x) for x in args.shard.split("/"))
        techs = [t for idx, t in enumerate(techs) if idx % n == i]

    failures = {}
    fail_path = OUT / "_failures.json"
    if fail_path.is_file():
        failures = json.loads(fail_path.read_text())
    for tech_id in techs:
        dest = OUT / f"{tech_id}.json"
        if dest.is_file():
            print(f"{tech_id}: exists, skipped", flush=True)
            continue
        try:
            res = run_tech(tech_id, registry.get(tech_id, set()))
        except Exception as exc:  # record and continue — a one-tech failure must not kill the sweep
            failures[tech_id] = {"error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc()[-2000:]}
            fail_path.write_text(json.dumps(failures, indent=2))
            print(f"{tech_id}: FAILED {type(exc).__name__}: {exc}", flush=True)
            continue
        failures.pop(tech_id, None)
        fail_path.write_text(json.dumps(failures, indent=2))
        dest.write_text(json.dumps(res, indent=2))
        b, a = res["before"], res["after"]
        print(
            f"{tech_id}: pool={res['pool_size']} on_page {len(b['registry_on_page'])}->{len(a['registry_on_page'])}"
            f" of {len(res['registry_keys'])} | injected={len(a['seed_prov'].get('injected') or [])}"
            f" matched={len(a['seed_prov'].get('matched_present') or [])}"
            f" fetch_failed={len(a['seed_prov'].get('fetch_failed') or [])}"
            f" capped_out={len(a['capped_out'])} | {res['elapsed_s']}s",
            flush=True,
        )


 if __name__ == "__main__":
    main()
	"""LAB-251 — cheap per-study pin-landing ledger for all 67 seeded studies.

	For each of the 67 registry studies: run the REAL curated seed + REAL selection path against an
	EMPTY organic pool (the strictest case — the pin must stand alone), and record whether it lands
	technique-direct, or is excluded with a stated reason (fetch_failed / retracted / citation_mismatch).

	This answers AC1 (how many reach the page) and AC2 (each present or excluded-with-reason) for all 67
	WITHOUT the throttle-bound organic-pool graph walk: for a vocab-gap technique the organic pool is
	~entirely GRADE-indirect and dropped by the topical-relevance gate before the cap (verified on
	TECH_023: 552/553 dropped), so the pin lands alone regardless of pool. The empty-pool run is that
	worst case made explicit. Fetch calls (~2/study) still hit SS/OpenAlex, so re-runs are resume-safe.

	Run: uv run python lab251_pin_ledger.py
	Output: pin_ledger.json
	"""

	from __future__ import annotations

	import json
	import logging
	import signal
	import time
	from pathlib import Path


	class _Timeout(Exception):
	pass


	def _deadline(seconds: int):
	"""Hard wall-clock cap on a throttle-prone fetch: SIGALRM aborts the 300s transient-retry block so
	a rate-limited study is deferred (retry-pending next wave) instead of stalling the whole run."""

	def _raise(signum, frame):
	raise _Timeout()

	class _Ctx:
	def __enter__(self):
	self.old = signal.signal(signal.SIGALRM, _raise)
	signal.alarm(seconds)

	def __exit__(self, *a):
	signal.alarm(0)
	signal.signal(signal.SIGALRM, self.old)

	return _Ctx()

	HERE = Path(__file__).resolve().parent
	logging.basicConfig(level=logging.ERROR)

	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import ( # noqa: E402
	load_curated_refs,
	pin_record_key,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.evidence_acquisition import ( # noqa: E402
	_STAGED_MAX_SELECTED_REFERENCES,
	_diversity_select_enabled,
	_merged_evidence_to_selected_reference,
	_packet_scope_aliases,
	_partition_admissible_records,
	_partition_topically_relevant_records,
	_quality_first_reorder,
	_select_with_quota,
	_source_context_terms,
	_topical_relevance_gate_enabled,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.input_packet_builder import ( # noqa: E402
	build_packet_for_tech_id,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import ( # noqa: E402
	apply_curated_references,
	)


	def select_from(records, scope_ctx, pins):
	reordered = _quality_first_reorder(records, scope_ctx=scope_ctx, pinned_keys=pins)
	admitted, _ = _partition_admissible_records(reordered)
	if _topical_relevance_gate_enabled():
	admitted, _ = _partition_topically_relevant_records(admitted, scope_ctx=scope_ctx, pinned_keys=pins)
	if _diversity_select_enabled():
	capped, _ = _select_with_quota(admitted)
	else:
	capped = admitted[: _STAGED_MAX_SELECTED_REFERENCES]
	refs = [
	_merged_evidence_to_selected_reference(
	rec,
	rank=i,
	technique=scope_ctx["technique"],
	aliases=scope_ctx["aliases"],
	modality=scope_ctx["modality"],
	mechanism_terms=scope_ctx["mechanism_terms"],
	category=scope_ctx["category"],
	is_pinned=pin_record_key(rec) in pins,
	)
	for i, rec in enumerate(capped, start=1)
	]
	return capped, refs


	def main() -> None:
	import csv

	repo_data = (
	HERE.parent
	/ "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data/missing-literature.csv"
	)
	reg_rows = list(csv.DictReader(open(repo_data, encoding="utf-8")))
	by_tech: dict[str, list[dict]] = {}
	for r in reg_rows:
	by_tech.setdefault(r["tech_id"], []).append(r)

	dest = HERE / "pin_ledger.json"
	ledger: dict = json.loads(dest.read_text()) if dest.is_file() else {}

	for tech_id in sorted(by_tech):
	if tech_id in ledger:
	print(f"{tech_id}: cached", flush=True)
	continue
	try:
	packet = build_packet_for_tech_id(tech_id)
	technique = packet.get("technique") or packet.get("display_label") or tech_id
	scope_ctx = {
	"technique": technique,
	"aliases": _packet_scope_aliases(packet),
	"modality": str(packet.get("modality") or ""),
	"mechanism_terms": _source_context_terms(packet).get("mechanism", []),
	"category": str(packet.get("category") or ""),
	}
	curated = load_curated_refs(tech_id)
	prov: dict = {}
	with _deadline(25):
	seeded = apply_curated_references([], curated, live=True, tech_id=tech_id, provenance=prov)
	pins = frozenset(prov.get("injected") or ()) \| frozenset(prov.get("matched_present") or ())
	capped, refs = select_from(seeded, scope_ctx, pins)
	on_page = {pin_record_key(r) for r in capped} & pins
	scope_by_key = {}
	for i, rec in enumerate(capped):
	k = pin_record_key(rec)
	if k in pins:
	scope_by_key[k] = refs[i].get("evidence_scope")
	ledger[tech_id] = {
	"technique": technique,
	"n_pins": len(curated.get("pinned") or []),
	"injected": prov.get("injected") or [],
	"fetch_failed": prov.get("fetch_failed") or [],
	"retracted_refused": prov.get("retracted_refused") or [],
	"citation_mismatch": prov.get("citation_mismatch") or [],
	"invalid": prov.get("invalid") or [],
	"on_page": sorted(on_page),
	"capped_out": sorted(pins - on_page),
	"scope_by_key": scope_by_key,
	"pins": prov.get("pins") or {},
	}
	e = ledger[tech_id]
	print(
	f"{tech_id}: pins={e['n_pins']} on_page={len(e['on_page'])} "
	f"tech_direct={sum(1 for v in scope_by_key.values() if v == 'technique')} "
	f"fetch_failed={len(e['fetch_failed'])} mismatch={len(e['citation_mismatch'])} "
	f"retracted={len(e['retracted_refused'])}",
	flush=True,
	)
	except _Timeout:
	# Throttled this wave — do NOT cache; a later wave retries once the per-IP quota recovers.
	print(f"{tech_id}: THROTTLED (deferred)", flush=True)
	dest.write_text(json.dumps(ledger, indent=2))
	time.sleep(3.0)
	continue
	except Exception as exc:
	ledger[tech_id] = {"error": f"{type(exc).__name__}: {exc}"}
	print(f"{tech_id}: ERROR {type(exc).__name__}: {exc}", flush=True)
	dest.write_text(json.dumps(ledger, indent=2))
	time.sleep(1.5) # pace targeted SS/OpenAlex lookups under the anonymous per-IP rate limit


	if __name__ == "__main__":
	main()
tech	technique	pool	reg on-page before→after	selected before→after	tech-direct% b→a	pins inj/matched/failed
TECH_023	Jhana Absorption	553	0→1 of 1	0→1	0%→100%	1/0/0
TECH_035	Counting Meditation	36	0→1 of 1	4→5	100%→100%	1/0/0
TECH_049	Trataka (Candle Gazing)	201	0→1 of 1	0→1	0%→100%	1/0/0
TECH_051	Karuna (Compassion)	271	0→1 of 1	1→2	0%→50%	0/1/0
TECH_075	Equanimity Practice	499	0→1 of 1	2→3	0%→33%	1/0/0
TECH_079	Noting Practice	598	0→1 of 1	0→1	0%→100%	1/0/0
TECH_107	4-7-8 Breathing	43	0→1 of 1	1→2	0%→50%	1/0/0
TECH_118	Physiological Sigh	689	0→1 of 1	8→9	0%→11%	0/1/0
TECH_127	Buteyko Breathing	244	1→1 of 1	50→50	86%→86%	0/1/0
TECH_147	Sitali / Sitkari	249	0→1 of 1	1→2	0%→50%	1/0/0
TECH_157	Breath Holds	371	0→1 of 1	26→27	100%→100%	1/0/0
TECH_162	Bhastrika	150	0→1 of 1	29→30	34%→37%	1/0/0
TECH_163	Bhramari (Bee Breath)	537	0→1 of 1	7→8	0%→12%	1/0/0
TECH_180	Six Healing Sounds	12	0→1 of 1	1→2	100%→100%	1/0/0
TECH_192	Tummo Breathing	615	0→2 of 2	3→5	0%→40%	2/0/0
TECH_207	Implementation Intention	390	0→1 of 1	1→2	100%→100%	0/1/0
TECH_210	Anti-Procrastination	75	0→1 of 1	3→4	100%→100%	1/0/0
TECH_240	The Work	512	1→1 of 1	15→15	80%→80%	0/1/0
TECH_252	Forgiveness Practice	447	0→1 of 1	1→2	100%→100%	1/0/0
TECH_256	Life Review Meditation	490	0→1 of 1	2→3	0%→33%	1/0/0
TECH_257	Naikan Reflection	123	0→1 of 1	0→1	0%→100%	1/0/0
TECH_274	Rosary Meditation	456	0→1 of 1	0→1	0%→100%	1/0/0
TECH_283	Japa Meditation	465	0→2 of 2	1→3	0%→67%	2/0/0
TECH_285	Simran (Naam Japna)	480	0→1 of 1	1→2	0%→50%	0/1/0
TECH_319	Jin Shin Jyutsu Self-Hel	589	0→1 of 1	1→2	0%→50%	1/0/0
TECH_338	Tummo Inner Fire	362	0→2 of 2	0→2	0%→100%	2/0/0
TECH_345	Technology Fast	170	0→1 of 1	2→3	100%→100%	1/0/0
TECH_356	Tea Meditation	54	0→1 of 1	4→5	100%→100%	1/0/0
TECH_359	Mindful Dishwashing	601	0→1 of 1	1→2	0%→50%	1/0/0
TECH_417	Bikram/Hot Yoga	487	0→1 of 1	50→50	46%→48%	1/0/0
TECH_421	Iyengar Yoga	299	0→1 of 1	50→50	24%→26%	1/0/0
TECH_431	Cognitive Defusion Pract	683	0→2 of 2	7→9	0%→22%	1/1/0
TECH_451	Emotional Labeling (Affe	400	0→1 of 1	4→5	0%→20%	1/0/0
TECH_460	Self-Compassion Break	101	0→1 of 1	50→50	100%→100%	1/0/0
TECH_476	Forgiveness Meditation	36	0→1 of 1	1→2	100%→100%	1/0/0
TECH_477	Gratitude Meditation	150	0→2 of 2	1→3	100%→100%	2/0/0
TECH_479	Savoring Practice	544	0→1 of 1	1→2	0%→50%	1/0/0
TECH_488	Sensorimotor Awareness P	532	0→1 of 1	0→1	0%→100%	1/0/0
TECH_492	Family Constellation Cir	641	0→1 of 1	1→2	0%→50%	1/0/0
TECH_508	Imago Dialogue Practice	320	0→1 of 1	0→1	0%→100%	1/0/0
TECH_512	Eye-Gazing	223	0→1 of 1	1→2	100%→100%	1/0/0
TECH_520	Autogenic Training	412	1→1 of 1	50→50	96%→96%	0/1/0
TECH_550	Heart Coherence Practice	540	0→1 of 1	0→1	0%→100%	1/0/0
TECH_555	Heartfulness Practice	19	0→1 of 1	1→2	100%→100%	1/0/0
TECH_578	Alexander Technique Awar	448	0→2 of 2	0→2	0%→100%	2/0/0
TECH_579	Feldenkrais Awareness Th	108	0→1 of 1	5→6	100%→100%	1/0/0
TECH_627	Tense-and-Release	640	0→2 of 2	0→2	0%→100%	2/0/0
TECH_633	Pendulation Practice	486	0→2 of 2	0→2	0%→100%	2/0/0
TECH_636	Titration Practice	315	0→2 of 2	1→3	100%→100%	2/0/0
TECH_655	Gregorian Chant	66	0→1 of 1	10→11	100%→100%	1/0/0
TECH_676	ASMR	163	0→1 of 1	50→50	100%→100%	1/0/0
TECH_677	Nature Sound Immersion	393	0→1 of 1	23→24	91%→92%	1/0/0
TECH_703	Humming Meditation	479	0→1 of 1	2→3	50%→67%	1/0/0
TECH_725	Success Rehearsal	632	0→2 of 2	0→2	0%→100%	2/0/0
TECH_733	Memory Reconsolidation I	443	0→1 of 1	0→1	0%→100%	1/0/0
TECH_763	Pranayama	434	0→2 of 2	50→50	100%→100%	2/0/0
tech	study	#315 bucket	outcome	scope
TECH_023	`10.1155/2013/653572`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_035	`10.3389/fpsyg.2014.01202`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_049	`10.25259/jnrp_157_2025`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_051	`10.1038/s41598-018-20299-z`	found_but_dropped	ON PAGE (matched-organic (pin authority))	technique
TECH_075	`10.1007/s11089-021-00945-6`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_079	`10.1016/j.brat.2017.09.010`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_107	`10.1007/s11695-022-06405-1`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_118	`10.1016/j.xcrm.2022.100895`	found_but_dropped	ON PAGE (matched-organic (pin authority))	technique
TECH_127	`10.1016/j.explore.2026.103414`	on_page_temporal	ON PAGE (was already on-page before seeding; matched-organic (pin authority))	technique
TECH_147	`pmid:30936803`	source_bound_not_indexed	ON PAGE (injected)	technique
TECH_157	`10.3389/fphys.2022.964144`	query_fixable	ON PAGE (injected)	technique
TECH_162	`10.4103/ym.ym_9_23`	?	ON PAGE (injected)	technique
TECH_163	`10.1016/j.jtcme.2017.02.003`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_180	`10.1186/s12906-020-03104-1`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_192	`10.1371/journal.pone.0058244`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_192	`10.1038/295234a0`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_207	`10.1016/s0065-2601(06)38002-1`	found_but_dropped	ON PAGE (matched-organic (pin authority))	technique
TECH_210	`10.1016/j.edurev.2018.09.002`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_240	`10.1002/jclp.23120`	on_page_temporal	ON PAGE (was already on-page before seeding; matched-organic (pin authority))	technique
TECH_252	`10.1037/a0035268`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_256	`10.1002/gps.1018`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_257	`10.11919/j.issn.1002-0829.215055`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_274	`10.1136/bmj.323.7327.1446`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_283	`10.1111/jpm.12886`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_283	`10.1176/appi.ajp.2018.17060611`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_285	`10.1111/jpm.12886`	found_but_dropped	ON PAGE (matched-organic (pin authority))	technique
TECH_319	`10.1177/0898010120938922`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_338	`10.1038/295234a0`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_338	`10.1371/journal.pone.0058244`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_345	`10.1177/20501579211028647`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_356	`10.2196/63078`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_359	`10.1007/s12671-014-0360-9`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_417	`10.4088/jcp.22m14621`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_421	`10.1097/brs.0b013e3181b315cc`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_431	`10.1016/j.jcbs.2026.100999`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_431	`10.1016/j.beth.2012.05.003`	found_but_dropped	ON PAGE (matched-organic (pin authority))	technique
TECH_451	`10.1111/j.1467-9280.2007.01916.x`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_460	`10.1007/s12671-019-01134-6`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_476	`10.1037/a0035268`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_477	`10.1037/cou0000107`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_477	`10.1007/s10902-020-00236-6`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_479	`10.1111/aphw.70134`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_488	`10.1080/15299732.2020.1760173`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_492	`10.1111/famp.12636`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_508	`10.1080/15332691.2016.1253518`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_512	`10.1016/0092-6566(89)90020-2`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_520	`10.1023/a:1014576505223`	found_but_dropped	ON PAGE (was already on-page before seeding; matched-organic (pin authority))	technique
TECH_550	`10.1017/s0033291717001003`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_555	`10.1371/journal.pone.0304093`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_578	`10.1136/bmj.a884`	source_bound_deep	ON PAGE (injected)	technique
TECH_578	`10.7326/m15-0667`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_579	`10.3390/ijerph192113734`	baseline_should_have_found	ON PAGE (injected)	technique
TECH_627	`10.1186/1471-244x-8-41`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_627	`10.1016/j.jpsychores.2026.112563`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_633	`10.1080/20008198.2021.1929023`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_633	`10.1002/jts.22189`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_636	`10.1080/20008198.2021.1929023`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_636	`10.1002/jts.22189`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_655	`10.1136/bmj.323.7327.1446`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_676	`10.1037/cns0000368`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_677	`10.1073/pnas.2013097118`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_703	`10.25259/ijpp_325_2023`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_725	`10.1016/j.psychsport.2020.101672`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_725	`10.1037/0021-9010.79.4.481`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_733	`10.1016/j.jbtep.2016.11.003`	query_fixable_title_only	ON PAGE (injected)	technique
TECH_763	`10.3389/fpsyt.2025.1616996`	query_fixable	ON PAGE (injected)	technique
TECH_763	`10.1038/s41598-022-27247-y`	query_fixable	ON PAGE (injected)	technique
	"""LAB-251 — final report from the KEYED full-pool run (production .env: all providers + cachekit
	edge + Lever-3 Haiku design pass; perplexity_dr 401 = same 6/7 state as the original #315 audit).

	Every page: live organic acquisition ONCE, then the production selection path twice from that
	identical pool — BEFORE (no seeding) and AFTER (curated seeding, authoritative pins).
	"""

	from __future__ import annotations

	import csv
	import json
	import re
	from pathlib import Path

	HERE = Path(__file__).resolve().parent
	OUT = HERE / "out"
	REPO_DATA = HERE.parent / "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data"


	def registry_key(raw: str) -> str \| None:
	raw = (raw or "").strip().strip("`")
	if not raw:
	return None
	if raw.upper().startswith("PMID"):
	m = re.search(r"\d+", raw.split("/")[0])
	return f"pmid:{m.group()}" if m else None
	doi = raw.split(" ")[0] if raw.startswith("10.") else raw
	return doi.lower() if doi.startswith("10.") else None


	def load_buckets() -> dict[tuple[str, str], str]:
	text = (HERE / "geo315_report.md").read_text(encoding="utf-8")
	out: dict[tuple[str, str], str] = {}

	def rows(a: str, b: str) -> list[list[str]]:
	blk = text.split(a, 1)[1].split(b, 1)[0]
	return [[c.strip() for c in ln.strip("\|").split("\|")] for ln in blk.splitlines() if ln.startswith("\| TECH_")]

	for c in rows("## found_but_dropped", "## acquisition_miss"):
	if (k := registry_key(c[2])):
	out[(c[0], k)] = "found_but_dropped"
	for c in rows("## acquisition_miss", "## on_page"):
	if (k := registry_key(c[2])):
	out[(c[0], k)] = c[3]
	for c in rows("## on_page", "## Caveats"):
	if (k := registry_key(c[1])):
	out[(c[0], k)] = "on_page_temporal"
	return out


	def main() -> None:
	buckets = load_buckets()
	reg_rows = list(csv.DictReader(open(REPO_DATA / "missing-literature.csv", encoding="utf-8")))
	full = {p.stem: json.loads(p.read_text()) for p in sorted(OUT.glob("TECH_*.json"))}

	def pct(counts: dict, scope: str) -> float:
	t = sum(counts.values())
	return 100.0 * counts.get(scope, 0) / t if t else 0.0

	page_lines = [
	"\| tech \| technique \| pool \| reg on-page before→after \| selected before→after \| tech-direct% b→a \| pins inj/matched/failed \| capped_out \|",
	"\|---\|---\|---:\|---\|---\|---\|---\|---:\|",
	]
	regressions: list[str] = []
	tot_before = tot_after = tot_reg = 0
	tot_inj = tot_mat = tot_ff = tot_cap = 0
	scope_tot = {"before": {"technique": 0, "n": 0}, "after": {"technique": 0, "n": 0}}
	empty_before_pages = []
	for tid, d in full.items():
	b, a, sp = d["before"], d["after"], d["after"]["seed_prov"]
	tot_before += len(b["registry_on_page"])
	tot_after += len(a["registry_on_page"])
	tot_reg += len(d["registry_keys"])
	tot_inj += len(sp.get("injected") or [])
	tot_mat += len(sp.get("matched_present") or [])
	tot_ff += len(sp.get("fetch_failed") or [])
	tot_cap += len(a["capped_out"])
	for lane_name, lane in (("before", b), ("after", a)):
	scope_tot[lane_name]["technique"] += lane["scope_counts"].get("technique", 0)
	scope_tot[lane_name]["n"] += sum(lane["scope_counts"].values())
	if b["n_selected"] == 0:
	empty_before_pages.append(tid)
	lost = set(b["registry_on_page"]) - set(a["registry_on_page"])
	if lost or len(a["registry_on_page"]) < len(b["registry_on_page"]):
	regressions.append(f"{tid}: {sorted(lost)}")
	page_lines.append(
	f"\| {tid} \| {d['technique'][:24]} \| {d['pool_size']}"
	f" \| {len(b['registry_on_page'])}→{len(a['registry_on_page'])} of {len(d['registry_keys'])}"
	f" \| {b['n_selected']}→{a['n_selected']}"
	f" \| {pct(b['scope_counts'], 'technique'):.0f}%→{pct(a['scope_counts'], 'technique'):.0f}%"
	f" \| {len(sp.get('injected') or [])}/{len(sp.get('matched_present') or [])}/{len(sp.get('fetch_failed') or [])}"
	f" \| {len(a['capped_out'])} \|"
	)

	led_lines = ["\| tech \| study \| #315 bucket \| outcome \| scope \|", "\|---\|---\|---\|---\|---\|"]
	on_page = tech_direct = excluded = 0
	vocab_total = vocab_on = 0
	for row in reg_rows:
	tid = row["tech_id"]
	key = registry_key(row["doi"]) or f"?:{row['doi'][:30]}"
	bucket = buckets.get((tid, key), "?")
	d = full.get(tid)
	outcome, scope = "PAGE NOT RUN", "—"
	if d:
	b, a, sp = d["before"], d["after"], d["after"]["seed_prov"]
	pin_scopes = {e["key"]: e["scope"] for e in a.get("selected_pinned_ranks", [])}
	if key in a["registry_on_page"]:
	on_page += 1
	scope = pin_scopes.get(key, "?")
	if scope == "technique":
	tech_direct += 1
	lane = (
	"injected"
	if key in (sp.get("injected") or [])
	else "matched-organic (pin authority)"
	if key in (sp.get("matched_present") or [])
	else "organic"
	)
	pre = "was already on-page before seeding; " if key in b["registry_on_page"] else ""
	outcome = f"ON PAGE ({pre}{lane})"
	elif key in (sp.get("fetch_failed") or []):
	excluded += 1
	outcome = "EXCLUDED: fetch_failed"
	elif key in (sp.get("citation_mismatch") or []):
	excluded += 1
	outcome = "EXCLUDED: citation mismatch (wrong-paper guard)"
	elif key in (sp.get("retracted_refused") or []):
	excluded += 1
	outcome = "EXCLUDED: retracted"
	elif key in a["capped_out"]:
	outcome = "CAPPED OUT (regression!)"
	else:
	outcome = "NOT MATCHED (check)"
	if bucket == "query_fixable_title_only":
	vocab_total += 1
	if outcome.startswith("ON PAGE"):
	vocab_on += 1
	led_lines.append(f"\| {tid} \| `{key}` \| {bucket} \| {outcome} \| {scope} \|")

	lines = [
	"# LAB-251 — live recall validation (KEYED production run): 67 seeded studies across 56 pages",
	"",
	"Config: production `.env` (all provider keys + cachekit edge + Lever-3 Haiku design pass;",
	"`perplexity_dr` 401-expired = same 6/7-provider state as the original #315 audit and the",
	"2026-06-30 batch). Per page: one live organic staged acquisition, then the production",
	"selection path twice from the identical pool — before (no seeding) vs after (curated seeding).",
	"",
	"## Headline",
	"",
	f"- Pages: {len(full)}/56 run, 0 failures. Registry studies on-page: {tot_before}→{tot_after} of {tot_reg}.",
	f"- Pin lanes: {tot_inj} injected + {tot_mat} matched-organic (pin authority); {tot_ff} fetch-failed; "
	f"{tot_cap} capped out; {excluded} excluded.",
	f"- {tech_direct}/{tot_after} on-page studies classify technique-direct.",
	f"- Vocabulary-gap studies on-page (AC target ~45/47): {vocab_on}/{vocab_total}.",
	f"- Aggregate technique-direct share of selected refs: "
	f"{100 * scope_tot['before']['technique'] / max(scope_tot['before']['n'], 1):.1f}%→"
	f"{100 * scope_tot['after']['technique'] / max(scope_tot['after']['n'], 1):.1f}%.",
	f"- Regressions: {len(regressions)}" + ("" if not regressions else " — " + "; ".join(regressions)),
	f"- Pages whose BEFORE selection is EMPTY (would EvidenceFloorError/hold without seeding): "
	f"{len(empty_before_pages)} — {', '.join(empty_before_pages) if empty_before_pages else '-'}",
	"",
	"## Per-page recall (before→after, same live pool per page)",
	"",
	*page_lines,
	"",
	"## 67-study ledger",
	"",
	*led_lines,
	]
	(HERE / "lab251_report.md").write_text("\n".join(lines), encoding="utf-8")
	print("\n".join(lines[: lines.index("## Per-page recall (before→after, same live pool per page)")]))


	if __name__ == "__main__":
	main()
	"""LAB-251 — live recall validation of the curated-seeding path (67 pins / 56 pages).

	Per technique: acquire the organic pool ONCE (live, keyless Stage-1 providers), then run the
	production selection path twice from that identical pool — BEFORE (no seeding = the 6cc50bdc
	baseline behaviour on current spine) and AFTER (curated seeding, authoritative pins). Selection
	calls the SAME node functions in the SAME order as run_staged_acquisition; the only node step
	skipped is the live Haiku design pre-pass (GEO_ABSTRACT_DESIGN_CLASSIFY — needs an LLM key;
	same omission as the original geo315 audit, caveat 3).

	Run: uv run --project <repo> python lab251_validate.py [--tech TECH_023] [--all]
	Outputs: out/{tech_id}.json (resume-safe: existing files are skipped), out/_failures.json
	"""

	from __future__ import annotations

	import argparse
	import json
	import logging
	import os
	import time
	import traceback
	from pathlib import Path

	HERE = Path(__file__).resolve().parent
	OUT = HERE / "out"
	OUT.mkdir(exist_ok=True)

	logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")

	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.curated_refs import ( # noqa: E402
	apply_curated_references,
	pin_record_key,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.insight_labs_web_first.nodes.evidence_acquisition import ( # noqa: E402
	_STAGED_MAX_SELECTED_REFERENCES,
	_abstract_design_enabled,
	_classify_designs_via_llm,
	_diversity_select_enabled,
	_extract_outcome_dimension,
	_extract_outcome_dimensions,
	_merged_evidence_to_selected_reference,
	_packet_scope_aliases,
	_partition_admissible_records,
	_partition_topically_relevant_records,
	_quality_first_reorder,
	_select_with_quota,
	_source_context_terms,
	_topical_relevance_gate_enabled,
	load_input_packet,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.dedupe import ( # noqa: E402
	apply_cochrane_design_override,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.identity_norm import ( # noqa: E402
	normalize_doi,
	normalize_pmid,
	)
	from ambient_intelligence.workflows.evidence_taxonomy.nodes.evidence_acquisition.staged_orchestrator import ( # noqa: E402
	acquire_evidence_staged_with_taxonomy,
	)


	def registry_key(raw: str) -> str \| None:
	"""Identity key for a missing-literature.csv doi cell: normalized DOI, else pmid:{digits}.

	Mirrors curated_refs._key_for_value ordering (PMID first) but tolerates the CSV's
	annotated cells ("PMID 30936803 / PMCID PMC6438091", "10.1136/bmj.a884 (PMID ...)").
	"""
	raw = (raw or "").strip()
	if not raw:
	return None
	if raw.upper().startswith("PMID"):
	pmid = normalize_pmid(raw.split("/")[0])
	return f"pmid:{pmid}" if pmid else None
	return normalize_doi(raw.split(" ")[0] if raw.startswith("10.") else raw)


	def record_keys(rec) -> set[str]:
	keys = set()
	if nd := normalize_doi(getattr(rec, "doi", None)):
	keys.add(nd)
	if pmid := normalize_pmid(getattr(rec, "pmid", None)):
	keys.add(f"pmid:{pmid}")
	return keys


	def run_lane(pool, curated, *, tech_id, technique, scope_ctx, registry_keys_for_tech):
	prov: dict = {}
	records = apply_curated_references(list(pool), curated, live=True, tech_id=tech_id, provenance=prov)
	pins = frozenset(prov.get("injected") or ()) \| frozenset(prov.get("matched_present") or ())
	# Lever 3 (node parity, seed → LLM classify → cochrane): abstract-aware Haiku design pass on the
	# coarse residual. Skips any record already stamped (curated-pin assertions protected) and is
	# edge-cached, so the second lane re-reads identical verdicts; in-place stamps on shared pool
	# objects keep both lanes consistent even cache-cold. Soft-fails internally; requires live keys.
	if _abstract_design_enabled():
	_classify_designs_via_llm(records)
	apply_cochrane_design_override(records)
	reordered = _quality_first_reorder(records, scope_ctx=scope_ctx, pinned_keys=pins)
	admitted, gate_rejected = _partition_admissible_records(reordered)
	relevance_rejected: list = []
	if _topical_relevance_gate_enabled():
	admitted, relevance_rejected = _partition_topically_relevant_records(admitted, scope_ctx=scope_ctx, pinned_keys=pins)
	if _diversity_select_enabled():
	capped, _ = _select_with_quota(admitted)
	else:
	capped = admitted[: _STAGED_MAX_SELECTED_REFERENCES]

	selected = [
	_merged_evidence_to_selected_reference(
	rec,
	rank=i,
	technique=technique,
	aliases=scope_ctx["aliases"],
	modality=scope_ctx["modality"],
	mechanism_terms=scope_ctx["mechanism_terms"],
	category=scope_ctx["category"],
	is_pinned=pin_record_key(rec) in pins,
	)
	for i, rec in enumerate(capped, start=1)
	]
	capped_keys = {k for r in capped for k in record_keys(r)}
	scope_counts: dict[str, int] = {}
	for ref in selected:
	scope = ref.get("evidence_scope") or "indirect"
	scope_counts[scope] = scope_counts.get(scope, 0) + 1
	pool_keys = {k for r in records for k in record_keys(r)}
	return {
	"n_selected": len(selected),
	"scope_counts": scope_counts,
	"registry_on_page": sorted(k for k in registry_keys_for_tech if k in capped_keys),
	"registry_in_pool": sorted(k for k in registry_keys_for_tech if k in pool_keys),
	"capped_out": sorted(pins - {k for r in capped if (k := pin_record_key(r))}),
	"gate_rejected": len(gate_rejected),
	"relevance_rejected": len(relevance_rejected),
	"seed_prov": {
	k: prov.get(k)
	for k in (
	"injected",
	"matched_present",
	"fetch_failed",
	"retracted_refused",
	"citation_mismatch",
	"drop_conflicts",
	"invalid",
	"asserted",
	"pins",
	)
	if prov.get(k)
	},
	"selected_pinned_ranks": [
	{"rank": i, "key": pin_record_key(rec), "scope": selected[i - 1].get("evidence_scope")}
	for i, rec in enumerate(capped, start=1)
	if pin_record_key(rec) in pins
	],
	}


	def run_tech(tech_id: str, registry_keys_for_tech: set[str]) -> dict:
	t0 = time.time()
	outdir = OUT / "runs" / tech_id
	outdir.mkdir(parents=True, exist_ok=True)
	state: dict = {"evidence_mode": "fresh", "tech_id": tech_id, "live": True, "output_dir": str(outdir)}
	state = load_input_packet(state)
	packet = state["input_packet"]
	technique = packet.get("technique") or packet.get("display_label") or tech_id
	outcome = _extract_outcome_dimension(state)
	extras: list[str] = []
	if os.environ.get("GEO_MULTI_OUTCOME_ACQUISITION", "1") != "0":
	extras = _extract_outcome_dimensions(state)[1:]

	# All 7 providers enabled — the audit's default config. The 3 keyless ones (consensus/openai/
	# perplexity) contribute 0 records in this runtime, so the organic pool is production-minus-those-
	# keys; keeping them enabled ALSO paces the SS/OpenAlex index calls (consensus's transient retry
	# delay spaces the burst) which, without an S2 API key, is what keeps them under the anonymous
	# 429 rate limit. Disabling them sped each tech up but throttled the real index providers into
	# 120s timeouts — net slower and a degraded pool. Slow-and-paced is both faithful and reliable.
	result = acquire_evidence_staged_with_taxonomy(tech_id, outcome, extra_outcome_dimensions=extras)
	pool = result.records

	scope_ctx = {
	"technique": technique,
	"aliases": _packet_scope_aliases(packet),
	"modality": str(packet.get("modality") or ""),
	"mechanism_terms": _source_context_terms(packet).get("mechanism", []),
	"category": str(packet.get("category") or ""),
	}

	provider_status = {
	name: {"status": ps.status, "records": ps.record_count} for name, ps in (result.provider_status or {}).items()
	}

	before = run_lane(
	pool,
	{"pinned": [], "dropped": []},
	tech_id=tech_id,
	technique=technique,
	scope_ctx=scope_ctx,
	registry_keys_for_tech=registry_keys_for_tech,
	)
	after = run_lane(
	pool,
	packet.get("curated_references") or {},
	tech_id=tech_id,
	technique=technique,
	scope_ctx=scope_ctx,
	registry_keys_for_tech=registry_keys_for_tech,
	)
	return {
	"tech_id": tech_id,
	"technique": technique,
	"outcome_dimension": outcome,
	"extra_outcome_dimensions": extras,
	"pool_size": len(pool),
	"provider_status": provider_status,
	"registry_keys": sorted(registry_keys_for_tech),
	"before": before,
	"after": after,
	"elapsed_s": round(time.time() - t0, 1),
	}


	def load_registry() -> dict[str, set[str]]:
	import csv

	repo_data = (
	Path(__file__).resolve().parents[1]
	/ "ambient-intelligence/ambient_intelligence/workflows/evidence_taxonomy/insight_labs_web_first/data/missing-literature.csv"
	)
	reg: dict[str, set[str]] = {}
	for row in csv.DictReader(open(repo_data, encoding="utf-8")):
	key = registry_key(row["doi"])
	if key:
	reg.setdefault(row["tech_id"], set()).add(key)
	return reg


	def main() -> None:
	ap = argparse.ArgumentParser()
	ap.add_argument("--tech", action="append", default=None)
	ap.add_argument("--all", action="store_true")
	ap.add_argument("--shard", default=None, help="i/n — process techs where index %% n == i (parallel workers)")
	args = ap.parse_args()

	registry = load_registry()
	techs = args.tech or (sorted(registry) if args.all else [])
	if not techs:
	ap.error("pass --tech TECH_xxx or --all")
	if args.shard:
	i, n = (int(x) for x in args.shard.split("/"))
	techs = [t for idx, t in enumerate(techs) if idx % n == i]

	failures = {}
	fail_path = OUT / "_failures.json"
	if fail_path.is_file():
	failures = json.loads(fail_path.read_text())
	for tech_id in techs:
	dest = OUT / f"{tech_id}.json"
	if dest.is_file():
	print(f"{tech_id}: exists, skipped", flush=True)
	continue
	try:
	res = run_tech(tech_id, registry.get(tech_id, set()))
	except Exception as exc: # record and continue — a one-tech failure must not kill the sweep
	failures[tech_id] = {"error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc()[-2000:]}
	fail_path.write_text(json.dumps(failures, indent=2))
	print(f"{tech_id}: FAILED {type(exc).__name__}: {exc}", flush=True)
	continue
	failures.pop(tech_id, None)
	fail_path.write_text(json.dumps(failures, indent=2))
	dest.write_text(json.dumps(res, indent=2))
	b, a = res["before"], res["after"]
	print(
	f"{tech_id}: pool={res['pool_size']} on_page {len(b['registry_on_page'])}->{len(a['registry_on_page'])}"
	f" of {len(res['registry_keys'])} \| injected={len(a['seed_prov'].get('injected') or [])}"
	f" matched={len(a['seed_prov'].get('matched_present') or [])}"
	f" fetch_failed={len(a['seed_prov'].get('fetch_failed') or [])}"
	f" capped_out={len(a['capped_out'])} \| {res['elapsed_s']}s",
	flush=True,
	)


	if __name__ == "__main__":
	main()