# Deterministic Forward State (Sub-project A) Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Replace the mutable, hand-editable paper-tracker JSON state with an immutable `(T0, definition_hash)` anchor in Postgres; the forward record is either recomputed from point-in-time warehouse data (`recompute` tracks) or an append-only observed log (`observed` tracks); definition changes auto-re-inception.

**Architecture:** A per-strategy `forward_anchor` row is the only persistent per-track state. Each nightly run computes the strategy's `definition_hash`; a changed hash archives the old anchor and inserts a new one (T0=today). `recompute` tracks derive `[T0, today]` from `advance(None, {})` filtered to `date >= T0`; `observed` tracks append today's booked row to an append-only `forward_record` log plus a minimal carried book-state. `forward_nav`/`forward_summary` become derived caches. A one-time migration seeds anchors from the existing `*_state.json` files, reconciling recompute tracks and importing observed logs.

**Tech Stack:** Python 3.12, SQLAlchemy 2.0 ORM (SQLite for tests via `sqlite://` + StaticPool, PostgreSQL/TimescaleDB in prod), Dagster assets, pytest (`uv run pytest`), Typer CLI.

## Global Constraints

- Spec: `docs/superpowers/specs/2026-07-08-deterministic-forward-state-design.md` (authoritative).
- Real money: the gate must err toward **false-WAIT, never false-PASS** — insufficient/absent data → the summary stays `WAIT`, never fabricates a PASS.
- SSOT: the DB anchor + record are the single source of truth; the PVC `*_state.json` path is retired (Task 9), not left as a shadow store.
- Idempotency: every DB write is an upsert (`Session.merge`) — re-running any run/asset is safe and byte-identical.
- Determinism: `recompute` tracks read only point-in-time (PIT) inputs; a recompute run twice yields byte-identical rows. `Date.now()`-style wall-clock enters only via an injected `today` parameter (never read inside pure functions), so tests are deterministic.
- Persistence pattern: new tables are ORM rows on `CockpitBase` (`src/fxhnt/adapters/persistence/cockpit_models.py`); `ForwardNavRepo.migrate()` (`src/fxhnt/adapters/persistence/forward_nav.py`) creates them via `CockpitBase.metadata.create_all`. Repo methods use `Session(self._engine)` + `s.merge(...)` + `s.commit()`.
- Never save working files/tests to the repo root; tests live under `tests/` mirroring the existing layout (`tests/unit/`, `tests/integration/`).
- Commit after every task (frequent commits). Commit messages end with the `Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>` trailer.
- `record_mode` map (from spec §3b): `observed` = {`xsfunding`, `combined`}; every other active track is `recompute`.

---

### Task 1: `forward_anchor` table + repo methods

