# Declarative Gate + Promotion Policy (Sub-project B1) 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:** Consolidate the scattered gate + promotion defaults into one in-code, versioned `GatePolicy`; resolve the effective per-track policy at a single point; stamp every gate verdict with the `policy_version` + content-derived `policy_hash` that produced it; and add an automatic correlation backstop to promotion.

**Architecture:** A new `gate_policy.py` holds the policy defaults (values byte-identical to today's module constants) + a `POLICY_VERSION` + a `policy_hash`. `resolve_policy(POLICY, gate_spec) -> ResolvedPolicy` layers per-track registry overrides on the defaults at one point; `evaluate_reconciliation_gate`/`evaluate_gate` take the `ResolvedPolicy` (their module constants / inline defaults are removed). `forward_ingest` resolves once per track, stamps the verdict onto `forward_summary`, and adds a deploy-book correlation backstop to `evaluate_promotions`. Behavior is byte-identical under the current thresholds.

**Tech Stack:** Python 3.12, dataclasses, SQLAlchemy 2.0 ORM (sqlite for tests, Postgres prod), pytest (`uv run pytest`), ruff (E,F,I,UP,B,SIM; line-length 120).

## Global Constraints

- Spec: `docs/superpowers/specs/2026-07-09-declarative-gate-promotion-policy-b1-design.md` (authoritative).
- Builds on sub-project A (merged to master): reads `forward_summary` + `forward_nav` (A's record) and extends `forward_ingest` / the gate functions.
- REAL MONEY: the gate must stay WAIT-safe — never a false PASS. The consolidation must produce **byte-identical verdicts** under the current thresholds (a regression anchor test proves it). Gate verdict LOGIC (the comparisons) is unchanged; only where the thresholds come from changes.
- Policy lives in **git-committed code**, never runtime-editable in the DB. A policy change = a code commit + a `POLICY_VERSION` bump; the content-derived `policy_hash` self-check forces the bump.
- No capital: B1 touches only gate STATUS + promotion (a tracking tier). Capital is B2.
- Idempotency: DB writes are `Session.merge` upserts. New `forward_summary` columns are nullable (migration-tolerant). Determinism: no wall-clock inside pure gate/policy functions.
- Never save tests to the repo root; tests under `tests/unit/` or `tests/integration/` per the existing layout. Commit after every task; messages end with the `Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>` trailer.
- Promotion is ONE-WAY + persistent (unchanged): once promoted, never reverts.

---

### Task 1: `gate_policy.py` — the versioned policy object + hash

**Files:**
- Create: `src/fxhnt/application/gate_policy.py`
- Test: `tests/unit/test_gate_policy.py`

**Interfaces:**
- Produces:
  - `GatePolicy` frozen dataclass with fields (defaults = today's constants): `min_forward_days: int = 14`, `recon_tolerance_frac: float = 0.5`, `recon_min_band: float = 0.05`, `recon_min_corr: float = 0.0`, `max_gap_days: int = 1`, `min_days: int = 20`, `min_total_return: float = 0.0`, `min_sharpe: float = 0.0`, `promotion_max_corr: float = 0.5`, `promotion_min_overlap_days: int = 10`.
  - `POLICY: GatePolicy` — the single current instance (all defaults).
  - `POLICY_VERSION: int = 1`.
  - `policy_hash(policy: GatePolicy, version: int) -> str` — sha256 of the canonical field values + version.

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

```python
# tests/unit/test_gate_policy.py
from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, GatePolicy, policy_hash


def test_policy_defaults_match_todays_constants():
    # these values must equal the pre-B1 reconciliation_gate / gate.py constants (byte-identical verdicts)
    assert (POLICY.min_forward_days, POLICY.recon_tolerance_frac, POLICY.recon_min_band,
            POLICY.recon_min_corr, POLICY.max_gap_days) == (14, 0.5, 0.05, 0.0, 1)
    assert (POLICY.min_days, POLICY.min_total_return, POLICY.min_sharpe) == (20, 0.0, 0.0)


def test_policy_hash_is_stable_and_content_derived():
    assert policy_hash(POLICY, POLICY_VERSION) == policy_hash(POLICY, POLICY_VERSION)          # stable
    changed = GatePolicy(min_forward_days=21)                                                  # a field changed
    assert policy_hash(changed, POLICY_VERSION) != policy_hash(POLICY, POLICY_VERSION)         # → different hash
    assert policy_hash(POLICY, 2) != policy_hash(POLICY, 1)                                    # version bump → different


def test_policy_hash_selfcheck_forces_version_bump():
    # PIN the current hash. If a policy field changes without this literal being updated (and the version
    # bumped), this test fails — forcing a deliberate, audited change. Update BOTH on any real policy change.
    assert policy_hash(POLICY, POLICY_VERSION) == _EXPECTED_HASH  # noqa: F821 — defined below in the test file
```

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

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

- [ ] **Step 3: Implement `gate_policy.py`**

```python
# src/fxhnt/application/gate_policy.py
"""The declarative, versioned gate + promotion POLICY (sub-project B1).

Single source for the gate/promotion DEFAULTS that used to be scattered as module constants in
reconciliation_gate.py + inline literals in gate.py. Per-track overrides still live in the registry
gate_spec; `resolve_policy` (gate_policy_resolve.py) layers them on top. A policy change = a code commit +
a POLICY_VERSION bump; the content-derived `policy_hash` self-check test forces the bump (mirrors sub-project
A's definition_version discipline). Every gate verdict is stamped with (POLICY_VERSION, policy_hash) for
audit — 'PASSed under policy v3' is queryable."""
from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass


@dataclass(frozen=True, slots=True)
class GatePolicy:
    # reconciliation gate
    min_forward_days: int = 14
    recon_tolerance_frac: float = 0.5
    recon_min_band: float = 0.05
    recon_min_corr: float = 0.0
    max_gap_days: int = 1
    # absolute gate
    min_days: int = 20
    min_total_return: float = 0.0
    min_sharpe: float = 0.0
    # promotion (research -> deploy) correlation backstop
    promotion_max_corr: float = 0.5
    promotion_min_overlap_days: int = 10


POLICY = GatePolicy()
POLICY_VERSION = 1


def policy_hash(policy: GatePolicy, version: int) -> str:
    canonical = json.dumps(asdict(policy), sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(f"{canonical}|v{version}".encode()).hexdigest()
```

- [ ] **Step 4: Pin the expected hash in the test, run to verify it passes**

Compute the hash once and paste it into the test file:
Run: `uv run python -c "from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash; print(policy_hash(POLICY, POLICY_VERSION))"`
Add `_EXPECTED_HASH = "<the printed value>"` near the top of `tests/unit/test_gate_policy.py`.
Run: `uv run pytest tests/unit/test_gate_policy.py -q`
Expected: PASS (3 passed).

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/gate_policy.py tests/unit/test_gate_policy.py
git commit -F - <<'EOF'
feat(policy): versioned GatePolicy + content-derived policy_hash (B1)

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

---

### Task 2: `resolve_policy` — layer per-track overrides on the policy defaults

**Files:**
- Create: `src/fxhnt/application/gate_policy_resolve.py`
- Test: `tests/unit/test_gate_policy_resolve.py`

**Interfaces:**
- Consumes: `GatePolicy` (Task 1).
- Produces:
  - `ResolvedPolicy` frozen dataclass with the SAME threshold fields the two gates need: `min_forward_days`, `recon_tolerance_frac`, `recon_min_band`, `recon_min_corr`, `max_gap_days`, `min_days`, `min_total_return`, `min_sharpe`.
  - `resolve_policy(policy: GatePolicy, gate_spec: dict) -> ResolvedPolicy` — each field = `gate_spec.get(<key>, getattr(policy, <field>))`.

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

```python
# tests/unit/test_gate_policy_resolve.py
from fxhnt.application.gate_policy import POLICY
from fxhnt.application.gate_policy_resolve import resolve_policy


def test_unset_falls_back_to_policy_default():
    r = resolve_policy(POLICY, {})
    assert r.min_forward_days == 14 and r.recon_tolerance_frac == 0.5 and r.max_gap_days == 1
    assert r.min_days == 20 and r.min_sharpe == 0.0


def test_gate_spec_override_wins():
    r = resolve_policy(POLICY, {"min_forward_days": 21, "max_gap_days": 4, "min_sharpe": 0.5})
    assert r.min_forward_days == 21 and r.max_gap_days == 4 and r.min_sharpe == 0.5
    assert r.recon_tolerance_frac == 0.5   # unset field still the default
```

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

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

- [ ] **Step 3: Implement `gate_policy_resolve.py`**

```python
# src/fxhnt/application/gate_policy_resolve.py
"""Resolve the effective per-track gate policy: registry gate_spec overrides layered on the GatePolicy
defaults at ONE point (no `.get(..., CONSTANT)` scattered across the gate functions). The gate evaluators
take the ResolvedPolicy — the single place defaults and per-track overrides combine."""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from fxhnt.application.gate_policy import GatePolicy


@dataclass(frozen=True, slots=True)
class ResolvedPolicy:
    min_forward_days: int
    recon_tolerance_frac: float
    recon_min_band: float
    recon_min_corr: float
    max_gap_days: int
    min_days: int
    min_total_return: float
    min_sharpe: float


def resolve_policy(policy: GatePolicy, gate_spec: dict[str, Any]) -> ResolvedPolicy:
    return ResolvedPolicy(
        min_forward_days=int(gate_spec.get("min_forward_days", policy.min_forward_days)),
        recon_tolerance_frac=float(gate_spec.get("recon_tolerance_frac", policy.recon_tolerance_frac)),
        recon_min_band=float(gate_spec.get("recon_min_band", policy.recon_min_band)),
        recon_min_corr=float(gate_spec.get("recon_min_corr", policy.recon_min_corr)),
        max_gap_days=int(gate_spec.get("max_gap_days", policy.max_gap_days)),
        min_days=int(gate_spec.get("min_days", policy.min_days)),
        min_total_return=float(gate_spec.get("min_total_return", policy.min_total_return)),
        min_sharpe=float(gate_spec.get("min_sharpe", policy.min_sharpe)),
    )
```

- [ ] **Step 4: Run test to verify it passes**

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

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/gate_policy_resolve.py tests/unit/test_gate_policy_resolve.py
git commit -F - <<'EOF'
feat(policy): resolve_policy — per-track overrides on GatePolicy defaults, one resolution point (B1)

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

---

### Task 3: Wire the gate functions to `ResolvedPolicy` (remove scattered constants)

**Files:**
- Modify: `src/fxhnt/application/reconciliation_gate.py` (signature + remove module constants)
- Modify: `src/fxhnt/application/gate.py` (signature)
- Modify: `src/fxhnt/application/forward_ingest.py:55-61` (`_gate_verdict` resolves once)
- Modify: existing gate tests that call these functions (re-point to pass a `ResolvedPolicy`)
- Test: `tests/unit/test_gate_policy_regression.py`

**Interfaces:**
- Consumes: `resolve_policy` (Task 2), `ResolvedPolicy`.
- Produces:
  - `evaluate_reconciliation_gate(summary, rows, backtest, policy: ResolvedPolicy) -> GateVerdict` (was `gate_spec: dict`).
  - `evaluate_gate(summary, policy: ResolvedPolicy) -> GateVerdict` (was `gate_spec: dict`).

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

```python
# tests/unit/test_gate_policy_regression.py
from fxhnt.application.forward_models import ForwardNavRow, ForwardSummary
from fxhnt.application.gate_policy import POLICY
from fxhnt.application.gate_policy_resolve import resolve_policy
from fxhnt.application.reconciliation_gate import BacktestReference, evaluate_reconciliation_gate


def _summary(days, total_return, sid="x"):
    return ForwardSummary(strategy_id=sid, as_of="2026-07-01", days=days, nav=1.0 + total_return,
                          total_return=total_return, sharpe=1.0, maxdd=-0.01)


def test_recon_gate_takes_resolved_policy_and_verdicts_are_unchanged():
    pol = resolve_policy(POLICY, {})
    rows = [ForwardNavRow(strategy_id="x", date=f"2026-06-{d:02d}", ret=0.001, nav=1.0 + 0.001 * i)
            for i, d in enumerate(range(1, 16), 0)]
    # 15 days >= 14, forward +1.5% vs expected +1.0% (outperforms → in band), clean → PASS
    v = evaluate_reconciliation_gate(_summary(15, 0.015), rows, BacktestReference(expected_window_return=0.01), pol)
    assert v.status == "PASS"
    # short window → WAIT building
    v2 = evaluate_reconciliation_gate(_summary(5, 0.01), rows[:5], BacktestReference(expected_window_return=0.01), pol)
    assert v2.status == "WAIT" and "building" in v2.reason
    # no backtest ref → WAIT (PASS withheld), unchanged
    v3 = evaluate_reconciliation_gate(_summary(15, 0.015), rows, None, pol)
    assert v3.status == "WAIT"
```

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

Run: `uv run pytest tests/unit/test_gate_policy_regression.py -q`
Expected: FAIL — `evaluate_reconciliation_gate` still takes `gate_spec` (a `ResolvedPolicy` has no `.get`), so it raises `AttributeError`.

- [ ] **Step 3: Rewire `reconciliation_gate.py`**

Remove the module constants `MIN_FORWARD_DAYS`, `RECON_TOLERANCE_FRAC`, `RECON_MIN_BAND`, `RECON_MIN_CORR` (lines 36, 43, 47, 58) and change the function to read from the resolved policy. Replace the signature + the four `gate_spec.get(...)` lines (114-117) and the `max_gap_days` read (126):

```python
from fxhnt.application.gate_policy_resolve import ResolvedPolicy  # add to imports

def evaluate_reconciliation_gate(summary: ForwardSummary, rows: list[ForwardNavRow],
                                 backtest: BacktestReference | None,
                                 policy: ResolvedPolicy) -> GateVerdict:
    """... (keep the docstring; note thresholds now come from the resolved policy) ..."""
    if summary.days < policy.min_forward_days:
        return GateVerdict("WAIT", f"building ({summary.days}/{policy.min_forward_days} days)")
    if _has_gaps(rows, policy.max_gap_days):
        return GateVerdict("WAIT", "execution gap (missing/duplicate forward day in the record)")
    if backtest is None:
        return GateVerdict("WAIT", f"{summary.days}d: no backtest reference to reconcile against — PASS withheld "
                                   f"(persist a backtest_summary row for this strategy)")
    expected = backtest.expected_window_return
    band = policy.recon_tolerance_frac * max(abs(expected), policy.recon_min_band)
    shortfall = expected - summary.total_return
    if shortfall > band:
        return GateVerdict("WAIT", f"forward diverging from backtest (Δ {100.0 * shortfall:+.2f}% below "
                                   f"expected, band ±{100.0 * band:.2f}%)")
    if backtest.daily_returns is not None:
        corr = _corr([r.ret for r in rows], backtest.daily_returns)
        if corr is not None and corr < policy.recon_min_corr:
            return GateVerdict("WAIT", f"forward diverging from backtest (daily corr {corr:+.2f} < "
                                       f"{policy.recon_min_corr:.2f})")
    return GateVerdict("PASS", f"{summary.days}d: +{summary.total_return_pct:.2f}%, reconciles with backtest, "
                               f"clean execution")
```

- [ ] **Step 4: Rewire `gate.py`**

Change `evaluate_gate(summary, gate_spec)` to take the resolved policy:

```python
from fxhnt.application.gate_policy_resolve import ResolvedPolicy

def evaluate_gate(summary: ForwardSummary, policy: ResolvedPolicy) -> GateVerdict:
    if summary.days < policy.min_days:
        return GateVerdict("WAIT", f"{summary.days}/{policy.min_days} forward days")
    if summary.total_return <= policy.min_total_return:
        return GateVerdict("NO_GO", f"forward return {summary.total_return_pct:.2f}% <= "
                                    f"{policy.min_total_return * 100:.2f}%")
    if summary.sharpe < policy.min_sharpe:
        return GateVerdict("NO_GO", f"forward Sharpe {summary.sharpe:.2f} < {policy.min_sharpe}")
    return GateVerdict("GO", f"forward Sharpe {summary.sharpe:.2f}, +{summary.total_return_pct:.2f}%")
```

(Preserve the EXACT reason strings that the current `gate.py` emits — read the current file lines 11-21 and keep them byte-identical, only swapping the threshold source. If the current NO_GO/GO reason wording differs from above, keep the current wording.)

- [ ] **Step 5: Rewire `_gate_verdict` in `forward_ingest.py`**

Replace the `MIN_FORWARD_DAYS` import and resolve the policy once per track:

```python
# imports: drop MIN_FORWARD_DAYS; add:
from fxhnt.application.gate_policy import POLICY
from fxhnt.application.gate_policy_resolve import resolve_policy

def _gate_verdict(summary, rows, gate_spec, backtest_ref, sid):
    policy = resolve_policy(POLICY, gate_spec)
    if gate_spec.get("gate_type") == "reconciliation":
        ref = backtest_ref.reference(sid, max(summary.days, policy.min_forward_days)) if backtest_ref else None
        return evaluate_reconciliation_gate(summary, rows, ref, policy)
    return evaluate_gate(summary, policy)
```

- [ ] **Step 6: Fix the existing gate tests + run**

Update any existing test that calls `evaluate_reconciliation_gate(..., gate_spec)` or `evaluate_gate(summary, gate_spec)` to pass `resolve_policy(POLICY, gate_spec)` instead (grep: `grep -rn "evaluate_reconciliation_gate\|evaluate_gate\|MIN_FORWARD_DAYS" tests/ src/`). Then run the full suite ONCE:
Run: `uv run pytest -q`
Expected: PASS (all — the verdicts are byte-identical because `POLICY` == the old constants).

- [ ] **Step 7: Commit**

```bash
git add src/fxhnt/application/reconciliation_gate.py src/fxhnt/application/gate.py src/fxhnt/application/forward_ingest.py tests/
git commit -F - <<'EOF'
refactor(policy): gate functions take ResolvedPolicy; remove scattered gate constants (B1)

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

---

### Task 4: Stamp the verdict — `forward_summary` gains policy_version + policy_hash

**Files:**
- Modify: `src/fxhnt/adapters/persistence/cockpit_models.py` (`ForwardSummaryRow` columns)
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (`upsert_summary` signature)
- Test: `tests/integration/test_forward_summary_policy_stamp.py`

**Interfaces:**
- Produces:
  - `ForwardSummaryRow.policy_version: Mapped[int | None]`, `ForwardSummaryRow.policy_hash: Mapped[str | None]`.
  - `ForwardNavRepo.upsert_summary(sm, gate_status, gate_reason, at, *, policy_version=None, policy_hash=None)`.

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

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

from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.application.forward_models import ForwardSummary


def test_upsert_summary_stamps_policy_version_and_hash(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/s.db"); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    sm = ForwardSummary(strategy_id="x", as_of="2026-07-01", days=15, nav=1.02, total_return=0.02,
                        sharpe=1.0, maxdd=-0.01)
    repo.upsert_summary(sm, "PASS", "ok", at, policy_version=3, policy_hash="abc123")
    row = {s.strategy_id: s for s in repo.all_summaries()}["x"]
    assert row.policy_version == 3 and row.policy_hash == "abc123"


def test_upsert_summary_stamp_defaults_none(tmp_path):
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/s2.db"); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    sm = ForwardSummary(strategy_id="y", as_of="2026-07-01", days=1, nav=1.0, total_return=0.0,
                        sharpe=0.0, maxdd=0.0)
    repo.upsert_summary(sm, "WAIT", "building", at)   # no stamp passed → columns nullable
    row = {s.strategy_id: s for s in repo.all_summaries()}["y"]
    assert row.policy_version is None and row.policy_hash is None
```

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

Run: `uv run pytest tests/integration/test_forward_summary_policy_stamp.py -q`
Expected: FAIL with `TypeError: upsert_summary() got an unexpected keyword argument 'policy_version'`.

- [ ] **Step 3: Add the columns**

In `cockpit_models.py`, add to `ForwardSummaryRow` (after `gate_reason`):

```python
    policy_version: Mapped[int | None] = mapped_column(Integer, nullable=True, default=None)
    policy_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, default=None)
```

- [ ] **Step 4: Stamp them in `upsert_summary`**

In `forward_nav.py`, change `upsert_summary`:

```python
    def upsert_summary(self, sm: ForwardSummary, gate_status: str, gate_reason: str, at: dt.datetime,
                       *, policy_version: int | None = None, policy_hash: str | None = None) -> None:
        with Session(self._engine) as s:
            s.merge(ForwardSummaryRow(
                strategy_id=sm.strategy_id, as_of=sm.as_of, days=sm.days, nav=sm.nav,
                total_return=sm.total_return, sharpe=sm.sharpe, maxdd=sm.maxdd,
                gate_status=gate_status, gate_reason=gate_reason, src_updated_at=at,
                policy_version=policy_version, policy_hash=policy_hash))
            s.commit()
```

Note the prod DB already has a `forward_summary` table; the two nullable columns need an additive ALTER on Postgres. If `ForwardNavRepo.migrate()` uses only `create_all` (which does NOT add columns to an existing table), add a guarded `ALTER TABLE forward_summary ADD COLUMN IF NOT EXISTS policy_version INTEGER` / `... policy_hash VARCHAR(64)` in `migrate()` for the Postgres branch (mirror the existing hypertable-DDL pattern). Read `migrate()` first; if it already handles additive columns elsewhere, follow that pattern. On SQLite (tests) `create_all` creates the full table, so tests pass without the ALTER.

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

Run: `uv run pytest tests/integration/test_forward_summary_policy_stamp.py -q`
Expected: PASS (2 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_summary_policy_stamp.py
git commit -F - <<'EOF'
feat(policy): stamp forward_summary with policy_version + policy_hash (B1 audit)

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

---

### Task 5: `forward_ingest` stamps the policy on every verdict

**Files:**
- Modify: `src/fxhnt/application/forward_ingest.py` (`ForwardIngestService.ingest`)
- Test: `tests/integration/test_forward_ingest.py` (add a stamp assertion)

**Interfaces:**
- Consumes: `POLICY`, `POLICY_VERSION`, `policy_hash` (Task 1); `upsert_summary(..., policy_version, policy_hash)` (Task 4).

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

```python
# add to tests/integration/test_forward_ingest.py
def test_ingest_stamps_policy_on_verdict(tmp_path):
    import datetime as dt
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_ingest import ForwardIngestService
    from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO, ForwardSummary
    from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash
    repo = ForwardNavRepo(f"sqlite:///{tmp_path}/st.db"); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    sid = "crypto_tstrend"
    repo.upsert_rows([NavRowDTO(strategy_id=sid, date="2026-06-01", ret=0.001, nav=1.001)], at=at)
    repo.upsert_summary(ForwardSummary(strategy_id=sid, as_of="2026-06-01", days=1, nav=1.001,
                                       total_return=0.001, sharpe=0.0, maxdd=0.0),
                        "WAIT", "building", at)
    ForwardIngestService(repo).ingest(at=at)
    row = {s.strategy_id: s for s in repo.all_summaries()}[sid]
    assert row.policy_version == POLICY_VERSION
    assert row.policy_hash == policy_hash(POLICY, POLICY_VERSION)
```

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

Run: `uv run pytest tests/integration/test_forward_ingest.py::test_ingest_stamps_policy_on_verdict -q`
Expected: FAIL — `row.policy_version` is `None` (ingest doesn't stamp yet).

- [ ] **Step 3: Stamp in `ingest`**

In `forward_ingest.py`, add the imports and stamp the `upsert_summary` call inside `ingest`'s per-tracker try block:

```python
from fxhnt.application.gate_policy import POLICY, POLICY_VERSION, policy_hash
# ... inside ingest(), replace the upsert_summary call:
                self._repo.upsert_summary(summary, verdict.status, verdict.reason, at,
                                          policy_version=POLICY_VERSION,
                                          policy_hash=policy_hash(POLICY, POLICY_VERSION))
```

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

Run: `uv run pytest tests/integration/test_forward_ingest.py -q`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/fxhnt/application/forward_ingest.py tests/integration/test_forward_ingest.py
git commit -F - <<'EOF'
feat(policy): ingest stamps POLICY_VERSION + policy_hash on every gate verdict (B1)

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

---

### Task 6: Promotion correlation backstop

**Files:**
- Modify: `src/fxhnt/adapters/persistence/forward_nav.py` (add a deploy-book returns helper if needed)
- Modify: `src/fxhnt/application/forward_ingest.py` (`evaluate_promotions`)
- Create: `src/fxhnt/application/promotion_corr.py` (pure corr helpers)
- Test: `tests/integration/test_promotion_corr.py`

**Interfaces:**
- Consumes: `ForwardNavRepo.nav_history(sid)`, `ForwardNavRepo.promoted_strategy_ids()`, `STRATEGY_REGISTRY` (for `tier`, `promote_on_pass`), `POLICY.promotion_max_corr`, `POLICY.promotion_min_overlap_days`.
- Produces:
  - `promotion_corr.deploy_book_returns(returns_by_sid: dict[str, dict[str, float]]) -> dict[str, float]` — equal-weight per-date average of the given deploy tracks' daily returns (keyed ISO date).
  - `promotion_corr.corr_to_book(candidate: dict[str, float], book: dict[str, float], min_overlap: int) -> float | None` — Pearson over the overlapping dates; `None` if fewer than `min_overlap` overlapping days.
  - `promotion_corr.passes_backstop(corr: float | None, book_empty: bool, max_corr: float) -> bool` — `True` if the book is empty (first edge); else `corr is not None and corr < max_corr` (indeterminate corr → `False` = withhold).

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

```python
# tests/integration/test_promotion_corr.py
from fxhnt.application.promotion_corr import corr_to_book, deploy_book_returns, passes_backstop


def test_deploy_book_is_equal_weight_per_date_average():
    book = deploy_book_returns({"a": {"2026-06-01": 0.02, "2026-06-02": 0.00},
                                "b": {"2026-06-01": 0.00, "2026-06-02": 0.04}})
    assert book == {"2026-06-01": 0.01, "2026-06-02": 0.02}


def test_corr_needs_min_overlap():
    cand = {"2026-06-01": 0.01, "2026-06-02": -0.01}
    book = {"2026-06-01": 0.01, "2026-06-02": -0.01}
    assert corr_to_book(cand, book, min_overlap=5) is None            # only 2 overlapping < 5
    assert abs(corr_to_book(cand, book, min_overlap=2) - 1.0) < 1e-9  # perfectly correlated


def test_backstop_rules():
    assert passes_backstop(None, book_empty=True, max_corr=0.5) is True     # first edge → pass
    assert passes_backstop(None, book_empty=False, max_corr=0.5) is False   # indeterminate → withhold
    assert passes_backstop(0.30, book_empty=False, max_corr=0.5) is True    # low corr → pass
    assert passes_backstop(0.60, book_empty=False, max_corr=0.5) is False   # high corr → withhold
```

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

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

- [ ] **Step 3: Implement `promotion_corr.py`**

```python
# src/fxhnt/application/promotion_corr.py
"""Pure helpers for the promotion correlation backstop (B1): build the equal-weight deploy-book daily
return series and correlate a promotion candidate against it. A candidate promotes only if its daily-return
correlation to the current deploy book is below the policy threshold (or the book is empty — the first edge
trivially diversifies). An indeterminate correlation (too little overlap) withholds promotion — the safe,
capital-adjacent error."""
from __future__ import annotations

import math


def deploy_book_returns(returns_by_sid: dict[str, dict[str, float]]) -> dict[str, float]:
    per_date: dict[str, list[float]] = {}
    for series in returns_by_sid.values():
        for d, r in series.items():
            per_date.setdefault(d, []).append(r)
    return {d: sum(v) / len(v) for d, v in per_date.items()}


def corr_to_book(candidate: dict[str, float], book: dict[str, float], min_overlap: int) -> float | None:
    dates = sorted(set(candidate) & set(book))
    if len(dates) < min_overlap:
        return None
    a = [candidate[d] for d in dates]
    b = [book[d] for d in dates]
    n = len(dates)
    ma, mb = sum(a) / n, sum(b) / n
    cov = sum((x - ma) * (y - mb) for x, y in zip(a, b))
    va = sum((x - ma) ** 2 for x in a)
    vb = sum((y - mb) ** 2 for y in b)
    if va <= 0.0 or vb <= 0.0:
        return None
    return cov / math.sqrt(va * vb)


def passes_backstop(corr: float | None, *, book_empty: bool, max_corr: float) -> bool:
    if book_empty:
        return True
    return corr is not None and corr < max_corr
```

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

Run: `uv run pytest tests/integration/test_promotion_corr.py -q`
Expected: PASS (3 passed). (Note: `passes_backstop` uses a keyword-only `book_empty` — update the test call to `passes_backstop(None, book_empty=True, max_corr=0.5)` as written.)

- [ ] **Step 5: Wire the backstop into `evaluate_promotions`**

In `forward_ingest.py`, extend `evaluate_promotions` so a `promote_on_pass` candidate at PASS is only promoted when the backstop passes. Read the current `evaluate_promotions` first, then:

```python
from fxhnt.application.gate_policy import POLICY
from fxhnt.application.promotion_corr import corr_to_book, deploy_book_returns, passes_backstop
from fxhnt.registry import STRATEGY_REGISTRY

def evaluate_promotions(dsn: str) -> list[str]:
    repo = ForwardNavRepo(dsn); repo.migrate()
    gate = {r.strategy_id: r.gate_status for r in repo.all_summaries()}
    already = repo.promoted_strategy_ids()
    # the current deploy book = registry deploy-tier tracks + already-promoted tracks (effective deploy set),
    # each as its {date: ret} series from A's record.
    deploy_sids = {sid for sid, m in STRATEGY_REGISTRY.items() if m.get("tier") == "deploy"} | already
    def _series(sid: str) -> dict[str, float]:
        return {r.date: r.ret for r in repo.nav_history(sid)}
    newly: list[str] = []
    for sid, meta in STRATEGY_REGISTRY.items():
        if not (meta.get("promote_on_pass") and sid not in already and gate.get(sid) == "PASS"):
            continue
        book_sids = deploy_sids - {sid}                      # never correlate a candidate against itself
        book = deploy_book_returns({s: _series(s) for s in book_sids}) if book_sids else {}
        corr = corr_to_book(_series(sid), book, POLICY.promotion_min_overlap_days) if book else None
        if not passes_backstop(corr, book_empty=not book, max_corr=POLICY.promotion_max_corr):
            _log.info("evaluate_promotions: %s gate=PASS but corr backstop failed (corr=%s, max=%.2f) "
                      "→ promotion withheld", sid, f"{corr:.2f}" if corr is not None else "n/a",
                      POLICY.promotion_max_corr)
            continue
        repo.promote(sid, dt.datetime.now(dt.timezone.utc).replace(tzinfo=None))
        newly.append(sid)
        _log.info("evaluate_promotions: %s cleared gate + corr backstop → PROMOTED research→deploy", sid)
    return newly
```

- [ ] **Step 6: Add an integration test for the backstop end to end + run**

```python
# add to tests/integration/test_promotion.py (or the promotion test file)
def test_promotion_withheld_when_correlated_and_fires_when_diversifying(tmp_path, monkeypatch):
    import datetime as dt
    from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
    from fxhnt.application.forward_ingest import evaluate_promotions
    from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO, ForwardSummary
    from fxhnt import registry
    dsn = f"sqlite:///{tmp_path}/promo.db"
    repo = ForwardNavRepo(dsn); repo.migrate()
    at = dt.datetime(2026, 7, 9, 23, 30)
    # a deploy-tier book track "d" and a PASS candidate "c" flagged promote_on_pass
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "d",
                        {"display_name": "D", "sleeve": "x", "gate_spec": {}, "tier": "deploy"})
    monkeypatch.setitem(registry.STRATEGY_REGISTRY, "c",
                        {"display_name": "C", "sleeve": "x", "gate_spec": {}, "tier": "research",
                         "promote_on_pass": True})
    dates = [f"2026-06-{d:02d}" for d in range(1, 21)]
    dret = [0.01 if i % 2 else -0.01 for i in range(20)]
    for d, r in zip(dates, dret):
        repo.upsert_rows([NavRowDTO(strategy_id="d", date=d, ret=r, nav=1.0)], at=at)
    repo.upsert_summary(ForwardSummary(strategy_id="c", as_of=dates[-1], days=20, nav=1.0,
                                       total_return=0.0, sharpe=0.0, maxdd=0.0), "PASS", "ok", at)
    # candidate PERFECTLY correlated with the book → withheld
    for d, r in zip(dates, dret):
        repo.upsert_rows([NavRowDTO(strategy_id="c", date=d, ret=r, nav=1.0)], at=at)
    assert evaluate_promotions(dsn) == []
    # candidate ANTI-correlated (diversifying) → promoted
    for d, r in zip(dates, dret):
        repo.upsert_rows([NavRowDTO(strategy_id="c", date=d, ret=-r, nav=1.0)], at=at)
    assert evaluate_promotions(dsn) == ["c"]
```

Run: `uv run pytest tests/integration/test_promotion_corr.py tests/integration/test_promotion.py -q`
Expected: PASS.

- [ ] **Step 7: Full suite + commit**

Run: `uv run pytest -q`
Expected: PASS (all).

```bash
git add src/fxhnt/application/promotion_corr.py src/fxhnt/application/forward_ingest.py tests/integration/test_promotion_corr.py tests/integration/test_promotion.py
git commit -F - <<'EOF'
feat(policy): promotion correlation backstop — human flag + data-confirmed diversification (B1)

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

---

## Self-Review notes

- **Spec coverage:** policy object (§1)→T1; resolution (§2)→T2; gate wiring / remove scatter (§1-2)→T3; verdict stamp (§3)→T4+T5; promotion backstop (§4)→T6; byte-identical verdicts (error-handling)→T3 regression anchor; hash self-check→T1; WAIT-safe preserved (verdict logic unchanged)→T3.
- **`promotion_min_overlap_days`** (the spec's open parameter) is a `GatePolicy` field (=10), resolved in T6.
- **Deploy set** = registry `tier=="deploy"` ∪ `promoted_strategy_ids()`, minus the candidate itself (T6) — the effective deploy set A's dashboard already uses.
- **Type consistency:** `resolve_policy(POLICY, gate_spec) -> ResolvedPolicy` is the single resolution; both gates + `_gate_verdict` consume it. `upsert_summary(..., policy_version, policy_hash)` matches the stamp in T5. `deploy_book_returns`/`corr_to_book`/`passes_backstop` signatures match T6's caller.
- **No capital:** T6 promotes to deploy-tier (tracking) only; no allocation/capital (B2).
