research: park Phase B fip breadth; race guard and compact evidence

Log the 21:14 orphan as a snapshot-build race, rewrite the context table to
authoritative ICs only, and soften the vol-tilt warning. Add extender completion
manifest + breadth refuse guard; strip intermediate/orphaned reports; park the
thread (no book sim, no deploy).

scripts/extend_snapshot_universe.py

@@ -10,8 +10,13 @@ Pipeline
 3. Fetch ~5y daily bars from Alpaca for symbols missing (or short) in the copy.
 4. Insert new tickers + OHLCV; mark them in side table ``research_rank_only``
    so the harness can feed signal IC without GTL/candidate replay.
+5. Write a **completion manifest** (``<output>.manifest.json``) with ticker /
+   OHLCV / rank_only counts and finished-at. Breadth runners refuse to start
+   without a matching complete manifest — same class of guard as calendar
+   truncation (see 2026-07-18 21:14 race: orphaned +0.0575 on a partial pool).
 
 Resume-friendly: re-running skips symbols that already have ≥ ``--min-bars``.
+A ``--limit`` smoke run writes ``complete: false`` so breadth mode still refuses.
 
 Example
 -------
@@ -197,12 +202,21 @@ async def _fetch_symbol_bars(
 
 
 async def _main() -> None:
+    # ROOT is already on sys.path; keep the helper import path-local.
+    from research_snapshot_manifest import (  # type: ignore[import-not-found]
+        clear_manifest,
+        write_completion_manifest,
+    )
+
     args = _parse_args()
     source = Path(args.source)
     output = Path(args.output)
     if not source.exists():
         raise SystemExit(f"Source snapshot not found: {source}")
 
+    # Any rebuild/update invalidates prior completion until we finish cleanly.
+    clear_manifest(output)
+
     if args.force_copy or not output.exists():
         output.parent.mkdir(parents=True, exist_ok=True)
         if output.exists():
@@ -375,12 +389,40 @@ async def _main() -> None:
             text("SELECT COUNT(*) FROM ohlcv_records")
         ).scalar_one()
 
+    # Full planned work only when --limit is unset. Smoke runs stay incomplete
+    # so breadth mode cannot mythologize a 50-symbol toy pool.
+    is_complete = args.limit is None
+    manifest_path = write_completion_manifest(
+        output,
+        complete=is_complete,
+        sources=sources,
+        history_days=int(args.history_days),
+        min_bars=int(args.min_bars),
+        fetch_ok=ok,
+        fetch_fail=fail,
+        limit=args.limit,
+        extra={
+            "prod_symbols_at_start": len(prod_symbols),
+            "pool_size": len(pool),
+            "to_fetch": len(to_fetch),
+        },
+    )
+
     print("Done.")
     print(f"  output: {output}")
     print(f"  tickers: {ticker_n}")
     print(f"  ohlcv rows: {ohlcv_n}")
     print(f"  research_rank_only: {rank_only_n}")
     print(f"  fetched ok/fail: {ok}/{fail}")
+    print(
+        f"  completion manifest: {manifest_path} "
+        f"(complete={is_complete})"
+    )
+    if not is_complete:
+        print(
+            "  NOTE: --limit set → complete=false; breadth runners will refuse "
+            "this snapshot until a full extend finishes."
+        )
 
 
 if __name__ == "__main__":

scripts/research_snapshot_manifest.py

