experiment-os: registry + CLI (register/list/get, immutable)

Plan 006 step 1-2. registry.apply_schema honors PG_SKIP_SCHEMA; register()
validates required keys incl. fidelity floors and refuses to overwrite an
existing id (pre-registration is immutable). CLI: python -m research.experiments
{register|run|adjudicate} with deferred imports. 6 registry tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

research/experiments/__init__.py

@@ -0,0 +1,11 @@
+"""Experiment OS (plan 006) — pre-registered, auto-adjudicated counterfactual
+experiments over logged fires.
+
+An experiment is a config row (``experiments/*.json``): a population (from
+``strategy/population.py``), a typed intervention, pre-registered pass/fail
+bars, and hard slippage-fidelity floors. The replay engine re-runs the
+intervention counterfactually over every logged fire; the adjudicator scores
+the accumulated runs against the bars and emits a verdict. ALL-SHADOW: no
+experiment result ever alters live gates/sizing/exits — promotion is a
+separate, owner-approved change.
+"""

research/experiments/__main__.py

@@ -0,0 +1,3 @@
+from research.experiments.cli import main
+
+raise SystemExit(main())

research/experiments/cli.py

@@ -0,0 +1,67 @@
+"""CLI: ``python -m research.experiments {register|run|adjudicate} [args]``.
+
+Thin wrapper. Builds ONE EventStore (its Postgres connection is shared for the
+experiment tables) and dispatches. Deferred imports so each subcommand only
+pulls what it needs. NEVER run against the production DB from a test — the
+harness's real verdict lands when the owner first schedules it.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+
+def _store():
+    from storage.event_store import EventStore
+    return EventStore(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum"))
+
+
+def main(argv=None) -> int:
+    argv = list(sys.argv[1:] if argv is None else argv)
+    if not argv:
+        print("usage: python -m research.experiments {register <json>|run|adjudicate}")
+        return 2
+    cmd, rest = argv[0], argv[1:]
+
+    if cmd == "register":
+        if not rest:
+            print("usage: python -m research.experiments register <json_path>")
+            return 2
+        from research.experiments import registry
+        store = _store()
+        try:
+            registry.apply_schema(store.con)
+            exp_id = registry.register(store.con, rest[0])
+            print(f"registered {exp_id}")
+        finally:
+            store.close()
+        return 0
+
+    if cmd == "run":
+        from research.experiments import replay
+        store = _store()
+        try:
+            n = replay.run(store)
+            print(f"experiments run: {n} new run rows")
+        finally:
+            store.close()
+        return 0
+
+    if cmd == "adjudicate":
+        from research.experiments import adjudicate
+        store = _store()
+        try:
+            verdicts = adjudicate.adjudicate(store)
+            for v in verdicts:
+                print(f"{v['experiment_id']}: {v['verdict']} (n={v['n']})")
+        finally:
+            store.close()
+        return 0
+
+    print(f"unknown command: {cmd}")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

research/experiments/registry.py

@@ -0,0 +1,109 @@
+"""Experiment registry — schema apply + immutable register/list/get.
+
+The registry owns the ``experiments`` contract table. Registration is
+one-shot and immutable: an id can never be overwritten (a pre-registered bar
+that could be re-tuned after seeing the data is not pre-registered). The
+required-keys check refuses any experiment missing its fidelity floors — the
+single guard that keeps a zero-slippage artifact (vr1/liq1 class) out.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+_SCHEMA_SQL = Path(__file__).resolve().parents[2] / "storage" / "experiments_schema.sql"
+
+# Every experiment JSON must carry these; fidelity must carry these floors.
+_REQUIRED = ("id", "hypothesis", "population", "intervention", "bars", "fidelity")
+_REQUIRED_FIDELITY = ("atr_frac", "spread_mult", "adverse_ticks")
+
+_TERMINAL = {"passed", "failed", "killed"}
+
+
+def apply_schema(con) -> None:
+    """Create the experiment tables idempotently on ``con``.
+
+    Honors ``PG_SKIP_SCHEMA`` exactly like storage/db_pg.get_connection: set it
+    when the owner provisions the DDL out-of-band and the harness should not
+    touch DDL. On an isolated test schema the connection's search_path is
+    already pinned, so the CREATEs land in that schema.
+    """
+    if os.environ.get("PG_SKIP_SCHEMA", "").strip().lower() in {"1", "true", "yes"}:
+        return
+    con.execute(_SCHEMA_SQL.read_text())
+
+
+def _validate(spec: dict) -> None:
+    missing = [k for k in _REQUIRED if k not in spec]
+    if missing:
+        raise ValueError(f"experiment missing required keys: {missing}")
+    fid = spec["fidelity"]
+    if not isinstance(fid, dict):
+        raise ValueError("fidelity must be an object with slippage floors")
+    fmissing = [k for k in _REQUIRED_FIDELITY if k not in fid]
+    if fmissing:
+        raise ValueError(f"fidelity missing required floors: {fmissing}")
+
+
+def register(con, json_path: str | Path) -> str:
+    """Register one experiment from a JSON file. Immutable: refuses to overwrite
+    an existing id. Returns the registered id."""
+    spec = json.loads(Path(json_path).read_text())
+    _validate(spec)
+    exp_id = spec["id"]
+    if get(con, exp_id) is not None:
+        raise ValueError(f"experiment {exp_id!r} already registered (immutable)")
+    con.execute(
+        "INSERT INTO experiments "
+        "(id, registered_at, hypothesis, population, intervention, bars, futility, "
+        " fidelity, status) "
+        "VALUES (?, ?, ?, ?::jsonb, ?::jsonb, ?::jsonb, ?::jsonb, ?::jsonb, 'running')",
+        [
+            exp_id,
+            spec.get("registered_at") or datetime.now(timezone.utc).isoformat(),
+            spec["hypothesis"],
+            json.dumps(spec["population"]),
+            json.dumps(spec["intervention"]),
+            json.dumps(spec["bars"]),
+            json.dumps(spec.get("futility")) if spec.get("futility") is not None else None,
+            json.dumps(spec["fidelity"]),
+        ],
+    )
+    return exp_id
+
+
+def _row_to_dict(row) -> dict:
+    return {
+        "id": row[0],
+        "registered_at": row[1],
+        "hypothesis": row[2],
+        "population": row[3],
+        "intervention": row[4],
+        "bars": row[5],
+        "futility": row[6],
+        "fidelity": row[7],
+        "status": row[8],
+        "verdict_at": row[9],
+        "verdict_note": row[10],
+    }
+
+
+_COLS = ("id, registered_at, hypothesis, population, intervention, bars, futility, "
+         "fidelity, status, verdict_at, verdict_note")
+
+
+def get(con, exp_id: str) -> dict | None:
+    row = con.execute(
+        f"SELECT {_COLS} FROM experiments WHERE id = ?", [exp_id]).fetchone()
+    return _row_to_dict(row) if row else None
+
+
+def list_experiments(con, running_only: bool = False) -> list[dict]:
+    sql = f"SELECT {_COLS} FROM experiments"
+    if running_only:
+        sql += " WHERE status = 'running'"
+    sql += " ORDER BY registered_at"
+    return [_row_to_dict(r) for r in con.execute(sql).fetchall()]

