diff --git a/README.md b/README.md
index 5b15213..b580c64 100644
--- a/README.md
+++ b/README.md
@@ -98,10 +98,37 @@ COTDATA_STORE=/store  cotdata-update --cot-legacy                # CFTC Legacy (
 COTDATA_STORE=/store  cotdata-update --cot-disagg                # CFTC Disaggregated (cross-platform)
 COTDATA_STORE=/store  cotdata-update --cot-tff                   # CFTC Traders in Financial Futures (cross-platform)
 COTDATA_STORE=/store  cotdata-update --cot-all                   # Update all CFTC COT pipelines
+COTDATA_STORE=/store  cotdata-update --check                     # Store status (read-only, any platform)
+COTDATA_STORE=/store  cotdata-update --reconcile                 # Prune stale manifest ghosts (any platform)
 ```
 
+COT tables are stored per code as **`{symbol}_{code}`** (e.g. `RTY_23977A`), so a symbol's current and predecessor (`hist_codes`) contracts are both attributable to it; `get_cot("RTY")` stitches them. `--reconcile` drops manifest entries whose parquet file is missing — bare-code ghosts and retired domains left by older naming schemes — so `--check` and `status.json` show only real, consistently-prefixed entries. It never touches data (only removes bookkeeping for files that don't exist).
+
 Schedule nightly (prices, after the Norgate Data Updater) and weekly (COT Friday releases).
 
+Each run prints a per-symbol line (row counts and the date advance, e.g. `ES: … [2026-07-13 -> 2026-07-14]`) and a summary footer (OK/failed counts, rows written, elapsed, newest date). `--check` reports current store status from the manifest — per-domain row counts, newest data date, last write, and any entries lagging behind their peers (a partial-run signal) — without touching the network, so it runs anywhere the store is visible.
+
+### `status.json` — new-data signal for downstream tools
+
+Every producer run writes `$COTDATA_STORE/status.json` (atomically, alongside the data), so tools that trigger on fresh data can poll one small structured file instead of scanning the store:
+
+```json
+{
+  "generated_at": "2026-07-15T10:15:24Z",
+  "schema_version": 2,
+  "newest_data": { "prices": "2026-07-14", "cot_legacy": "2026-07-07", "cot_disagg": "2026-07-07", "cot_tff": "2026-07-07" },
+  "domains":     { "prices": { "newest_data": "2026-07-14", "last_write": "2026-07-15T10:15:24Z", "entries": 84, "rows": 829096, "lagging": 0 }, "...": {} },
+  "last_run":    { "kinds": ["prices"], "ok": ["ES", "..."], "symbols_failed": [], "rows": 1658000, "seconds": 88, "at": "2026-07-15T10:15:24Z" }
+}
+```
+
+**Polling contract:**
+- To detect **new data**, compare `newest_data.<domain>` (e.g. `newest_data.prices`, `newest_data.cot_legacy`) against your last-seen value. It advances **only when genuinely new daily data arrives** — a no-op run leaves it unchanged.
+- To detect that **a run happened at all** (new data or not), use `generated_at`.
+- `last_run` carries the most recent run's outcome (which domains, per-symbol failures) for alerting.
+
+Prices and each COT report (`cot_legacy`, `cot_disagg`, `cot_tff`) are separate domains, so a price-triggered tool and a COT-triggered tool each watch their own key.
+
 ## Design rules
 
 ### Why Back-Adjusted vs Unadjusted?
diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py
index fca2eb1..02f336e 100644
--- a/src/cotdata/providers/norgate.py
+++ b/src/cotdata/providers/norgate.py
@@ -233,7 +233,13 @@ def update(symbols=None, full: bool = False) -> None:
     than the trailing-60-day incremental update — use it after a reconstruction
     logic change so old rows are recomputed under the new algorithm.
     """
+    import time
+    from .. import status
+
     syms = symbols or [s.internal for s in all_symbols()]
+    prior = store.load_manifest().get("prices", {})  # to report per-symbol date deltas
+    t0 = time.time()
+    ok, failed, total_rows, newest = [], [], 0, None
     for sym in syms:
         try:
             # 1. Back-Adjusted
@@ -246,14 +252,29 @@ def update(symbols=None, full: bool = False) -> None:
             # 3. Volume Reconstruction (Additive)
             out_backadj = _reconstruct_volume(sym, out_backadj, "backadj", full=full)
             out_unadj = _reconstruct_volume(sym, out_unadj, "unadj", full=full)
-            
+
             store.write_prices(sym, "backadj", out_backadj, source="norgate")
             store.write_prices(sym, "unadj", out_unadj, source="norgate")
-            
-            print(f"{sym:5s}: {len(out_backadj):6d} backadj, {len(out_unadj):6d} unadj -> store")
+
+            ok.append(sym)
+            total_rows += len(out_backadj) + len(out_unadj)
+            new = str(out_backadj.index.max().date()) if len(out_backadj) else "—"
+            newest = max(newest, new) if newest else new
+            was = (prior.get(f"{sym}_backadj") or {}).get("last_date")
+            delta = new if (was is None or was == new) else f"{was} -> {new}"
+            print(f"{sym:5s}: {len(out_backadj):6d} backadj, {len(out_unadj):6d} unadj  [{delta}]")
         except Exception as e:  # noqa: BLE001
+            failed.append((sym, e))
             print(f"{sym:5s}: FAILED — {e}")
 
+    seconds = round(time.time() - t0, 1)
+    print(status.run_summary("prices update", ok, failed, total_rows, seconds, newest=newest))
+    return {
+        "kind": "prices", "ok": ok, "failed": [(s, str(e)) for s, e in failed],
+        "symbols_failed": [s for s, _ in failed], "rows": total_rows,
+        "seconds": seconds, "newest": newest,
+    }
+
 
 def get_symbol_metadata(internal_symbol: str) -> dict | None:
     """Fetch contract specifications for a single continuous futures symbol."""
diff --git a/src/cotdata/status.py b/src/cotdata/status.py
new file mode 100644
index 0000000..4b7273b
--- /dev/null
+++ b/src/cotdata/status.py
@@ -0,0 +1,186 @@
+"""Read-only store status (``cotdata-update --check``) and the producer run
+summary. Both work off manifest.json only — no network, no Windows/norgatedata,
+so ``--check`` runs anywhere the store is visible.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import os
+from typing import Optional
+
+from . import store, config
+
+STATUS_FILENAME = "status.json"
+
+
+def status_path():
+    return config.store_root() / STATUS_FILENAME
+
+# Domains shown by --check, in report order.
+_DOMAINS = ["prices", "metadata", "cot", "cot_legacy", "cot_disagg", "cot_tff"]
+# An entry lagging more than this many days behind its domain's newest data
+# probably failed to update while its peers succeeded (a partial run).
+_LAG_DAYS = 3
+
+
+def _parse_date(s: Optional[str]) -> Optional[dt.date]:
+    try:
+        return dt.date.fromisoformat(s) if s else None
+    except (ValueError, TypeError):
+        return None
+
+
+def hist_code_names() -> set:
+    """Entry names for symbols' predecessor (hist_code) COT contracts, e.g.
+    ``RTY_23977A``. These are retired codes, legitimately frozen at an old date,
+    so they should not be flagged as "lagging". Built from the registry."""
+    from .registry import all_symbols, hist_code_scales
+    names = set()
+    for s in all_symbols():
+        for hc, _ in hist_code_scales(s.hist_codes):
+            names.add(f"{s.internal}_{hc}")
+    return names
+
+
+def summarize(manifest: dict, today: Optional[dt.date] = None,
+              ignore_lag: Optional[set] = None) -> dict:
+    """Pure summary of a manifest — one record per non-empty domain.
+
+    ignore_lag: entry names never reported as lagging (e.g. retired hist_codes,
+    which are frozen by design). They still count toward entries/rows/newest."""
+    today = today or dt.date.today()
+    ignore_lag = ignore_lag or set()
+    out = {"schema_version": manifest.get("schema_version"),
+           "today": today.isoformat(), "domains": {}}
+    for domain in _DOMAINS:
+        entries = manifest.get(domain, {})
+        if not isinstance(entries, dict) or not entries:
+            continue
+        dated = {n: _parse_date(e.get("last_date")) for n, e in entries.items()}
+        dates = [d for d in dated.values() if d]
+        newest = max(dates) if dates else None
+        lagging = []
+        if newest:
+            for name, d in dated.items():
+                if name in ignore_lag:
+                    continue
+                behind = (newest - d).days if d else None
+                if behind is not None and behind > _LAG_DAYS:
+                    lagging.append((name, entries[name].get("last_date"), behind))
+        writes = [e.get("updated_at") for e in entries.values() if e.get("updated_at")]
+        out["domains"][domain] = {
+            "entries": len(entries),
+            "rows": sum(int(e.get("n_rows") or 0) for e in entries.values()),
+            "newest": newest.isoformat() if newest else None,
+            "oldest": min(dates).isoformat() if dates else None,
+            "last_write": max(writes) if writes else None,
+            "behind_today": (today - newest).days if newest else None,
+            "lagging": sorted(lagging, key=lambda t: -t[2]),
+        }
+    return out
+
+
+def format_report(manifest: dict, root: str = "", today: Optional[dt.date] = None,
+                  ignore_lag: Optional[set] = None) -> str:
+    """Human-readable --check report."""
+    s = summarize(manifest, today, ignore_lag=ignore_lag)
+    target = config.SCHEMA_VERSION
+    sv = s["schema_version"]
+    sv_note = "" if sv == target else f"  ⚠ target {target} — run the producer to migrate"
+    L = [f"cotdata store  ·  {root or '(COTDATA_STORE)'}",
+         f"as of {s['today']}  ·  schema_version {sv}{sv_note}",
+         ""]
+    if not s["domains"]:
+        L.append("store is empty — no data written yet.")
+        return "\n".join(L)
+
+    L.append(f"{'domain':<12}{'entries':>8}{'rows':>13}{'newest data':>14}"
+             f"{'last write (UTC)':>22}{'behind':>8}")
+    for domain, d in s["domains"].items():
+        behind = "—" if d["behind_today"] is None else f"{d['behind_today']}d"
+        L.append(f"{domain:<12}{d['entries']:>8}{d['rows']:>13,}"
+                 f"{str(d['newest']):>14}{str(d['last_write']):>22}{behind:>8}")
+
+    for domain, d in s["domains"].items():
+        if d["lagging"]:
+            L.append("")
+            L.append(f"⚠ {domain}: {len(d['lagging'])} entr{'y' if len(d['lagging'])==1 else 'ies'} "
+                     f"lag >{_LAG_DAYS}d behind newest ({d['newest']}):")
+            for name, last, behind in d["lagging"][:15]:
+                L.append(f"    {name:<18}{last}  ({behind}d behind)")
+            if len(d["lagging"]) > 15:
+                L.append(f"    … and {len(d['lagging']) - 15} more")
+    if not any(d["lagging"] for d in s["domains"].values()):
+        L.append("")
+        L.append("✓ all entries current (none lag behind their domain's newest).")
+    return "\n".join(L)
+
+
+def print_check() -> None:
+    """Entry point for ``cotdata-update --check``."""
+    root = str(config.store_root())
+    print(format_report(store.load_manifest(), root=root, ignore_lag=hist_code_names()))
+
+
+def build_status_doc(manifest: dict, last_run: Optional[dict] = None,
+                     today: Optional[dt.date] = None,
+                     ignore_lag: Optional[set] = None) -> dict:
+    """Machine-readable status document written to ``status.json`` after each run.
+
+    Contract for external pollers:
+      * ``newest_data[<domain>]`` — the date of the newest daily data for that
+        domain. Advances ONLY when genuinely new data arrives → key on this to
+        detect "there is new data" (e.g. compare prices vs your last-seen date).
+      * ``generated_at`` — refreshed on every producer run (new data or not) →
+        key on this only to detect "a run happened".
+      * ``last_run`` — outcome of the most recent run (kinds, ok/failed counts).
+    """
+    s = summarize(manifest, today, ignore_lag=ignore_lag)
+    domains = {
+        name: {
+            "newest_data": d["newest"],
+            "last_write": d["last_write"],
+            "entries": d["entries"],
+            "rows": d["rows"],
+            "lagging": len(d["lagging"]),
+        }
+        for name, d in s["domains"].items()
+    }
+    doc = {
+        "generated_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
+        "schema_version": s["schema_version"],
+        # Flat map for dead-simple polling: domain -> newest data date.
+        "newest_data": {name: d["newest_data"] for name, d in domains.items()},
+        "domains": domains,
+    }
+    if last_run is not None:
+        doc["last_run"] = last_run
+    return doc
+
+
+def write_status_file(last_run: Optional[dict] = None) -> str:
+    """Rebuild status.json from the current manifest, atomically. Called at the end
+    of a producer run so pollers see a consistent snapshot."""
+    doc = build_status_doc(store.load_manifest(), last_run=last_run,
+                           ignore_lag=hist_code_names())
+    path = status_path()
+    tmp = path.with_suffix(".json.tmp")
+    tmp.parent.mkdir(parents=True, exist_ok=True)
+    tmp.write_text(json.dumps(doc, indent=2, sort_keys=True))
+    os.replace(tmp, path)
+    return str(path)
+
+
+def run_summary(kind: str, ok: list, failed: list, total_rows: int,
+                seconds: float, newest: Optional[str] = None) -> str:
+    """Footer printed at the end of a producer run (also emailable later)."""
+    n = len(ok) + len(failed)
+    L = ["-" * 64,
+         f"{kind}: {len(ok)}/{n} OK · {len(failed)} failed · "
+         f"{total_rows:,} rows · {seconds:.0f}s"
+         + (f" · newest {newest}" if newest else "")]
+    if failed:
+        for sym, err in failed:
+            L.append(f"  ✗ {sym}: {str(err)[:80]}")
+    return "\n".join(L)
diff --git a/src/cotdata/store.py b/src/cotdata/store.py
index ff3fd4a..f012b42 100644
--- a/src/cotdata/store.py
+++ b/src/cotdata/store.py
@@ -122,7 +122,51 @@ def _touch_manifest(kind: str, name: str, df: pd.DataFrame, source: str) -> None
         "updated_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
     }
     m["schema_version"] = config.SCHEMA_VERSION
+    _write_manifest(m)
+
+
+def _write_manifest(m: dict) -> None:
     tmp = config.manifest_path().with_suffix(".json.tmp")
     tmp.parent.mkdir(parents=True, exist_ok=True)
     tmp.write_text(json.dumps(m, indent=2, sort_keys=True))
     os.replace(tmp, config.manifest_path())
+
+
+# domain -> the directory holding its {name}.parquet files
+_DOMAIN_DIRS = {
+    "prices": config.prices_dir,
+    "metadata": config.metadata_dir,
+    "cot_legacy": config.cot_legacy_dir,
+    "cot_disagg": config.cot_disagg_dir,
+    "cot_tff": config.cot_tff_dir,
+}
+
+
+def _domain_dir(domain: str) -> Path:
+    fn = _DOMAIN_DIRS.get(domain)
+    return fn() if fn else (config.store_root() / domain)  # unknown/dead domain
+
+
+def reconcile_manifest() -> dict:
+    """Prune manifest entries whose parquet file is missing — ghosts left by old
+    naming schemes (bare CFTC codes before the ``{symbol}_{code}`` convention, the
+    retired ``cot`` domain, …) — and drop domains left empty. Returns
+    ``{domain: [pruned names]}``.
+
+    Provably safe: only removes bookkeeping for files that do not exist on disk;
+    never deletes or renames data.
+    """
+    m = load_manifest()
+    pruned: dict = {}
+    for domain in [k for k, v in m.items() if isinstance(v, dict)]:
+        d = _domain_dir(domain)
+        gone = [name for name in m[domain] if not (d / f"{name}.parquet").exists()]
+        if gone:
+            for name in gone:
+                del m[domain][name]
+            pruned[domain] = sorted(gone)
+        if not m[domain]:
+            del m[domain]
+    if pruned:
+        _write_manifest(m)
+    return pruned
diff --git a/src/cotdata/update.py b/src/cotdata/update.py
index d80bece..22c4f33 100644
--- a/src/cotdata/update.py
+++ b/src/cotdata/update.py
@@ -1,8 +1,11 @@
 """Producer CLI:  cotdata-update --prices --symbols ES NQ
-                  cotdata-update --cot
+                  cotdata-update --cot-all
+                  cotdata-update --check        # read-only store status
+                  cotdata-update --reconcile    # prune stale manifest ghosts
 Writes to $COTDATA_STORE. Schedule prices nightly (after the Norgate Data
 Updater) and COT weekly (Friday, after the CFTC release)."""
 import argparse
+import datetime as _dt
 
 from . import config
 
@@ -19,33 +22,76 @@ def main() -> None:
     p.add_argument("--full", action="store_true",
                    help="Full rebuild of reconstructed volume (ignore the incremental "
                         "60-day window). Use after a reconstruction-logic change.")
+    p.add_argument("--check", action="store_true",
+                   help="Print store status (row counts, newest data, staleness) from "
+                        "the manifest and exit. Read-only, cross-platform, no network.")
+    p.add_argument("--reconcile", action="store_true",
+                   help="Prune manifest entries whose parquet file is missing (ghosts "
+                        "from old naming), refresh status.json, and exit. Never touches data.")
     args = p.parse_args()
 
     config.store_root()  # fail fast if COTDATA_STORE unset
-    
+
+    if args.check:
+        from . import status
+        status.print_check()
+        return
+
+    if args.reconcile:
+        from . import store, status
+        pruned = store.reconcile_manifest()
+        if not pruned:
+            print("manifest reconcile: nothing to prune (all entries have files).")
+        else:
+            total = sum(len(v) for v in pruned.values())
+            print(f"manifest reconcile: pruned {total} ghost entr{'y' if total == 1 else 'ies'} "
+                  f"with no parquet file:")
+            for domain, names in sorted(pruned.items()):
+                print(f"  {domain}: {len(names)} removed — {', '.join(names[:8])}"
+                      + (f", … (+{len(names) - 8})" if len(names) > 8 else ""))
+            status.write_status_file(last_run={"kinds": ["reconcile"],
+                                               "at": _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z"})
+        return
+
     if not (args.prices or args.metadata or args.cot_legacy or args.cot_disagg or args.cot_tff or args.cot_all):
-        p.error("nothing to do — pass --prices, --metadata, --cot-legacy, --cot-disagg, --cot-tff, or --cot-all")
+        p.error("nothing to do — pass --check, --prices, --metadata, --cot-legacy, --cot-disagg, --cot-tff, or --cot-all")
 
     if args.prices or args.metadata:
         from .providers import norgate
-        
+
+    kinds = []
+    last_run = None
     if args.prices:
-        norgate.update(symbols=args.symbols, full=args.full)
-        
+        last_run = norgate.update(symbols=args.symbols, full=args.full)
+        kinds.append("prices")
+
     if args.metadata:
         norgate.update_metadata(symbols=args.symbols)
-        
+        kinds.append("metadata")
+
     if args.cot_legacy or args.cot_all:
         from .providers import cftc
         cftc.update()
-        
+        kinds.append("cot_legacy")
+
     if args.cot_disagg or args.cot_all:
         from .providers import cftc_disagg
         cftc_disagg.update()
-        
+        kinds.append("cot_disagg")
+
     if args.cot_tff or args.cot_all:
         from .providers import cftc_tff
         cftc_tff.update()
+        kinds.append("cot_tff")
+
+    # Structured heartbeat for downstream tools: rebuild status.json from the now-
+    # updated manifest. Pollers detect new data via newest_data[<domain>].
+    from . import status
+    run = dict(last_run or {})
+    run["kinds"] = kinds
+    run["at"] = _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z"
+    path = status.write_status_file(last_run=run)
+    print(f"status written -> {path}")
 
 if __name__ == "__main__":
     main()
diff --git a/tests/test_status.py b/tests/test_status.py
new file mode 100644
index 0000000..80a9fcc
--- /dev/null
+++ b/tests/test_status.py
@@ -0,0 +1,113 @@
+import datetime as dt
+
+from cotdata import status
+
+
+def _manifest():
+    return {
+        "schema_version": 2,
+        "prices": {
+            "ES_backadj": {"last_date": "2026-07-14", "n_rows": 100, "updated_at": "2026-07-15T10:00:00Z"},
+            "ES_unadj":   {"last_date": "2026-07-14", "n_rows": 100, "updated_at": "2026-07-15T10:00:00Z"},
+            "NQ_backadj": {"last_date": "2026-07-05", "n_rows": 50,  "updated_at": "2026-07-06T10:00:00Z"},
+        },
+        "cot_legacy": {
+            "001602": {"last_date": "2026-07-07", "n_rows": 1488, "updated_at": "2026-07-14T04:00:00Z"},
+        },
+    }
+
+
+def test_summarize_counts_and_newest():
+    s = status.summarize(_manifest(), today=dt.date(2026, 7, 15))
+    p = s["domains"]["prices"]
+    assert p["entries"] == 3
+    assert p["rows"] == 250
+    assert p["newest"] == "2026-07-14"
+    assert p["oldest"] == "2026-07-05"
+    assert p["behind_today"] == 1
+    assert s["schema_version"] == 2
+
+
+def test_summarize_flags_lagging_entry():
+    s = status.summarize(_manifest(), today=dt.date(2026, 7, 15))
+    lagging = s["domains"]["prices"]["lagging"]
+    # NQ_backadj (2026-07-05) is 9 days behind the domain newest (2026-07-14).
+    assert [name for name, _, _ in lagging] == ["NQ_backadj"]
+    assert lagging[0][2] == 9
+
+
+def test_ignore_lag_suppresses_hist_codes():
+    # RTY's predecessor code is frozen at 2018 but must NOT be flagged as lagging.
+    m = {
+        "schema_version": 2,
+        "cot_legacy": {
+            "RTY_current": {"last_date": "2026-07-07", "n_rows": 100, "updated_at": "x"},
+            "RTY_23977A":  {"last_date": "2018-06-05", "n_rows": 500, "updated_at": "x"},
+            "NQ_current":  {"last_date": "2020-01-01", "n_rows": 100, "updated_at": "x"},
+        },
+    }
+    today = dt.date(2026, 7, 15)
+    # Without the ignore set, both stale entries are flagged.
+    plain = status.summarize(m, today)["domains"]["cot_legacy"]["lagging"]
+    assert {n for n, _, _ in plain} == {"RTY_23977A", "NQ_current"}
+    # With the hist_code suppressed, only the genuinely-stale current code remains.
+    filt = status.summarize(m, today, ignore_lag={"RTY_23977A"})["domains"]["cot_legacy"]["lagging"]
+    assert {n for n, _, _ in filt} == {"NQ_current"}
+
+
+def test_empty_store_report():
+    out = status.format_report({"schema_version": 2}, root="/tmp/store", today=dt.date(2026, 7, 15))
+    assert "store is empty" in out
+
+
+def test_format_report_contains_domain_and_lag_warning():
+    out = status.format_report(_manifest(), root="/store", today=dt.date(2026, 7, 15))
+    assert "prices" in out and "829" not in out  # our synthetic totals, not the real store
+    assert "250" in out                            # prices row total
+    assert "NQ_backadj" in out                     # lag warning lists the stale entry
+    assert "schema_version 2" in out
+
+
+def test_schema_mismatch_note():
+    out = status.format_report({"schema_version": 1, "prices": {}}, today=dt.date(2026, 7, 15))
+    assert "target" in out  # warns that on-disk schema is behind the library target
+
+
+def test_build_status_doc_flat_map_and_domains():
+    doc = status.build_status_doc(_manifest(), today=dt.date(2026, 7, 15))
+    # flat newest_data map is the primary polling primitive
+    assert doc["newest_data"]["prices"] == "2026-07-14"
+    assert doc["newest_data"]["cot_legacy"] == "2026-07-07"
+    assert doc["domains"]["prices"]["rows"] == 250
+    assert doc["domains"]["prices"]["lagging"] == 1        # NQ is stale
+    assert doc["schema_version"] == 2
+    assert "generated_at" in doc
+
+
+def test_write_status_file_roundtrip(tmp_path, monkeypatch):
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    from cotdata import store, status as st
+    import json
+    # seed the store manifest via a real write
+    idx = __import__("pandas").date_range("2026-07-10", periods=3, freq="D", name="Date")
+    df = __import__("pandas").DataFrame({"Open": [1, 2, 3], "High": [1, 2, 3], "Low": [1, 2, 3],
+                                         "Close": [1, 2, 3], "Volume": [1, 2, 3],
+                                         "Open Interest": [1, 2, 3]}, index=idx)
+    store.write_prices("ES", "backadj", df, source="test")
+
+    path = st.write_status_file(last_run={"kinds": ["prices"], "ok": ["ES"], "failed": []})
+    assert path.endswith("status.json")
+    doc = json.loads((tmp_path / "status.json").read_text())
+    assert doc["newest_data"]["prices"] == "2026-07-12"
+    assert doc["last_run"]["kinds"] == ["prices"]
+    assert doc["domains"]["prices"]["entries"] == 1
+
+
+def test_run_summary_ok_and_failed():
+    line = status.run_summary("prices update", ok=["ES", "NQ"], failed=[("GC", "boom")],
+                              total_rows=1234, seconds=12.0, newest="2026-07-14")
+    assert "2/3 OK" in line
+    assert "1 failed" in line
+    assert "1,234 rows" in line
+    assert "newest 2026-07-14" in line
+    assert "✗ GC: boom" in line
diff --git a/tests/test_store.py b/tests/test_store.py
index b9c42b3..fb8c2f6 100644
--- a/tests/test_store.py
+++ b/tests/test_store.py
@@ -34,6 +34,35 @@ def test_prices_roundtrip_and_manifest(store_env):
     assert m["prices"]["ES_backadj"]["last_date"] == "2020-01-05"
 
 
+def test_reconcile_prunes_ghosts_keeps_real(store_env):
+    from cotdata import store
+    # a real cot_legacy entry — writes a parquet file AND a manifest entry
+    idx = pd.date_range("2026-01-06", periods=3, freq="7D", name="Report_Date")
+    store.write_cot_legacy("ES_001602", pd.DataFrame({"Open_Interest_All": [1, 2, 3]}, index=idx), source="test")
+
+    # inject a bare-code ghost (no file) and a retired 'cot' domain (no dir/files)
+    m = store.load_manifest()
+    m["cot_legacy"]["001602"] = {"last_date": "2018-06-05", "n_rows": 10, "source": "cftc", "updated_at": "x"}
+    m["cot"] = {"099741": {"last_date": "2018-01-01", "n_rows": 5, "source": "cftc", "updated_at": "x"}}
+    store._write_manifest(m)
+
+    pruned = store.reconcile_manifest()
+    assert pruned["cot_legacy"] == ["001602"]   # bare ghost pruned
+    assert "cot" in pruned                       # dead domain pruned
+
+    m2 = store.load_manifest()
+    assert "ES_001602" in m2["cot_legacy"]       # real prefixed entry kept
+    assert "001602" not in m2["cot_legacy"]      # ghost gone
+    assert "cot" not in m2                        # emptied domain removed
+
+
+def test_reconcile_noop_when_clean(store_env):
+    from cotdata import store
+    idx = pd.date_range("2026-01-06", periods=2, freq="7D", name="Report_Date")
+    store.write_cot_legacy("ES_001602", pd.DataFrame({"x": [1, 2]}, index=idx), source="test")
+    assert store.reconcile_manifest() == {}      # nothing to prune
+
+
 def test_roll_dates_from_delivery_month(store_env):
     from cotdata import store, roll_dates
     store.write_prices("ES", "backadj", _sample(), source="test")