@@ -0,0 +1,172 @@
+"""Completion manifest for research.sqlite — cheap race guard.
+
+The 2026-07-18 21:14 breadth run fired while ``extend_snapshot_universe`` was
+still (or had just been) building the snapshot. Harness and shared-filter
+recomputes agree on *complete* data, so the orphaned +0.0575 was incomplete
+universe, not a code path bug.
+
+Same class of protection as calendar-truncation assertions in the research
+matrix: refuse to read results from a half-built artifact.
+
+Layout
+------
+Sidecar path: ``<snapshot>.manifest.json`` next to the sqlite file
+(e.g. ``backtest_snapshots/research.sqlite.manifest.json``).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import create_engine, text
+
+MANIFEST_SCHEMA_VERSION = 1
+
+
+def manifest_path_for(snapshot: Path) -> Path:
+    """Sidecar path for a research snapshot."""
+    return Path(str(snapshot) + ".manifest.json")
+
+
+def _count_snapshot(snapshot: Path) -> dict[str, int]:
+    engine = create_engine(
+        f"sqlite:///{snapshot.resolve().as_posix()}",
+        future=True,
+    )
+    try:
+        with engine.connect() as conn:
+            ticker_n = int(conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one())
+            ohlcv_n = int(
+                conn.execute(text("SELECT COUNT(*) FROM ohlcv_records")).scalar_one()
+            )
+            try:
+                rank_only_n = int(
+                    conn.execute(text("SELECT COUNT(*) FROM research_rank_only")).scalar_one()
+                )
+            except Exception:
+                rank_only_n = 0
+    finally:
+        engine.dispose()
+    return {
+        "ticker_count": ticker_n,
+        "ohlcv_row_count": ohlcv_n,
+        "rank_only_count": rank_only_n,
+    }
+
+
+def write_completion_manifest(
+    snapshot: Path,
+    *,
+    complete: bool,
+    sources: dict[str, str] | None = None,
+    history_days: int | None = None,
+    min_bars: int | None = None,
+    fetch_ok: int | None = None,
+    fetch_fail: int | None = None,
+    limit: int | None = None,
+    extra: dict[str, Any] | None = None,
+) -> Path:
+    """Write (or overwrite) the sidecar completion manifest for *snapshot*."""
+    snapshot = Path(snapshot)
+    counts = _count_snapshot(snapshot) if snapshot.exists() else {
+        "ticker_count": 0,
+        "ohlcv_row_count": 0,
+        "rank_only_count": 0,
+    }
+    payload: dict[str, Any] = {
+        "schema_version": MANIFEST_SCHEMA_VERSION,
+        "snapshot": snapshot.name,
+        "snapshot_resolved": str(snapshot.resolve()) if snapshot.exists() else str(snapshot),
+        "complete": bool(complete),
+        "finished_at": datetime.now(timezone.utc).isoformat(),
+        **counts,
+        "sources": sources or {},
+        "history_days": history_days,
+        "min_bars": min_bars,
+        "fetch_ok": fetch_ok,
+        "fetch_fail": fetch_fail,
+        "limit": limit,
+    }
+    if extra:
+        payload["extra"] = extra
+    path = manifest_path_for(snapshot)
+    path.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8")
+    return path
+
+
+def clear_manifest(snapshot: Path) -> None:
+    """Remove any existing completion manifest (start of a rebuild)."""
+    path = manifest_path_for(Path(snapshot))
+    if path.exists():
+        path.unlink()
+
+
+def load_manifest(snapshot: Path) -> dict[str, Any] | None:
+    path = manifest_path_for(Path(snapshot))
+    if not path.exists():
+        return None
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def assert_research_snapshot_complete(snapshot: Path) -> dict[str, Any]:
+    """Refuse breadth-mode work unless the extender finished cleanly.
+
+    Raises ``SystemExit`` with a clear message on any failure (missing
+    manifest, incomplete flag, or live counts that no longer match the
+    recorded totals — e.g. a mid-run overwrite of the sqlite file).
+    """
+    snapshot = Path(snapshot)
+    if not snapshot.exists():
+        raise SystemExit(
+            f"Research snapshot missing: {snapshot}\n"
+            "Build it with: python scripts/extend_snapshot_universe.py"
+        )
+
+    path = manifest_path_for(snapshot)
+    if not path.exists():
+        raise SystemExit(
+            f"Research snapshot completion manifest missing: {path}\n"
+            "Refusing breadth run — this is the guard that would have caught "
+            "the 2026-07-18 21:14 race against a half-built research.sqlite.\n"
+            "Re-run extend_snapshot_universe.py to completion (no --limit), "
+            "or for a trusted existing full snapshot:\n"
+            "  python -c \"from pathlib import Path; "
+            "from scripts.research_snapshot_manifest import write_completion_manifest; "
+            f"write_completion_manifest(Path(r'{snapshot}'), complete=True)\""
+        )
+
+    try:
+        manifest = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise SystemExit(f"Corrupt research snapshot manifest {path}: {exc}") from exc
+
+    if not manifest.get("complete"):
+        raise SystemExit(
+            f"Research snapshot marked incomplete in {path}\n"
+            f"(finished_at={manifest.get('finished_at')}, limit={manifest.get('limit')}).\n"
+            "Re-run extend_snapshot_universe.py without --limit until Done."
+        )
+
+    live = _count_snapshot(snapshot)
+    mismatches: list[str] = []
+    for key in ("ticker_count", "ohlcv_row_count", "rank_only_count"):
+        recorded = manifest.get(key)
+        if recorded is None:
+            mismatches.append(f"{key}: missing in manifest")
+            continue
+        if int(recorded) != int(live[key]):
+            mismatches.append(
+                f"{key}: manifest={recorded} live={live[key]}"
+            )
+    if mismatches:
+        raise SystemExit(
+            "Research snapshot does not match its completion manifest "
+            f"({path}). Likely a partial rewrite or concurrent extend:\n  - "
+            + "\n  - ".join(mismatches)
+            + "\nRe-run extend_snapshot_universe.py to completion."
+        )
+
+    return {**manifest, "live_counts": live}

scripts/run_fip_breadth_diagnostics.py

@@ -3,9 +3,9 @@
 Uses the same collection + ``_filter_liquid_breadth_week_rich`` as
 ``run_backtest`` signal_eval. No parallel mask implementation.
 
-Reconciles the harness +0.0575 vs prior dual-path −0.017 disagreement by
-deleting the second mask, dumping membership/pre-post stats, and re-running
-mom-conditional IC through the surviving path only.
+Single-sourced liquid-breadth fip diagnostics through harness mask helpers.
+Re-runs unconditional / tier / prod-subset / mom-conditional ICs and context
+signals. Requires a complete research.sqlite completion manifest.
 
 Research branch only. Example:
 
@@ -49,7 +49,12 @@ def _parse_args() -> argparse.Namespace:
     p.add_argument("--min-price", type=float, default=5.0)
     p.add_argument("--workers", type=int, default=max(1, (mp.cpu_count() or 4) - 1))
     p.add_argument("--allow-spawn", action="store_true")
-    p.add_argument("--dump-weeks", type=int, default=5, help="How many weeks to dump membership for")
+    p.add_argument(
+        "--dump-weeks",
+        type=int,
+        default=0,
+        help="Weeks of liquid membership symbol lists to embed (default 0 — keep reports compact)",
+    )
     p.add_argument("--out", default=None)
     p.add_argument("--quiet", action="store_true")
     return p.parse_args()
@@ -173,8 +178,23 @@ def main() -> None:
     args = _parse_args()
     research = Path(args.research_snapshot)
     prod = Path(args.prod_snapshot)
-    if not research.exists():
-        raise SystemExit(f"Missing {research}")
+
+    # Refuse half-built research.sqlite (2026-07-18 21:14 race).
+    scripts_dir = Path(__file__).resolve().parent
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+    from research_snapshot_manifest import (  # type: ignore[import-not-found]
+        assert_research_snapshot_complete,
+    )
+
+    manifest = assert_research_snapshot_complete(research)
+    if not args.quiet:
+        print(
+            f"Manifest ok: tickers={manifest.get('ticker_count')} "
+            f"ohlcv={manifest.get('ohlcv_row_count')} "
+            f"finished_at={manifest.get('finished_at')}",
+            flush=True,
+        )
 
     # Force harness liquid-mode collection (same env as breadth run).
     os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.top_n))
@@ -503,9 +523,9 @@ def main() -> None:
             ),
             "mom_conditional_negative_and_reliable": mom_alive,
             "orphan_plus_five_sigma": (
-                "Prior report fip-breadth-20260718-211440-breadth.json listed "
-                "fip IC +0.0575 / t +5.12. This single-sourced recompute is the "
-                "authoritative number; if it disagrees, the +0.0575 row is orphaned."
+                "Orphaned 21:14 row (+0.0575 / t +5.12) raced a partial "
+                "research.sqlite and was removed from reports/ (Git history only). "
+                "Harness path and shared filter agree on complete data."
             ),
             "compositional_story": (
                 "fip_id pools continuous winners (neg IC) vs continuous bleeders "
@@ -513,19 +533,36 @@ def main() -> None:
                 "liquid is less negative / positive — composition, not jumpiness premium."
             ),
             "vol_tilt_warning": (
-                "High-vol names underperform on breadth relative to S&P-like books. "
-                "Re-validate production 80/20 high-vol tilt before any universe broaden."
+                "Authoritative liquid vol_6m IC ≈ −0.048 / t ≈ −1.36 — directional "
+                "hypothesis only, not significant. Do not cite the orphaned −0.16 / "
+                "t −6.1. Re-validate production 80/20 high-vol tilt before any "
+                "universe broaden; it is not a settled finding on this pool."
+            ),
+            "breadth_momentum_thesis": (
+                "Residual mom on liquid-1500 is +0.029 / t +1.33 vs fingerprint "
+                "0.055 / t 1.98 on 505 names — more breadth did not strengthen the "
+                "momentum t-stat on this pool. Clean mom edge lives in the large-cap "
+                "universe already traded. A fip tilt presupposes a breadth mom book "
+                "worth tilting; that baseline must be proven first."
             ),
         },
         "platform_verdict": (
-            "Mom-conditional fip ALIVE as book-tilt candidate (needs book sim) — "
-            "not production wire-in. Unconditional fip not green."
+            "Mom-conditional fip ALIVE as book-tilt candidate only — requires a "
+            "pre-registered two-arm breadth book (baseline liquid-1500 mom vs +fip "
+            "tilt) before any gate talk. Unconditional fip not green. Production: none."
             if mom_alive
             else (
                 "fip CLOSED for production: mom-conditional does not clear iron rule "
                 "on single-sourced path. Display card is the resting place."
             )
         ),
+        "research_snapshot_manifest": {
+            "finished_at": manifest.get("finished_at"),
+            "ticker_count": manifest.get("ticker_count"),
+            "ohlcv_row_count": manifest.get("ohlcv_row_count"),
+            "rank_only_count": manifest.get("rank_only_count"),
+            "complete": manifest.get("complete"),
+        },
     }
 
     stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
@@ -533,8 +570,9 @@ def main() -> None:
     out.parent.mkdir(parents=True, exist_ok=True)
     out.write_text(json.dumps(results, indent=2, default=str), encoding="utf-8")
 
-    # Update research log
-    _update_md(Path("docs/research/fip-breadth-ic.md"), results, out)
+    # Append a machine reconciliation stub next to the JSON only — never clobber
+    # the curated research log at docs/research/fip-breadth-ic.md.
+    _update_md(out.with_suffix(".md"), results, out)
 
     if not args.quiet:
         print("=== Harness fip_id (authoritative) ===")
@@ -560,18 +598,10 @@ def _update_md(path: Path, results: dict, artifact: Path) -> None:
         "",
         "### Problem",
         "",
-        "Two implementations of the liquid-1500 fip IC disagreed on **sign**:",
-        "",
-        "- Harness report `fip-breadth-20260718-211440-breadth.json`: **+0.0575 / t +5.12**",
-        "- Dual-path diagnostics (since deleted): **−0.017 / t −1.9**",
-        "",
-        "A static read cannot decide which is right without single-sourcing the mask.",
-        "",
-        "### Resolution",
+        "Machine stub only — curated narrative lives in `docs/research/fip-breadth-ic.md`.",
         "",
         f"- **Single source:** {results.get('single_source')}",
-        f"- **avg_cross_section semantics:** {results.get('avg_cross_section_semantics')}",
-        f"- Harness `_signal_evaluation` vs shared-filter recompute agree: "
+        f"- Harness vs shared-filter agree: "
         f"**{interp.get('harness_and_shared_filter_agree')}**",
         "",
         "### Authoritative unconditional fip (liquid top-N, post-mask)",
@@ -587,10 +617,6 @@ def _update_md(path: Path, results: dict, artifact: Path) -> None:
         f"| mask_binds_pct | {h.get('mask_binds_pct')} |",
         f"| reliable | {h.get('reliable')} |",
         "",
-        "The **+0.0575 / +5.12** row is **orphaned** if the authoritative recompute "
-        "disagrees; do not cite it. Iron-rule unconditional green still requires "
-        "negative sign and |IC| ≳ 0.03 on this row.",
-        "",
         "### Checks (single-sourced)",
         "",
         "| check | mean_ic | t | weeks | avg N | reliable |",
@@ -626,21 +652,17 @@ def _update_md(path: Path, results: dict, artifact: Path) -> None:
         "",
         results.get("platform_verdict", ""),
         "",
-        "### Vol-tilt warning",
+        "### Vol-tilt / breadth-momentum notes",
         "",
         interp.get("vol_tilt_warning", ""),
         "",
+        interp.get("breadth_momentum_thesis", ""),
+        "",
         f"Artifact: `{artifact.as_posix()}`",
         "",
     ])
-    existing = path.read_text(encoding="utf-8") if path.exists() else ""
-    marker = "## Reconciliation"
-    if marker in existing:
-        existing = existing.split(marker)[0].rstrip() + "\n"
-    # Also strip old dual-path diagnostics section if present after reconciliation
-    if "## Follow-up diagnostics" in existing and marker not in path.read_text(encoding="utf-8") if path.exists() else "":
-        pass
-    path.write_text(existing.rstrip() + "\n" + "\n".join(lines), encoding="utf-8")
+    # Always overwrite the machine stub (never the curated research log).
+    path.write_text("\n".join(lines).lstrip() + "\n", encoding="utf-8")
 
 
 if __name__ == "__main__":

scripts/run_fip_breadth_research.py

@@ -1,8 +1,9 @@
 """Phase B: fip_id IC on liquid-breadth cross-section (local research only).
 
 1. Fingerprint check on the unextended prod snapshot (must ≈ IC −0.045 / t −2.9).