tests/unit/experiments/conftest.py

@@ -0,0 +1,44 @@
+"""Shared fixtures for the experiment-OS tests.
+
+Every test runs against an ISOLATED throwaway Postgres schema (EventStore
+":memory:") with the experiment tables applied on top — never the production
+DB (mirrors tests/conftest.py isolation; the mem_* schema is swept at session
+end).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from research.experiments import registry
+from storage.event_store import EventStore
+
+
+@pytest.fixture
+def store():
+    s = EventStore(":memory:")
+    registry.apply_schema(s.con)
+    yield s
+    s.close()
+
+
+def write_spec(tmp_path, spec: dict) -> str:
+    p = tmp_path / f"{spec['id']}.json"
+    p.write_text(json.dumps(spec))
+    return str(p)
+
+
+def base_spec(**over) -> dict:
+    spec = {
+        "id": "demo_v1",
+        "hypothesis": "demo hypothesis",
+        "population": {"min_entry_gap_pct": 0.10},
+        "intervention": {"kind": "time_stop", "minutes": 10, "confirm": "new_high_or_0.5R"},
+        "bars": {"expR_delta_min": 0.05, "min_n": 300},
+        "futility": {"min_n": 500, "expR_delta_max": -0.02},
+        "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1},
+    }
+    spec.update(over)
+    return spec

tests/unit/experiments/test_registry.py

@@ -0,0 +1,49 @@
+"""Registry: schema apply, immutability, required-key/fidelity validation."""
+
+import pytest
+
+from research.experiments import registry
+from tests.unit.experiments.conftest import base_spec, write_spec
+
+
+def test_tables_exist(store):
+    # apply_schema (fixture) created all three; a select proves it.
+    for tbl in ("experiments", "experiment_runs", "experiment_verdicts"):
+        assert store.con.execute(f"SELECT count(*) FROM {tbl}").fetchone()[0] == 0
+
+
+def test_register_then_get(store, tmp_path):
+    exp_id = registry.register(store.con, write_spec(tmp_path, base_spec()))
+    assert exp_id == "demo_v1"
+    got = registry.get(store.con, "demo_v1")
+    assert got["hypothesis"] == "demo hypothesis"
+    assert got["population"] == {"min_entry_gap_pct": 0.10}   # jsonb round-trips to dict
+    assert got["status"] == "running"
+
+
+def test_immutable_refuses_overwrite(store, tmp_path):
+    path = write_spec(tmp_path, base_spec())
+    registry.register(store.con, path)
+    with pytest.raises(ValueError, match="already registered"):
+        registry.register(store.con, path)
+
+
+def test_missing_fidelity_refused(store, tmp_path):
+    spec = base_spec()
+    del spec["fidelity"]
+    with pytest.raises(ValueError, match="required keys"):
+        registry.register(store.con, write_spec(tmp_path, spec))
+
+
+def test_incomplete_fidelity_floors_refused(store, tmp_path):
+    spec = base_spec(fidelity={"atr_frac": 0.25})   # missing spread_mult + adverse_ticks
+    with pytest.raises(ValueError, match="fidelity missing"):
+        registry.register(store.con, write_spec(tmp_path, spec))
+
+
+def test_list_running_only(store, tmp_path):
+    registry.register(store.con, write_spec(tmp_path, base_spec(id="a_v1")))
+    registry.register(store.con, write_spec(tmp_path, base_spec(id="b_v1")))
+    store.con.execute("UPDATE experiments SET status='failed' WHERE id='b_v1'")
+    running = registry.list_experiments(store.con, running_only=True)
+    assert [e["id"] for e in running] == ["a_v1"]