The immutable anchor: one active row per strategy (`archived_at IS NULL`), full lineage kept.

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (add `StrategyAnchorRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (add anchor repo methods)
- Test: `tests/integration/test_forward_anchor.py`

**Interfaces:**
- Consumes: `CockpitBase`, `ForwardNavRepo(dsn)`, `ForwardNavRepo.migrate()`.
- Produces:
  - `StrategyAnchorRow(strategy_id, created_at, t0, definition_hash, definition_version, params_json, record_mode, archived_at)`.
  - `ForwardNavRepo.active_anchor(strategy_id: str) -> StrategyAnchorRow | None`
  - `ForwardNavRepo.set_anchor(strategy_id: str, t0: str, definition_hash: str, definition_version: int, params_json: dict, record_mode: str, at: dt.datetime) -> None`
  - `ForwardNavRepo.archive_anchor(strategy_id: str, at: dt.datetime) -> None`

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_forward_anchor.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/anchor.db")
    repo.migrate()
    return repo


def test_set_and_read_active_anchor(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 8, 23, 30)
    repo.set_anchor("multistrat", t0="2026-07-08", definition_hash="h1",
                    definition_version=1, params_json={"instruments": ["SPY"]},
                    record_mode="recompute", at=at)
    a = repo.active_anchor("multistrat")
    assert a is not None
    assert (a.t0, a.definition_hash, a.record_mode, a.archived_at) == ("2026-07-08", "h1", "recompute", None)


def test_archive_then_reset_keeps_lineage_and_flips_active(tmp_path):
    repo = _repo(tmp_path)
    t1 = dt.datetime(2026, 7, 1, 23, 30)
    t2 = dt.datetime(2026, 7, 8, 23, 30)
    repo.set_anchor("multistrat", "2026-07-01", "h1", 1, {"n": 6}, "recompute", at=t1)
    # re-inception: archive the old anchor, set a new one
    repo.archive_anchor("multistrat", at=t2)
    repo.set_anchor("multistrat", "2026-07-08", "h2", 1, {"n": 5}, "recompute", at=t2)

    a = repo.active_anchor("multistrat")
    assert a.definition_hash == "h2" and a.t0 == "2026-07-08" and a.archived_at is None
    # lineage preserved: the old row still exists, archived
    with __import__("sqlalchemy.orm", fromlist=["Session"]).Session(repo._engine) as s:
        from fxhnt.adapters.persistence.cockpit_models import StrategyAnchorRow
        rows = s.query(StrategyAnchorRow).filter_by(strategy_id="multistrat").all()
        assert len(rows) == 2
        assert sum(1 for r in rows if r.archived_at is None) == 1


def test_active_anchor_absent_returns_none(tmp_path):
    assert _repo(tmp_path).active_anchor("nope") is None
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_forward_anchor.py -q`
Expected: FAIL with `AttributeError: 'ForwardNavRepo' object has no attribute 'set_anchor'` (and an `ImportError` for `StrategyAnchorRow`).

- [ ] **Step 3: Add the ORM row**

In `src/fxhnt/adapters/persistence/cockpit_models.py`, append after `StrategyPromotionRow`:

```python
class StrategyAnchorRow(CockpitBase):
    """Immutable per-track anchor: the ONLY persistent state for a forward track (spec sub-project A).
    (strategy_id, created_at) PK keeps full lineage — a re-inception archives the old row (sets archived_at)
    and inserts a new one. The ACTIVE anchor is the row with archived_at IS NULL."""
    __tablename__ = "forward_anchor"
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    created_at: Mapped[dt.datetime] = mapped_column(DateTime, primary_key=True)
    t0: Mapped[str] = mapped_column(String(10))                       # frozen inception (ISO date) = OOS clock start
    definition_hash: Mapped[str] = mapped_column(String(64))
    definition_version: Mapped[int] = mapped_column(Integer)
    params_json: Mapped[dict[str, Any]] = mapped_column(JSON)         # audit of the hashed return-affecting params
    record_mode: Mapped[str] = mapped_column(String(12))             # "recompute" | "observed"
    archived_at: Mapped[dt.datetime | None] = mapped_column(DateTime, nullable=True, default=None)
```

- [ ] **Step 4: Add the repo methods**

In `src/fxhnt/adapters/persistence/forward_nav.py`, add `StrategyAnchorRow` to the `cockpit_models` import block, then add these methods to `ForwardNavRepo` (after `registry`):

```python
    def active_anchor(self, strategy_id: str) -> StrategyAnchorRow | None:
        stmt = (select(StrategyAnchorRow)
                .where(StrategyAnchorRow.strategy_id == strategy_id,
                       StrategyAnchorRow.archived_at.is_(None))
                .order_by(StrategyAnchorRow.created_at.desc()))
        with Session(self._engine) as s:
            return s.scalars(stmt).first()

    def set_anchor(self, strategy_id: str, t0: str, definition_hash: str, definition_version: int,
                   params_json: dict[str, Any], record_mode: str, at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(StrategyAnchorRow(
                strategy_id=strategy_id, created_at=at, t0=t0, definition_hash=definition_hash,
                definition_version=definition_version, params_json=params_json,
                record_mode=record_mode, archived_at=None))
            s.commit()

    def archive_anchor(self, strategy_id: str, at: dt.datetime) -> None:
        """Mark the currently-active anchor archived. No-op if none is active."""
        with Session(self._engine) as s:
            for row in s.query(StrategyAnchorRow).filter(
                    StrategyAnchorRow.strategy_id == strategy_id,
                    StrategyAnchorRow.archived_at.is_(None)):
                row.archived_at = at
            s.commit()
```

- [ ] **Step 5: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_forward_anchor.py -q`
Expected: PASS (3 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_forward_anchor.py
git commit -F - <<'EOF'
feat(forward): forward_anchor table + repo (immutable per-track anchor, lineage)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 2: `ForwardDefinition` + `definition_hash` + registry entries

The declared return-affecting definition and its hash (drives auto-re-inception).

**Files:**
- Create: `src/fxhnt/application/forward_definition.py`
- Modify: `src/fxhnt/registry.py` (add a `definition` entry per active track)
- Test: `tests/unit/test_forward_definition.py`

**Interfaces:**
- Produces:
  - `definition_hash(params: dict, version: int) -> str` — sha256 of canonical (key-sorted) JSON of `params` plus `str(version)`.
  - `track_definition(strategy_id: str) -> tuple[dict, int, str]` — returns `(params, version, record_mode)` from `STRATEGY_REGISTRY[strategy_id]["definition"]`.
  - Registry: each active track gains `"definition": {"params": {...}, "version": int, "record_mode": "recompute"|"observed"}`.

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_forward_definition.py
import pytest

from fxhnt.application.forward_definition import definition_hash, track_definition
from fxhnt.registry import STRATEGY_REGISTRY


def test_hash_is_stable_under_key_reordering():
    assert definition_hash({"a": 1, "b": 2}, 1) == definition_hash({"b": 2, "a": 1}, 1)


def test_hash_changes_when_a_param_changes():
    # the BTC-drop case: instruments 6 -> 5 must change the hash
    six = definition_hash({"instruments": ["SPY", "IEF", "GLD", "PDBC", "DBMF", "BTC-USD"]}, 1)
    five = definition_hash({"instruments": ["SPY", "IEF", "GLD", "PDBC", "DBMF"]}, 1)
    assert six != five


def test_hash_changes_when_version_bumps():
    assert definition_hash({"a": 1}, 1) != definition_hash({"a": 1}, 2)


def test_every_active_track_declares_a_definition():
    for sid, e in STRATEGY_REGISTRY.items():
        params, version, mode = track_definition(sid)
        assert isinstance(params, dict) and isinstance(version, int)
        assert mode in ("recompute", "observed")


def test_observed_modes_are_exactly_xsfunding_and_combined():
    observed = {sid for sid in STRATEGY_REGISTRY if track_definition(sid)[2] == "observed"}
    assert observed == {"xsfunding", "combined"}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_forward_definition.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.forward_definition'`.

- [ ] **Step 3: Create the hash helper**

```python
# src/fxhnt/application/forward_definition.py
"""The declared, return-affecting definition of a forward track and its content hash.

A CHANGE to the hash means the strategy's return-generating definition changed → the forward engine
auto-re-inceptions (spec sub-project A §2/§3). The hash covers the declared params (instrument list,
cost bps, universe key, sleeve set, key constants) PLUS an explicit `version` int the dev bumps for
pure-code changes the params don't capture. It never covers comments/refactors, so it never spuriously
re-inceptions; erring toward extra re-inception (a changed strategy re-proving itself OOS) is the safe
direction for real money."""
from __future__ import annotations

import hashlib
import json

from fxhnt.registry import STRATEGY_REGISTRY


def definition_hash(params: dict, version: int) -> str:
    canonical = json.dumps(params, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(f"{canonical}|v{version}".encode()).hexdigest()


def track_definition(strategy_id: str) -> tuple[dict, int, str]:
    """Return (params, version, record_mode) for a track. Raises KeyError if the track has no `definition`
    (a plan/registry error — every tracked strategy must declare one)."""
    d = STRATEGY_REGISTRY[strategy_id]["definition"]
    return d["params"], int(d["version"]), d["record_mode"]
```

- [ ] **Step 4: Add a `definition` to each active registry track**

In `src/fxhnt/registry.py`, add a `"definition"` key to each track's dict. Use the exact return-affecting params below (params can be any JSON-serialisable summary; what matters is that it changes iff the return series changes). `record_mode` is `observed` for `combined` and `xsfunding`, `recompute` otherwise.

```python
# combined
"definition": {"params": {"venue": "binance+glbx", "kill_dd": "settings", "reenter_dd": "settings"},
               "version": 1, "record_mode": "observed"},
# sixtyforty
"definition": {"params": {"weights": [["SPY", 0.6], ["IEF", 0.4]], "source": "tiingo-adjclose"},
               "version": 1, "record_mode": "recompute"},
# xsfunding
"definition": {"params": {"construction": "xs-funding-dispersion-delta-neutral", "venue": "binance-perp"},
               "version": 1, "record_mode": "observed"},
# unlock
"definition": {"params": {"edge": "unlock-dilution-short", "calendar": "defillama", "prices": "binance-perp"},
               "version": 1, "record_mode": "recompute"},
# crypto_tstrend
"definition": {"params": {"lookbacks": [20, 60, 120], "book_target_vol": 0.15, "panel": "warehouse-close"},
               "version": 1, "record_mode": "recompute"},
# multistrat
"definition": {"params": {"instruments": ["SPY", "IEF", "GLD", "PDBC", "DBMF"], "maxlev": 1.0,
                          "source": "yahoo-pit"}, "version": 1, "record_mode": "recompute"},
# stablecoin_rotation
"definition": {"params": {"universe": ["FDUSDUSDT", "TUSDUSDT", "USDPUSDT"], "venue": "binance-spot"},
               "version": 1, "record_mode": "recompute"},
# bybit_4edge
"definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"], "weighting": "naive-eq",
                          "cost_bps": 5.5}, "version": 1, "record_mode": "recompute"},
# bybit_4edge_levered
"definition": {"params": {"sleeves": ["tstrend", "unlock", "xsfunding", "positioning"],
                          "weighting": "carry-static", "gross_max": 1.5, "cost_bps": 5.5},
               "version": 1, "record_mode": "recompute"},
# positioning
"definition": {"params": {"sleeve": "positioning", "cost_bps": 5.5}, "version": 1, "record_mode": "recompute"},
```

- [ ] **Step 5: Run tests to verify they pass**

Run: `uv run pytest tests/unit/test_forward_definition.py -q`
Expected: PASS (5 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/forward_definition.py src/fxhnt/registry.py tests/unit/test_forward_definition.py
git commit -F - <<'EOF'
feat(forward): declared definition + content hash + per-track definition entries

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 3: `forward_record` append-only log + book-state (observed mode)

The immutable observed series + the minimal carried book-state observed tracks need to book the next day.

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (add `ForwardRecordRow`, `ForwardBookStateRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (add record/book-state methods)
- Test: `tests/integration/test_forward_record.py`

**Interfaces:**
- Produces:
  - `ForwardRecordRow(strategy_id, date, ret, booked_at)` — append-only, idempotent on `(strategy_id, date)`.
  - `ForwardBookStateRow(strategy_id, extra_json, updated_at)` — the carried `extra` (positions) for observed tracks.
  - `ForwardNavRepo.append_record(strategy_id, date, ret, at)`
  - `ForwardNavRepo.record_series(strategy_id, since: str | None = None) -> list[tuple[str, float]]`
  - `ForwardNavRepo.get_book_state(strategy_id) -> dict`
  - `ForwardNavRepo.set_book_state(strategy_id, extra: dict, at)`

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_forward_record.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/rec.db")
    repo.migrate()
    return repo


def test_append_is_idempotent_per_date(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 8, 23, 30)
    repo.append_record("xsfunding", "2026-07-08", 0.001, at)
    repo.append_record("xsfunding", "2026-07-08", 0.001, at)   # same date re-run → no double append
    assert repo.record_series("xsfunding") == [("2026-07-08", 0.001)]


def test_record_series_filters_since(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 8, 23, 30)
    for d, r in [("2026-07-06", 0.001), ("2026-07-07", 0.002), ("2026-07-08", 0.003)]:
        repo.append_record("xsfunding", d, r, at)
    assert repo.record_series("xsfunding", since="2026-07-07") == [("2026-07-07", 0.002), ("2026-07-08", 0.003)]


def test_book_state_round_trip(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 8, 23, 30)
    assert repo.get_book_state("xsfunding") == {}
    repo.set_book_state("xsfunding", {"positions": {"BTCUSDT": 0.5}}, at)
    assert repo.get_book_state("xsfunding") == {"positions": {"BTCUSDT": 0.5}}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_forward_record.py -q`
Expected: FAIL with `AttributeError: 'ForwardNavRepo' object has no attribute 'append_record'`.

- [ ] **Step 3: Add the ORM rows**

In `src/fxhnt/adapters/persistence/cockpit_models.py`, append:

```python
class ForwardRecordRow(CockpitBase):
    """Append-only, immutable observed forward return for one `observed`-mode track on one date (spec §3b).
    (strategy_id, date) PK → idempotent per date (a same-day re-run does not double-append). Reused by
    sub-project C for execution fills. Never edited in place — a re-inception starts a fresh log after
    the anchor's new T0."""
    __tablename__ = "forward_record"
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    date: Mapped[str] = mapped_column(String(10), primary_key=True)
    ret: Mapped[float] = mapped_column(Float)
    booked_at: Mapped[dt.datetime] = mapped_column(DateTime)


class ForwardBookStateRow(CockpitBase):
    """The minimal carried book-state (e.g. current delta-neutral positions) an `observed` track needs to
    book the next day. One row per track, replaced each run — NOT hand-editable JSON on a PVC. Not a record
    of returns (that is forward_record); just the live carry."""
    __tablename__ = "forward_book_state"
    strategy_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    extra_json: Mapped[dict[str, Any]] = mapped_column(JSON)
    updated_at: Mapped[dt.datetime] = mapped_column(DateTime)
```

- [ ] **Step 4: Add the repo methods**

Add `ForwardRecordRow, ForwardBookStateRow` to the `cockpit_models` import in `forward_nav.py`, then add to `ForwardNavRepo`:

```python
    def append_record(self, strategy_id: str, date: str, ret: float, at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(ForwardRecordRow(strategy_id=strategy_id, date=date, ret=ret, booked_at=at))
            s.commit()

    def record_series(self, strategy_id: str, since: str | None = None) -> list[tuple[str, float]]:
        stmt = (select(ForwardRecordRow).where(ForwardRecordRow.strategy_id == strategy_id)
                .order_by(ForwardRecordRow.date))
        if since is not None:
            stmt = stmt.where(ForwardRecordRow.date >= since)
        with Session(self._engine) as s:
            return [(r.date, r.ret) for r in s.scalars(stmt)]

    def get_book_state(self, strategy_id: str) -> dict[str, Any]:
        with Session(self._engine) as s:
            row = s.get(ForwardBookStateRow, strategy_id)
            return dict(row.extra_json) if row is not None else {}

    def set_book_state(self, strategy_id: str, extra: dict[str, Any], at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(ForwardBookStateRow(strategy_id=strategy_id, extra_json=extra, updated_at=at))
            s.commit()
```

- [ ] **Step 5: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_forward_record.py -q`
Expected: PASS (3 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py tests/integration/test_forward_record.py
git commit -F - <<'EOF'
feat(forward): append-only forward_record log + carried book-state (observed mode)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 4: Recompute + nav-derivation (pure functions)

The pure record-derivation used by `recompute` tracks and by nav/summary projection for both modes.

**Files:**
- Create: `src/fxhnt/application/forward_recompute.py`
- Test: `tests/unit/test_forward_recompute.py`

**Interfaces:**
- Consumes: `ForwardStrategy.advance(None, {})` (from `fxhnt.application.forward_tracker.ForwardStrategy`), `per_period_sharpe` (`fxhnt.domain.gauntlet`).
- Produces:
  - `recompute_series(strategy, t0: str) -> list[tuple[str, float]]` — `advance(None, {})` rows sorted, filtered to `date >= t0`.
  - `derive_nav(rows: list[tuple[str, float]]) -> tuple[list[tuple[str, float]], dict]` — returns `(nav_rows, summary)` where `nav_rows` is `[(date, nav)]` and `summary` is `{"days","nav","total_return","sharpe","maxdd","last_date"}`. Uses the SAME formulae as `ForwardTracker.step` (annualised √365 sharpe when ≥3 rows else 0; peak-to-trough maxdd).

- [ ] **Step 1: Write the failing test**

```python
# tests/unit/test_forward_recompute.py
import numpy as np

from fxhnt.application.forward_recompute import derive_nav, recompute_series


class _FakeStrategy:
    def __init__(self, series):
        self._series = series

    def advance(self, last_date, extra):
        return list(self._series), extra


def test_recompute_filters_to_since_t0():
    strat = _FakeStrategy([("2026-07-06", 0.01), ("2026-07-07", 0.02), ("2026-07-08", -0.01)])
    assert recompute_series(strat, "2026-07-07") == [("2026-07-07", 0.02), ("2026-07-08", -0.01)]


def test_recompute_is_deterministic():
    strat = _FakeStrategy([("2026-07-07", 0.02), ("2026-07-06", 0.01)])  # unsorted input
    a = recompute_series(strat, "2026-07-01")
    b = recompute_series(strat, "2026-07-01")
    assert a == b == [("2026-07-06", 0.01), ("2026-07-07", 0.02)]        # sorted, both runs identical


def test_derive_nav_matches_forward_tracker_formula():
    rows = [("2026-07-06", 0.01), ("2026-07-07", 0.02), ("2026-07-08", -0.01)]
    nav_rows, summary = derive_nav(rows)
    # nav compounds
    assert abs(nav_rows[-1][1] - (1.01 * 1.02 * 0.99)) < 1e-12
    assert summary["days"] == 3
    assert abs(summary["nav"] - 1.01 * 1.02 * 0.99) < 1e-12
    # sharpe uses per_period_sharpe * sqrt(365) for >=3 rows
    from fxhnt.domain.gauntlet import per_period_sharpe
    exp = float(per_period_sharpe(np.array([0.01, 0.02, -0.01])) * np.sqrt(365.0))
    assert abs(summary["sharpe"] - exp) < 1e-9


def test_derive_nav_empty_is_wait_safe():
    nav_rows, summary = derive_nav([])
    assert nav_rows == [] and summary["days"] == 0 and summary["nav"] == 1.0 and summary["last_date"] is None
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/unit/test_forward_recompute.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.forward_recompute'`.

- [ ] **Step 3: Implement the pure functions**

```python
# src/fxhnt/application/forward_recompute.py
"""Pure record-derivation for the deterministic forward engine (spec sub-project A).

`recompute_series` realises the recompute-from-anchor primitive: a recompute-mode strategy's full-history
series (`advance(None, {})`) filtered to dates on/after the frozen T0. `derive_nav` projects any (date, ret)
series into the nav/summary the cockpit + gate read, using the SAME formulae as the legacy
`ForwardTracker.step` so the migration reconciles byte-for-byte."""
from __future__ import annotations

from typing import Any, Protocol

import numpy as np

from fxhnt.domain.gauntlet import per_period_sharpe

_ANN = np.sqrt(365.0)


class _Strategy(Protocol):
    def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
        ...


def recompute_series(strategy: _Strategy, t0: str) -> list[tuple[str, float]]:
    rows, _ = strategy.advance(None, {})
    rows = sorted(((str(d), float(r)) for d, r in rows), key=lambda x: x[0])
    return [(d, r) for d, r in rows if d >= t0]


def derive_nav(rows: list[tuple[str, float]]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
    nav = 1.0
    nav_rows: list[tuple[str, float]] = []
    rets: list[float] = []
    for d, r in rows:
        nav *= (1.0 + r)
        nav_rows.append((d, nav))
        rets.append(r)
    if not rows:
        return [], {"days": 0, "nav": 1.0, "total_return": 0.0, "sharpe": 0.0, "maxdd": 0.0, "last_date": None}
    arr = np.array(rets)
    navs = np.array([n for _, n in nav_rows])
    sharpe = float(per_period_sharpe(arr) * _ANN) if len(arr) >= 3 else 0.0
    maxdd = float((navs / np.maximum.accumulate(navs) - 1.0).min())
    return nav_rows, {"days": len(rows), "nav": nav, "total_return": nav - 1.0,
                      "sharpe": sharpe, "maxdd": maxdd, "last_date": rows[-1][0]}
```

- [ ] **Step 4: Run tests to verify they pass**

Run: `uv run pytest tests/unit/test_forward_recompute.py -q`
Expected: PASS (4 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/forward_recompute.py tests/unit/test_forward_recompute.py
git commit -F - <<'EOF'
feat(forward): pure recompute_series + derive_nav (matches ForwardTracker formula)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 5: The forward engine (inception / re-inception / mode dispatch)

Ties the anchor, definition, recompute and observed-log together into one run.

**Files:**
- Create: `src/fxhnt/application/forward_engine.py`
- Test: `tests/integration/test_forward_engine.py`

**Interfaces:**
- Consumes: `ForwardNavRepo` (Tasks 1+3), `definition_hash`/`track_definition` (Task 2), `recompute_series`/`derive_nav` (Task 4).
- Produces:
  - `run_track(repo, strategy_id, strategy, today: str, at: dt.datetime) -> dict` — runs one track end to end and returns `{"t0","record_mode","reinception": bool, **summary}`. Reads the track's definition from the registry, manages the anchor, produces the record per mode, and writes nav + summary via existing repo methods.

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_forward_engine.py
import datetime as dt

import pytest

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_engine import run_track


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/eng.db")
    repo.migrate()
    return repo


class _RecomputeStrat:
    """A recompute strategy: full series every call (independent of last_date)."""
    def __init__(self, series):
        self._series = series

    def advance(self, last_date, extra):
        return list(self._series), extra


def test_first_run_inceptions_at_today(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    # register a throwaway recompute track definition
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "t_rec",
                        {"display_name": "T", "sleeve": "x", "gate_spec": {},
                         "definition": {"params": {"v": 1}, "version": 1, "record_mode": "recompute"}})
    strat = _RecomputeStrat([("2026-07-06", 0.01), ("2026-07-07", 0.02), ("2026-07-08", 0.03)])
    out = run_track(repo, "t_rec", strat, today="2026-07-08", at=dt.datetime(2026, 7, 8, 23, 30))
    a = repo.active_anchor("t_rec")
    assert a.t0 == "2026-07-08" and out["reinception"] is True
    # recompute filtered to >= t0 → only the 07-08 row is booked
    assert out["days"] == 1 and repo.nav_history("t_rec")[-1].date == "2026-07-08"


def test_unchanged_hash_preserves_t0_and_recomputes(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "t_rec",
                        {"display_name": "T", "sleeve": "x", "gate_spec": {},
                         "definition": {"params": {"v": 1}, "version": 1, "record_mode": "recompute"}})
    # seed an anchor at an earlier T0 by running "in the past"
    strat = _RecomputeStrat([("2026-07-06", 0.01), ("2026-07-07", 0.02), ("2026-07-08", 0.03)])
    run_track(repo, "t_rec", strat, today="2026-07-06", at=dt.datetime(2026, 7, 6, 23, 30))
    out = run_track(repo, "t_rec", strat, today="2026-07-08", at=dt.datetime(2026, 7, 8, 23, 30))
    assert repo.active_anchor("t_rec").t0 == "2026-07-06" and out["reinception"] is False
    assert out["days"] == 3          # recompute over [2026-07-06, today] = all 3 rows


def test_param_change_reinceptions(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "t_rec",
                        {"display_name": "T", "sleeve": "x", "gate_spec": {},
                         "definition": {"params": {"v": 1}, "version": 1, "record_mode": "recompute"}})
    strat = _RecomputeStrat([("2026-07-06", 0.01), ("2026-07-08", 0.03)])
    run_track(repo, "t_rec", strat, today="2026-07-06", at=dt.datetime(2026, 7, 6, 23, 30))
    # change a param → hash changes → re-inception at today
    registry.STRATEGY_REGISTRY["t_rec"]["definition"]["params"] = {"v": 2}
    out = run_track(repo, "t_rec", strat, today="2026-07-08", at=dt.datetime(2026, 7, 8, 23, 30))
    assert out["reinception"] is True and repo.active_anchor("t_rec").t0 == "2026-07-08"


class _ObservedStrat:
    """An observed strategy: books TODAY on the carried book, returns updated positions."""
    def __init__(self, ret_by_day):
        self._r = ret_by_day

    def advance(self, last_date, extra):
        # book today's return; carry a trivial position
        day = extra.get("_today")
        return [(day, self._r[day])], {"positions": {"X": 1.0}, "_today": day}


def test_observed_mode_appends_and_carries_state(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "t_obs",
                        {"display_name": "T", "sleeve": "x", "gate_spec": {},
                         "definition": {"params": {"v": 1}, "version": 1, "record_mode": "observed"}})
    strat = _ObservedStrat({"2026-07-07": 0.001, "2026-07-08": 0.002})
    # NOTE: run_track injects today into extra as "_today" for observed strategies (see impl)
    run_track(repo, "t_obs", strat, today="2026-07-07", at=dt.datetime(2026, 7, 7, 23, 30))
    out = run_track(repo, "t_obs", strat, today="2026-07-08", at=dt.datetime(2026, 7, 8, 23, 30))
    assert repo.record_series("t_obs") == [("2026-07-07", 0.001), ("2026-07-08", 0.002)]
    assert repo.get_book_state("t_obs")["positions"] == {"X": 1.0}
    assert out["days"] == 2
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_forward_engine.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.forward_engine'`.

- [ ] **Step 3: Implement the engine**

```python
# src/fxhnt/application/forward_engine.py
"""The deterministic forward engine (spec sub-project A §3).

One `run_track` call: read the track's declared definition, manage its immutable anchor (inception on first
run, auto-re-inception on a definition-hash change), then produce the forward record per record_mode —
`recompute` recomputes [T0, today] from PIT data; `observed` appends today's booked row to the immutable log
and carries the strategy's book-state. The nav/summary are derived caches. No mutable, hand-editable state."""
from __future__ import annotations

import datetime as dt
from typing import Any

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_definition import definition_hash, track_definition
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
from fxhnt.application.forward_models import ForwardSummary
from fxhnt.application.forward_recompute import derive_nav, recompute_series


def _write_nav_and_summary(repo: ForwardNavRepo, strategy_id: str,
                           rows: list[tuple[str, float]], at: dt.datetime) -> dict[str, Any]:
    nav_rows, summary = derive_nav(rows)
    repo.upsert_rows(
        [NavRowDTO(strategy_id=strategy_id, date=d, ret=r, nav=n)
         for (d, r), (_, n) in zip(rows, nav_rows)], at=at)
    sm = ForwardSummary(strategy_id=strategy_id, as_of=summary["last_date"] or "",
                        days=summary["days"], nav=summary["nav"], total_return=summary["total_return"],
                        sharpe=summary["sharpe"], maxdd=summary["maxdd"])
    # gate WAIT is the safe default here; the gate evaluation (forward_ingest) sets the real verdict later.
    repo.upsert_summary(sm, gate_status="WAIT", gate_reason="building", at=at)
    return summary


def run_track(repo: ForwardNavRepo, strategy_id: str, strategy: Any, today: str, at: dt.datetime) -> dict[str, Any]:
    params, version, record_mode = track_definition(strategy_id)
    h = definition_hash(params, version)

    active = repo.active_anchor(strategy_id)
    reinception = active is None or active.definition_hash != h
    if reinception:
        if active is not None:
            repo.archive_anchor(strategy_id, at=at)
        repo.set_anchor(strategy_id, t0=today, definition_hash=h, definition_version=version,
                        params_json=params, record_mode=record_mode, at=at)
        t0 = today
    else:
        t0 = active.t0

    if record_mode == "recompute":
        rows = recompute_series(strategy, t0)
    elif record_mode == "observed":
        # book TODAY on the carried book-state; inject `today` so the pure strategy needs no clock
        extra = repo.get_book_state(strategy_id)
        extra["_today"] = today
        booked, new_extra = strategy.advance(None, extra)
        for d, r in booked:
            if d >= t0:
                repo.append_record(strategy_id, str(d), float(r), at=at)
        new_extra.pop("_today", None)
        repo.set_book_state(strategy_id, new_extra, at=at)
        rows = repo.record_series(strategy_id, since=t0)
    else:  # pragma: no cover - guarded by track_definition + test_forward_definition
        raise ValueError(f"unknown record_mode {record_mode!r} for {strategy_id}")

    summary = _write_nav_and_summary(repo, strategy_id, rows, at=at)
    return {"t0": t0, "record_mode": record_mode, "reinception": reinception, **summary}
```

- [ ] **Step 4: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_forward_engine.py -q`
Expected: PASS (4 passed).

If `ForwardSummary`/`ForwardNavRow` field names differ, open `src/fxhnt/application/forward_models.py` and match the exact dataclass fields (the plan assumes `ForwardSummary(strategy_id, as_of, days, nav, total_return, sharpe, maxdd)` and `ForwardNavRow(strategy_id, date, ret, nav)`, consistent with `forward_nav.py:69-76,64-66`).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/forward_engine.py tests/integration/test_forward_engine.py
git commit -F - <<'EOF'
feat(forward): forward engine — anchor lifecycle + recompute/observed dispatch

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 6: Wire `_run_paper_tracker` to the engine

Replace the JSON `ForwardTracker` call with the DB-backed engine, per track, keeping transient-error tolerance.

**Files:**
- Modify: `src/fxhnt/adapters/orchestration/assets.py:435-465` (`_run_paper_tracker`)
- Test: `tests/integration/test_run_paper_tracker_engine.py`

**Interfaces:**
- Consumes: `run_track` (Task 5), the registry `strategy_id` (name minus `_nav`).
- Produces: `_run_paper_tracker` now writes via `run_track` and returns `{"forward_days","last_date","reinception"}`.

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_run_paper_tracker_engine.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_engine import run_track


class _RecomputeStrat:
    def advance(self, last_date, extra):
        return [("2026-07-07", 0.01), ("2026-07-08", 0.02)], extra


def test_engine_writes_nav_for_a_recompute_track(tmp_path, monkeypatch):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/pt.db")
    repo.migrate()
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "sixtyforty",
                        dict(registry.STRATEGY_REGISTRY["sixtyforty"],
                             definition={"params": {"v": 1}, "version": 1, "record_mode": "recompute"}))
    out = run_track(repo, "sixtyforty", _RecomputeStrat(), today="2026-07-06",
                    at=dt.datetime(2026, 7, 8, 23, 30))
    assert out["days"] == 2
    assert [r.date for r in repo.nav_history("sixtyforty")] == ["2026-07-07", "2026-07-08"]
```

(The end-to-end asset wiring is validated by the existing `tests/integration/test_orchestration_definitions.py`; this task's dedicated test asserts the engine writes correctly for a real registry id.)

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_run_paper_tracker_engine.py -q`
Expected: FAIL (the `sixtyforty` registry entry has no `definition` yet, unless Task 2 already added it — if Task 2 is done this passes; the failing signal for THIS task is the `_run_paper_tracker` change below).

- [ ] **Step 3: Rewrite `_run_paper_tracker`**

Replace `src/fxhnt/adapters/orchestration/assets.py:435-465` with:

```python
def _run_paper_tracker(context: AssetExecutionContext, name: str, build_strategy, state_file: str,
                       *, persist_backtest_ref: bool = False) -> dict:  # type: ignore[type-arg,no-untyped-def]
    """Run one forward track through the deterministic engine (spec sub-project A): the DB anchor is the
    only persistent state; the record is recomputed (recompute tracks) or appended (observed tracks). A
    transient network/API error while building the strategy logs a warning and no-ops (the anchor + prior
    record are untouched); genuine bugs propagate and fail the asset loudly.

    `state_file` is retained ONLY for the one-time migration seed (Task 8) and is unused at steady state.
    `persist_backtest_ref=True` also publishes this track's reconciliation-gate reference from its own
    strategy (see `_persist_track_backtest_ref`)."""
    import datetime as _dt
    import json
    import urllib.error

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_engine import run_track
    from fxhnt.config import get_settings

    sid = name[:-4] if name.endswith("_nav") else name
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    try:
        repo = ForwardNavRepo(get_settings().operational_dsn)
        repo.migrate()
        out = run_track(repo, sid, build_strategy(), today=today, at=at)
        context.log.info(f"{name}: {out['days']}d through {out.get('last_date')} "
                         f"(mode={out['record_mode']}, reinception={out['reinception']})")
        if persist_backtest_ref:
            _persist_track_backtest_ref(context, name, build_strategy)
        return {"forward_days": out["days"], "last_date": out.get("last_date"),
                "reinception": out["reinception"]}
    except (urllib.error.URLError, TimeoutError, ConnectionError, OSError, json.JSONDecodeError) as e:
        context.log.warning(f"{name} step failed (record unchanged): {e}")
        return {"forward_days": 0, "last_date": None, "reinception": False}
```

- [ ] **Step 4: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_run_paper_tracker_engine.py tests/integration/test_orchestration_definitions.py -q`
Expected: PASS. (If `test_orchestration_definitions` asserts a specific return shape, update it to include the `reinception` key.)

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/adapters/orchestration/assets.py tests/integration/test_run_paper_tracker_engine.py
git commit -F - <<'EOF'
feat(forward): wire _run_paper_tracker to the DB-backed deterministic engine

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 7: Yahoo PIT snapshot persistence + PIT-reading client

Make multistrat's recompute stable: freeze the Yahoo daily closes point-in-time in the warehouse and read those, not a live re-fetch.

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (add `YahooPitCloseRow`)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (add `snapshot_yahoo_closes`, `yahoo_pit_closes`)
- Create: `src/fxhnt/adapters/data/yahoo_pit.py` (`YahooPitClient`)
- Test: `tests/integration/test_yahoo_pit.py`

**Interfaces:**
- Produces:
  - `YahooPitCloseRow(symbol, date, adj_close, first_seen_at)` — PIT frozen, `(symbol, date)` PK, **never overwritten**.
  - `ForwardNavRepo.snapshot_yahoo_closes(symbol, series: dict[str, float], at)` — inserts only `(symbol, date)` pairs not already present (first-seen wins → PIT frozen).
  - `ForwardNavRepo.yahoo_pit_closes(symbol) -> dict[str, float]` — the frozen `{iso_date: adj_close}`.
  - `YahooPitClient(repo).adj_closes(symbol) -> dict[str, float]` — reads the PIT store (a `DailyBarClient`).

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_yahoo_pit.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/pit.db")
    repo.migrate()
    return repo


def test_snapshot_is_first_seen_wins(tmp_path):
    repo = _repo(tmp_path)
    at = dt.datetime(2026, 7, 8)
    repo.snapshot_yahoo_closes("SPY", {"2026-07-07": 500.0, "2026-07-08": 501.0}, at)
    # a later re-fetch REVISES 07-07 (dividend adj) + adds a new day
    repo.snapshot_yahoo_closes("SPY", {"2026-07-07": 499.5, "2026-07-08": 501.0, "2026-07-09": 502.0}, at)
    closes = repo.yahoo_pit_closes("SPY")
    assert closes["2026-07-07"] == 500.0     # FROZEN at first-seen, not revised
    assert closes["2026-07-09"] == 502.0     # new day appended


def test_pit_client_reads_frozen_series(tmp_path):
    from fxhnt.adapters.data.yahoo_pit import YahooPitClient
    repo = _repo(tmp_path)
    repo.snapshot_yahoo_closes("SPY", {"2026-07-07": 500.0}, dt.datetime(2026, 7, 8))
    assert YahooPitClient(repo).adj_closes("SPY") == {"2026-07-07": 500.0}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_yahoo_pit.py -q`
Expected: FAIL with `AttributeError: ... 'snapshot_yahoo_closes'`.

- [ ] **Step 3: Add the ORM row + repo methods**

In `cockpit_models.py`, append:

```python
class YahooPitCloseRow(CockpitBase):
    """Point-in-time frozen Yahoo daily adjusted close (spec §4). Yahoo revises adjusted closes
    retroactively (dividends/splits); freezing the first-seen value keeps the multistrat recompute stable.
    (symbol, date) PK — first-seen wins; NEVER overwritten in place."""
    __tablename__ = "yahoo_pit_close"
    symbol: Mapped[str] = mapped_column(String(16), primary_key=True)
    date: Mapped[str] = mapped_column(String(10), primary_key=True)
    adj_close: Mapped[float] = mapped_column(Float)
    first_seen_at: Mapped[dt.datetime] = mapped_column(DateTime)
```

In `forward_nav.py` (add `YahooPitCloseRow` to the import), add to `ForwardNavRepo`:

```python
    def snapshot_yahoo_closes(self, symbol: str, series: dict[str, float], at: dt.datetime) -> None:
        """Insert only (symbol, date) pairs not already stored — first-seen wins so the PIT value is frozen."""
        with Session(self._engine) as s:
            have = {d for (d,) in s.execute(
                select(YahooPitCloseRow.date).where(YahooPitCloseRow.symbol == symbol)).all()}
            for d, px in series.items():
                if d not in have:
                    s.add(YahooPitCloseRow(symbol=symbol, date=d, adj_close=float(px), first_seen_at=at))
            s.commit()

    def yahoo_pit_closes(self, symbol: str) -> dict[str, float]:
        stmt = (select(YahooPitCloseRow).where(YahooPitCloseRow.symbol == symbol)
                .order_by(YahooPitCloseRow.date))
        with Session(self._engine) as s:
            return {r.date: r.adj_close for r in s.scalars(stmt)}
```

- [ ] **Step 4: Create the PIT client**

```python
# src/fxhnt/adapters/data/yahoo_pit.py
"""A DailyBarClient that reads the POINT-IN-TIME frozen Yahoo closes from the warehouse (spec §4), so the
multistrat recompute is stable against Yahoo's retroactive adjusted-close revisions. The nightly asset
snapshots today's live fetch into the PIT store first (via ForwardNavRepo.snapshot_yahoo_closes), then the
recompute reads through this client."""
from __future__ import annotations

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


class YahooPitClient:
    def __init__(self, repo: ForwardNavRepo) -> None:
        self._repo = repo

    def adj_closes(self, symbol: str) -> dict[str, float]:
        return self._repo.yahoo_pit_closes(symbol)
```

- [ ] **Step 5: Wire the multistrat asset to snapshot-then-read PIT**

In `src/fxhnt/adapters/orchestration/assets.py`, change `multistrat_nav` (lines ~479-494) so it first snapshots the live Yahoo fetch into the PIT store, then builds `MultiStratStrategy` on the `YahooPitClient`:

```python
@asset
def multistrat_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
    """Paper-forward ADAPTIVE MULTI-STRAT book (5 pure-equity ETFs). Snapshots today's Yahoo adjusted
    closes PIT into the warehouse, then recomputes off the FROZEN PIT series (spec §4) so the record is
    stable against Yahoo's retroactive adjustments."""
    import datetime as _dt

    from fxhnt.adapters.data.yahoo_daily import YahooDailyClient
    from fxhnt.adapters.data.yahoo_pit import YahooPitClient
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.paper_strategies import MultiStratStrategy
    from fxhnt.config import get_settings
    from fxhnt.domain.strategies.multistrat import FUND_INSTRUMENTS

    repo = ForwardNavRepo(get_settings().operational_dsn)
    repo.migrate()
    live = YahooDailyClient()
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)
    for sym in FUND_INSTRUMENTS:
        repo.snapshot_yahoo_closes(sym, live.adj_closes(sym), at)   # freeze today's fetch PIT

    return _run_paper_tracker(
        context, "multistrat_nav",
        lambda: MultiStratStrategy(YahooPitClient(repo), instruments=FUND_INSTRUMENTS),
        "multistrat_state", persist_backtest_ref=True,
    )
```

- [ ] **Step 6: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_yahoo_pit.py -q`
Expected: PASS (2 passed).

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/adapters/persistence/cockpit_models.py src/fxhnt/adapters/persistence/forward_nav.py src/fxhnt/adapters/data/yahoo_pit.py src/fxhnt/adapters/orchestration/assets.py tests/integration/test_yahoo_pit.py
git commit -F - <<'EOF'
feat(forward): PIT Yahoo close snapshots + PIT-reading client (stable recompute)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 8: One-time migration (seed anchors, reconcile, import observed, report)

Seed anchors from the existing `*_state.json` files, preserving honest T0s.

**Files:**
- Create: `src/fxhnt/application/forward_migration.py`
- Modify: `src/fxhnt/cli.py` (add `migrate-forward-anchors` command)
- Test: `tests/integration/test_forward_migration.py`

**Interfaces:**
- Consumes: `ForwardNavRepo`, `recompute_series`/`derive_nav`, `track_definition`, the `state_file` basenames from the registry/assets.
- Produces:
  - `migrate_track(repo, strategy_id, state: dict, strategy, today, at, *, tol_frac=0.02) -> dict` — seeds the anchor from `state["inception"]`; for `recompute` reconciles the recompute vs the accumulated `state["days"]` nav (returns `{"mode","t0","status": "match"|"diverged","max_abs_nav_diff"}`); for `observed` imports `state["days"]` into `forward_record` verbatim + carries `state["extra"]` into book-state (`{"mode","t0","status":"imported","n": len}`).

- [ ] **Step 1: Write the failing test**

```python
# tests/integration/test_forward_migration.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_migration import migrate_track


def _repo(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/mig.db")
    repo.migrate()
    return repo


class _RecomputeStrat:
    def __init__(self, series):
        self._series = series

    def advance(self, last_date, extra):
        return list(self._series), extra


def _reg(monkeypatch, sid, mode):
    from fxhnt import registry
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, sid,
                        {"display_name": "T", "sleeve": "x", "gate_spec": {},
                         "definition": {"params": {"v": 1}, "version": 1, "record_mode": mode}})


def test_recompute_migration_preserves_t0_on_match(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    _reg(monkeypatch, "t_rec", "recompute")
    series = [("2026-07-06", 0.01), ("2026-07-07", 0.02)]
    # accumulated state whose nav MATCHES the recompute over [inception, ...]
    nav = 1.0; days = []
    for d, r in series:
        nav *= 1 + r; days.append({"date": d, "ret": r, "nav": nav})
    state = {"inception": "2026-07-06", "last_date": "2026-07-07", "nav": nav, "days": days, "extra": {}}
    out = migrate_track(repo, "t_rec", state, _RecomputeStrat(series), today="2026-07-07",
                        at=dt.datetime(2026, 7, 7, 23, 30))
    assert out["status"] == "match" and repo.active_anchor("t_rec").t0 == "2026-07-06"


def test_recompute_migration_flags_divergence(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    _reg(monkeypatch, "t_rec", "recompute")
    series = [("2026-07-06", 0.01), ("2026-07-07", 0.02)]
    # accumulated state that DRIFTED (wrong nav) → divergence flagged
    state = {"inception": "2026-07-06", "last_date": "2026-07-07", "nav": 5.0,
             "days": [{"date": "2026-07-06", "ret": 0.5, "nav": 1.5},
                      {"date": "2026-07-07", "ret": 2.0, "nav": 4.5}], "extra": {}}
    out = migrate_track(repo, "t_rec", state, _RecomputeStrat(series), today="2026-07-07",
                        at=dt.datetime(2026, 7, 7, 23, 30))
    assert out["status"] == "diverged" and out["max_abs_nav_diff"] > 0.02


def test_observed_migration_imports_verbatim(tmp_path, monkeypatch):
    repo = _repo(tmp_path)
    _reg(monkeypatch, "t_obs", "observed")
    state = {"inception": "2026-07-06", "last_date": "2026-07-07", "nav": 1.03,
             "days": [{"date": "2026-07-06", "ret": 0.01}, {"date": "2026-07-07", "ret": 0.02}],
             "extra": {"positions": {"BTCUSDT": 0.5}}}
    out = migrate_track(repo, "t_obs", state, _RecomputeStrat([]), today="2026-07-07",
                        at=dt.datetime(2026, 7, 7, 23, 30))
    assert out["status"] == "imported" and out["n"] == 2
    assert repo.record_series("t_obs") == [("2026-07-06", 0.01), ("2026-07-07", 0.02)]
    assert repo.get_book_state("t_obs") == {"positions": {"BTCUSDT": 0.5}}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/integration/test_forward_migration.py -q`
Expected: FAIL with `ModuleNotFoundError: No module named 'fxhnt.application.forward_migration'`.

- [ ] **Step 3: Implement the migration**

```python
# src/fxhnt/application/forward_migration.py
"""One-time migration from the mutable *_state.json trackers to the DB anchor model (spec §6).

Preserve-T0-and-reconcile: seed the anchor from the JSON inception; for recompute tracks compare the
recompute over [T0, today] against the accumulated nav (a divergence is evidence the old record drifted —
flag it, do not silently overwrite); for observed tracks import the accumulated booked series verbatim
(it is the only record of those observations) and carry the book-state."""
from __future__ import annotations

import datetime as dt
from typing import Any

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_definition import definition_hash, track_definition
from fxhnt.application.forward_recompute import derive_nav, recompute_series


def migrate_track(repo: ForwardNavRepo, strategy_id: str, state: dict[str, Any], strategy: Any,
                  today: str, at: dt.datetime, *, tol_frac: float = 0.02) -> dict[str, Any]:
    params, version, record_mode = track_definition(strategy_id)
    t0 = state["inception"]
    repo.set_anchor(strategy_id, t0=t0, definition_hash=definition_hash(params, version),
                    definition_version=version, params_json=params, record_mode=record_mode, at=at)

    if record_mode == "recompute":
        rows = recompute_series(strategy, t0)
        recomputed_nav, _ = derive_nav(rows)
        recomputed = {d: n for d, n in recomputed_nav}
        max_abs = 0.0
        for day in state.get("days", []):
            got = recomputed.get(day["date"])
            if got is not None:
                max_abs = max(max_abs, abs(got - day["nav"]) / max(abs(day["nav"]), 1e-9))
        status = "diverged" if max_abs > tol_frac else "match"
        return {"mode": record_mode, "t0": t0, "status": status, "max_abs_nav_diff": max_abs}

    # observed: import the accumulated booked returns verbatim + carry the book-state
    days = state.get("days", [])
    for day in days:
        if day["date"] >= t0:
            repo.append_record(strategy_id, day["date"], float(day["ret"]), at=at)
    extra = dict(state.get("extra", {}))
    if extra:
        repo.set_book_state(strategy_id, extra, at=at)
    return {"mode": record_mode, "t0": t0, "status": "imported", "n": len(days)}
```

- [ ] **Step 4: Add the CLI command**

In `src/fxhnt/cli.py`, add (near the other forward commands):

```python
@app.command("migrate-forward-anchors")
def migrate_forward_anchors() -> None:
    """One-time: seed forward_anchor rows from the on-disk *_state.json trackers (spec sub-project A §6).
    Recompute tracks reconcile (divergence => drift, logged); observed tracks import their booked log."""
    import datetime as _dt
    import json
    from pathlib import Path

    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_migration import migrate_track
    from fxhnt.config import get_settings
    from fxhnt.registry import STRATEGY_REGISTRY

    from fxhnt.adapters.orchestration.assets import _data_dir

    s = get_settings()
    data_dir = _data_dir()                  # the trackers' state directory (reuse the asset helper)
    repo = ForwardNavRepo(s.operational_dsn)
    repo.migrate()
    today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d")
    at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None)

    # strategy_id -> (state_file basename, build_strategy) for the recompute reconcile. The build callables
    # mirror each *_nav asset. Observed tracks pass a no-op strategy (their record is imported, not recomputed).
    from fxhnt.adapters.orchestration.migration_builders import track_builders   # see note below
    builders = track_builders(s, data_dir)

    for sid in STRATEGY_REGISTRY:
        sf, build = builders.get(sid, (None, None))
        if sf is None:
            typer.echo(f"{sid}: no state-file mapping, skipped")
            continue
        path = Path(data_dir) / f"{sf}.json"
        if not path.exists():
            typer.echo(f"{sid}: no {path.name}, skipped (fresh track)")
            continue
        state = json.loads(path.read_text())
        out = migrate_track(repo, sid, state, build(), today=today, at=at)
        typer.echo(f"{sid}: {out}")
```

Note: create a tiny `src/fxhnt/adapters/orchestration/migration_builders.py` exposing
`track_builders(settings, data_dir) -> dict[str, tuple[str, Callable[[], ForwardStrategy]]]` that maps each
registry id to its `(state_file, build_strategy)` — copy the exact `build_strategy` lambdas and `state_file`
strings already present in each `*_nav` asset in `assets.py` (e.g. `multistrat` →
`("multistrat_state", lambda: MultiStratStrategy(YahooPitClient(repo), FUND_INSTRUMENTS))`). This keeps the
migration DRY against the asset definitions.

- [ ] **Step 5: Run tests to verify they pass**

Run: `uv run pytest tests/integration/test_forward_migration.py -q`
Expected: PASS (3 passed).

- [ ] **Step 6: Commit**

```bash
git add src/fxhnt/application/forward_migration.py src/fxhnt/adapters/orchestration/migration_builders.py src/fxhnt/cli.py tests/integration/test_forward_migration.py
git commit -F - <<'EOF'
feat(forward): one-time migration — seed anchors, reconcile recompute, import observed

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

### Task 9: Retire the `state_reader` JSON path

Make the forward read-model read the DB record; remove the JSON dependency from steady-state reads.

**Files:**
- Read first: `src/fxhnt/adapters/persistence/state_reader.py` (find its callers via `grep -rn state_reader src/`)
- Modify: the forward read-model / dashboard read path that currently calls `state_reader`
- Test: `tests/integration/test_forward_readmodel_from_db.py`

**Interfaces:**
- The dashboard/gate read model now sources nav + summary from `ForwardNavRepo.nav_history`/`all_summaries` (already DB-backed after Tasks 5-6) and the anchor from `active_anchor`; the `*_state.json` normalisation is removed from the request path.

- [ ] **Step 1: Identify the callers**

Run: `grep -rn "state_reader\|read_state_file\|_state.json" src/fxhnt/ | grep -v test`
Read each caller. The dashboard forward view and any gate input that still reads the JSON must switch to the DB.

- [ ] **Step 2: Write the failing test**

```python
# tests/integration/test_forward_readmodel_from_db.py
import datetime as dt

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo


def test_read_model_sources_nav_and_anchor_from_db_only(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/rm.db")
    repo.migrate()
    at = dt.datetime(2026, 7, 8, 23, 30)
    repo.set_anchor("multistrat", "2026-07-01", "h", 1, {"n": 5}, "recompute", at=at)
    repo.upsert_rows([], at=at)   # no rows yet
    # the read model returns the anchor T0 without touching any JSON file
    a = repo.active_anchor("multistrat")
    assert a.t0 == "2026-07-01"
    assert repo.nav_history("multistrat") == []
```

- [ ] **Step 3: Run test to verify it fails / passes**

Run: `uv run pytest tests/integration/test_forward_readmodel_from_db.py -q`
Expected: PASS on the repo assertions (this test locks the DB-only contract; the real change is removing the JSON caller in Step 4).

- [ ] **Step 4: Remove the JSON read from the request path**

In the dashboard forward read model (the `state_reader` caller found in Step 1), replace the `*_state.json` load with `ForwardNavRepo(...).nav_history(sid)` + `active_anchor(sid)` (inception = `anchor.t0`). Delete `state_reader.py` only if it has no remaining callers after this change; otherwise leave it for the migration (Task 8) and mark it migration-only in its module docstring.

- [ ] **Step 5: Run the full suite**

Run: `uv run pytest -q`
Expected: PASS (all). Fix any test that asserted JSON-sourced inception to read `anchor.t0`.

- [ ] **Step 6: Commit**

```bash
git add -A src/fxhnt tests/integration/test_forward_readmodel_from_db.py
git commit -F - <<'EOF'
refactor(forward): read model sources anchor + nav from DB; retire state_reader JSON path

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
EOF
```

---

## Post-implementation: migration run (operational, not a code task)

After all tasks merge and deploy, run ONCE in-cluster (never on prod capital — this only touches the paper
forward records): `kubectl exec <dagster-pod> -c daemon -- fxhnt migrate-forward-anchors`. Verify each track
reports `match` or `imported`; investigate any `diverged` track (it reveals the old accumulated record was
drifting) before trusting its gate verdict. This is the "preserve T0 + reconcile" step from spec §6.

## Self-Review notes

- **Spec coverage:** anchor (§1)→T1; definition+hash (§2)→T2; recompute engine (§3)→T4/T5; two modes
  (§3b)→T3/T5; determinism+PIT (§4)→T4/T7; fills-boundary (§5)→scoped out (no task, by design); migration
  (§6)→T8; self-healing (§7)→T5 (recompute is stateless; observed appends idempotently); error handling→T5/T6
  (WAIT-safe empty; transient tolerance); testing→each task's tests. Retiring the JSON path→T9.
- **WAIT-safe:** `derive_nav([])` returns days=0/nav=1.0 and `run_track` writes `gate_status="WAIT"`; the real
  verdict is set by the existing `forward_ingest` gate evaluation, unchanged by A.
- **Type consistency:** `run_track`/`migrate_track` share `recompute_series`/`derive_nav`; the summary dict
  keys (`days, nav, total_return, sharpe, maxdd, last_date`) are identical across Tasks 4/5/8.