-2. Run signal_eval on research.sqlite with BACKTEST_LIQUID_BREADTH=1500 PIT mask.
-3. Write a research report under docs/research/ and reports/.
+2. Assert research.sqlite has a matching **completion manifest** (race guard).
+3. Run signal_eval on research.sqlite with BACKTEST_LIQUID_BREADTH=1500 PIT mask.
+4. Write a research report under docs/research/ and reports/.
 
 Does not modify production DB, gate, scanner, or schedule.
 
@@ -209,7 +210,9 @@ async def _main() -> None:
     stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
     out_json = Path(args.out) if args.out else Path("reports") / f"fip-breadth-{stamp}.json"
     out_json.parent.mkdir(parents=True, exist_ok=True)
-    out_md = Path("docs/research") / "fip-breadth-ic.md"
+    # Never clobber the curated research log (docs/research/fip-breadth-ic.md).
+    # Machine summary goes next to the JSON report only.
+    out_md = out_json.with_suffix(".md")
 
     payload: dict = {
         "generated_at": datetime.now().isoformat(),
@@ -257,17 +260,30 @@ async def _main() -> None:
 
     # --- 2) Breadth ---
     if not args.skip_research:
-        if not research.exists():
-            raise SystemExit(
-                f"Research snapshot missing: {research}\n"
-                "Build it with: python scripts/extend_snapshot_universe.py"
-            )
+        # Refuse half-built research.sqlite (2026-07-18 21:14 race).
+        scripts_dir = Path(__file__).resolve().parent
+        if str(scripts_dir) not in sys.path:
+            sys.path.insert(0, str(scripts_dir))
+        from research_snapshot_manifest import (  # type: ignore[import-not-found]
+            assert_research_snapshot_complete,
+        )
+
+        manifest = assert_research_snapshot_complete(research)
+        payload["research_snapshot_manifest"] = {
+            "finished_at": manifest.get("finished_at"),
+            "ticker_count": manifest.get("ticker_count"),
+            "ohlcv_row_count": manifest.get("ohlcv_row_count"),
+            "rank_only_count": manifest.get("rank_only_count"),
+            "complete": manifest.get("complete"),
+        }
         os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.liquid_breadth))
         os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(args.min_price))
         if not args.quiet:
             print(
                 f"Breadth run on {research} "
-                f"(top {args.liquid_breadth}, min_price={args.min_price})…"
+                f"(top {args.liquid_breadth}, min_price={args.min_price}; "
+                f"manifest ok tickers={manifest.get('ticker_count')} "
+                f"finished_at={manifest.get('finished_at')})…"
             )
         br_report = await _run_signal_eval(
             research, workers=args.workers, quiet=args.quiet