# C3 — Execution Unification (bybit→CLI) + Paper-Validation Envelope + Exec-Venue — 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:** Port the Bybit execution leg off Dagster into a `execute-bybit` CLI command (clean-cut the assets/job/schedule), give both execution legs a uniform `--paper-envelope` flag that bypasses B2a on paper-only accounts, and record + surface the real execution venue in the cockpit.

**Architecture:** Execution becomes CLI + CronJob for BOTH legs (Dagster keeps only pipelining). A shared `settings.paper_validation.envelope` default backs an identical `--paper-envelope` flag on `execute-multistrat` and the new `execute-bybit`; when >0 the code hard-refuses on any non-paper/non-testnet account and sizes `min(paper_envelope, nlv)` without ever reading B2a. The actual venue string is persisted in existing JSONB exec book-state and shown in the "Executed reality" badge.

**Tech Stack:** Python 3.12, pydantic-settings, Typer CLI, Dagster (asset removal), SQLAlchemy/Postgres (JSONB book-state), Jinja2, pytest.

## Global Constraints

- **Real money errs false-WAIT.** Paper envelope MUST hard-refuse (place ZERO orders) on any non-paper/non-testnet account.
- Paper mode NEVER reads/writes `strategy_allocation` (B2a real-capital policy untouched).
- `--paper-envelope 0` (default) ⇒ behavior byte-identical to today.
- No schema migration — `venue` rides in existing JSONB book-state.
- Venue format exactly `f"{broker.name}-{'paper' if is_paper else 'live'}"`; bybit records literal `"bybit-testnet"`.
- **Faithful port:** the `execute-bybit` path preserves EVERY guard the assets had — kill-switch (zero orders), per-day NAV-marker idempotency, open-orders resting guard, per-order `BrokerError`→gap row, low-confidence slippage exclusion, funding-cursor advance. Relocation, not rewrite.
- Cockpit badge fallback: `{{ d.exec_venue or d.venue }}`.
- Verify the cockpit LOCALLY in a real browser (read the numbers) — never claim "works" off prod.

---

### Task 1: `PaperValidationSettings` config

**Files:**
- Modify: `src/fxhnt/config.py` (add class after `AlpacaSettings` ~line 55; wire into `Settings` ~line 170)
- Test: `tests/unit/test_config_paper.py`

**Interfaces:**
- Produces: `settings.paper_validation.envelope: float` (default `0.0`), env `FXHNT_PAPER_VALIDATION_ENVELOPE`.

- [ ] **Step 1: Write the failing test** — append to `tests/unit/test_config_paper.py`:

```python
def test_paper_validation_envelope_default_off():
    from fxhnt.config import Settings
    assert Settings().paper_validation.envelope == 0.0


def test_paper_validation_envelope_env_override(monkeypatch):
    from fxhnt.config import Settings
    monkeypatch.setenv("FXHNT_PAPER_VALIDATION_ENVELOPE", "1000000")
    assert Settings().paper_validation.envelope == 1_000_000.0
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_config_paper.py -k paper_validation -v`
Expected: FAIL `AttributeError: 'Settings' object has no attribute 'paper_validation'`

- [ ] **Step 3: Implement** — in `src/fxhnt/config.py` after `AlpacaSettings`:

```python
class PaperValidationSettings(BaseSettings):
    """Fixed PAPER-only execution envelope for the exec-validation tracks (multistrat_exec, bybit_4edge_exec).
    0.0 = OFF (exec tracks stay B2a-gated). >0 = the exec book runs continuously at this fixed paper $ size,
    bypassing B2a — but ONLY against a paper/testnet account (hard-refused on live). B2a is never read in
    paper mode."""
    model_config = SettingsConfigDict(env_prefix="FXHNT_PAPER_VALIDATION_")
    envelope: float = 0.0
```

In `class Settings`, after `bybit_exec` (~line 170):

```python
    paper_validation: PaperValidationSettings = Field(default_factory=PaperValidationSettings)
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_config_paper.py -k paper_validation -v`
Expected: PASS (2 passed)

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/config.py tests/unit/test_config_paper.py
git commit -m "feat(config): PaperValidationSettings.envelope (paper-only exec envelope, default off)"
```

---

### Task 2: multistrat paper-envelope gate + venue in book-state

**Files:**
- Modify: `src/fxhnt/application/multistrat_exec_record.py` (`build_pos_state` ~21, `record_exec_track` ~54, `plan_and_record` ~64)
- Test: `tests/unit/test_multistrat_exec_helpers.py`

**Interfaces:**
- Produces:
  - `build_pos_state(positions, prices, envelope, venue="") -> dict` (adds `"venue"`).
  - `record_exec_track(repo, exec_return, new_positions, today_prices, envelope, today, at, venue="")`.
  - `plan_and_record(*, repo, exec_svc, within_weights, prices, nlv, today, at, max_gross, execute, paper_envelope=0.0, account_is_paper=True, venue="") -> tuple[plan, float]`.

- [ ] **Step 1: Write the failing tests** — append to `tests/unit/test_multistrat_exec_helpers.py`:

```python
def test_build_pos_state_persists_venue():
    from fxhnt.application.multistrat_exec_record import build_pos_state
    st = build_pos_state({"SPY": 10.0}, {"SPY": 5.0}, 1000.0, venue="alpaca-paper")
    assert st["venue"] == "alpaca-paper" and st["envelope"] == 1000.0


def test_build_pos_state_venue_defaults_empty():
    from fxhnt.application.multistrat_exec_record import build_pos_state
    assert build_pos_state({}, {}, 0.0)["venue"] == ""


class _FakeExecSvc:
    def __init__(self):
        self.calls = []

    def rebalance_weights(self, sid, weights, prices, *, max_gross, execute):
        self.calls.append((sid, dict(weights), execute))
        class _Plan:
            nlv = 100000.0; orders = []; notes = []; blocked = False; executed = execute
        return _Plan()


class _FakeRepo:
    def __init__(self, alloc):
        self._alloc = alloc; self.book = {}

    def current_allocation(self):  # B2a — must NOT be called in paper mode
        raise AssertionError("current_allocation must not be read in paper mode")

    def get_book_state(self, k):
        return self.book.get(k, {})

    def set_book_state(self, k, extra, at):
        self.book[k] = extra


class _B2aRepo(_FakeRepo):
    def current_allocation(self):
        return self._alloc


def test_plan_and_record_paper_mode_bypasses_b2a():
    import datetime as dt
    from fxhnt.application.multistrat_exec_record import plan_and_record
    repo = _FakeRepo(alloc={"multistrat": 0.0}); svc = _FakeExecSvc()
    _plan, envelope = plan_and_record(
        repo=repo, exec_svc=svc, within_weights={"SPY": 1.0}, prices={"SPY": 10.0},
        nlv=500000.0, today="2026-07-11", at=dt.datetime(2026, 7, 11), max_gross=2.0,
        execute=False, paper_envelope=1_000_000.0, account_is_paper=True, venue="alpaca-paper")
    assert envelope == 500000.0  # min(1e6 paper, 5e5 nlv); B2a NOT consulted


def test_plan_and_record_paper_mode_refuses_live_account():
    import datetime as dt
    import pytest
    from fxhnt.application.multistrat_exec_record import plan_and_record
    repo = _FakeRepo(alloc={"multistrat": 0.0}); svc = _FakeExecSvc()
    with pytest.raises(ValueError, match="paper-envelope refused"):
        plan_and_record(
            repo=repo, exec_svc=svc, within_weights={"SPY": 1.0}, prices={"SPY": 10.0},
            nlv=500000.0, today="2026-07-11", at=dt.datetime(2026, 7, 11), max_gross=2.0,
            execute=True, paper_envelope=1_000_000.0, account_is_paper=False, venue="alpaca-live")
    assert svc.calls == []  # NO orders on refusal


def test_plan_and_record_default_mode_reads_b2a():
    import datetime as dt
    from fxhnt.application.multistrat_exec_record import plan_and_record
    repo = _B2aRepo(alloc={"multistrat": 25000.0}); svc = _FakeExecSvc()
    _plan, envelope = plan_and_record(
        repo=repo, exec_svc=svc, within_weights={"SPY": 1.0}, prices={"SPY": 10.0},
        nlv=500000.0, today="2026-07-11", at=dt.datetime(2026, 7, 11), max_gross=2.0,
        execute=False, paper_envelope=0.0, account_is_paper=True)
    assert envelope == 25000.0  # min($25k B2a, $500k nlv)
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_multistrat_exec_helpers.py -k "venue or paper_mode or default_mode" -v`
Expected: FAIL (unexpected kwargs `venue`, `paper_envelope`)

- [ ] **Step 3: Implement** — in `src/fxhnt/application/multistrat_exec_record.py`:

Replace `build_pos_state`:
```python
def build_pos_state(positions: dict[str, float], prices: dict[str, float], envelope: float,
                    venue: str = "") -> dict[str, Any]:
    return {"positions": {s: float(q) for s, q in positions.items()},
            "prices": {s: float(p) for s, p in prices.items()}, "envelope": float(envelope),
            "venue": str(venue)}
```

Replace `record_exec_track`:
```python
def record_exec_track(repo: Any, exec_return: float | None, new_positions: dict[str, float],
                      today_prices: dict[str, float], envelope: float, today: str, at: dt.datetime,
                      venue: str = "") -> None:
    """Book `exec_return` as an observed forward_record for `multistrat_exec` via A's run_track, then carry
    the next run's position-state (incl. the execution `venue`) under book-state key `multistrat_exec_pos`."""
    from fxhnt.application.forward_engine import run_track
    run_track(repo, "multistrat_exec", _ExecBookingStrategy(exec_return), today=today, at=at)
    repo.set_book_state("multistrat_exec_pos", build_pos_state(new_positions, today_prices, envelope, venue), at=at)
```

Replace `plan_and_record` — new signature + envelope branch (keep the rest of the body identical):
```python
def plan_and_record(*, repo: Any, exec_svc: Any, within_weights: dict[str, float], prices: dict[str, float],
                    nlv: float, today: str, at: dt.datetime, max_gross: float, execute: bool,
                    paper_envelope: float = 0.0, account_is_paper: bool = True,
                    venue: str = "") -> tuple[Any, float]:
    """... (existing docstring) ...

    `paper_envelope > 0` sizes the book to min(paper_envelope, nlv) and does NOT read B2a — a PAPER run that
    bypasses the real-capital gate, hard-refused (no orders) unless `account_is_paper`. `venue` is recorded."""
    if paper_envelope > 0.0:
        if not account_is_paper:
            raise ValueError("paper-envelope refused: account is not paper (real-capital exposure guard)")
        envelope = min(float(paper_envelope), float(nlv))
    else:
        envelope = min(float(repo.current_allocation().get("multistrat", 0.0)), float(nlv))
    weights = scaled_weights(within_weights, envelope, nlv)
    plan = exec_svc.rebalance_weights("multistrat", weights, prices, max_gross=max_gross, execute=execute)
    if execute and not plan.blocked:
        pos_prior = repo.get_book_state("multistrat_exec_pos")
        exec_return = compute_exec_return(pos_prior, prices)
        positions_after = {s: (w * float(nlv)) / prices[s]
                           for s, w in weights.items() if prices.get(s, 0.0) > 0.0}
        record_exec_track(repo, exec_return, positions_after, prices, envelope, today, at, venue)
    return plan, envelope
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_multistrat_exec_helpers.py -v`
Expected: PASS (all)

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/application/multistrat_exec_record.py tests/unit/test_multistrat_exec_helpers.py
git commit -m "feat(exec): multistrat paper-envelope gate (bypasses B2a, paper-only) + venue in book-state"
```

---

### Task 3: `execute-multistrat` CLI — `--paper-envelope` flag + venue threading

**Files:**
- Modify: `src/fxhnt/cli.py` (`execute_multistrat` ~276-347; add two module helpers)
- Test: `tests/unit/test_cli_execute_helpers.py` (create)

**Interfaces:**
- Consumes: `plan_and_record(..., paper_envelope, account_is_paper, venue)` (Task 2); `settings.paper_validation.envelope`; `brk.name`, `acct.is_paper`.
- Produces: module helpers `_exec_venue_string(broker_name, is_paper) -> str`, `_resolve_paper_envelope(flag, config_default) -> float`; CLI option `--paper-envelope FLOAT` (default 0.0).

- [ ] **Step 1: Write the failing test** — create `tests/unit/test_cli_execute_helpers.py`:

```python
def test_venue_string():
    from fxhnt.cli import _exec_venue_string
    assert _exec_venue_string("alpaca", True) == "alpaca-paper"
    assert _exec_venue_string("ibkr", True) == "ibkr-paper"
    assert _exec_venue_string("alpaca", False) == "alpaca-live"


def test_resolve_paper_envelope():
    from fxhnt.cli import _resolve_paper_envelope
    assert _resolve_paper_envelope(1_000_000.0, 0.0) == 1_000_000.0   # flag wins
    assert _resolve_paper_envelope(0.0, 500_000.0) == 500_000.0       # config fallback
    assert _resolve_paper_envelope(0.0, 0.0) == 0.0                   # off
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_cli_execute_helpers.py -v`
Expected: FAIL `ImportError: cannot import name '_exec_venue_string'`

- [ ] **Step 3: Implement** — in `src/fxhnt/cli.py`, add near `execute_multistrat`:

```python
def _exec_venue_string(broker_name: str, is_paper: bool) -> str:
    return f"{broker_name}-{'paper' if is_paper else 'live'}"


def _resolve_paper_envelope(flag: float, config_default: float) -> float:
    """Explicit flag wins; otherwise the shared settings default. Both default 0.0 = off (B2a behavior)."""
    return flag if flag > 0.0 else config_default
```

Add the option to `execute_multistrat`'s signature (after `broker_name`):
```python
    paper_envelope: float = typer.Option(
        0.0, "--paper-envelope",
        help="PAPER-only validation envelope in $ (bypasses B2a; refused on live). 0 = B2a-gated."),
```

Replace the `with broker as brk:` block's `plan_and_record` call:
```python
        with broker as brk:
            svc = ExecutionService(data, brk, settings)
            acct = brk.account_state()
            paper_env = _resolve_paper_envelope(paper_envelope, settings.paper_validation.envelope)
            venue = _exec_venue_string(brk.name, acct.is_paper)
            plan, envelope = plan_and_record(
                repo=repo, exec_svc=svc, within_weights=bweights, prices=bprices, nlv=acct.nlv,
                today=today, at=at, max_gross=max_gross, execute=do_execute,
                paper_envelope=paper_env, account_is_paper=acct.is_paper, venue=venue)
```

Change the summary echo (line ~341) — drop the now-inaccurate `(B2a)` literal:
```python
    typer.echo(f"NLV ${plan.nlv:,.0f}  | envelope ${envelope:,.0f} | orders {len(plan.orders)}")
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_cli_execute_helpers.py -v`
Expected: PASS (2 passed)

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/cli.py tests/unit/test_cli_execute_helpers.py
git commit -m "feat(cli): execute-multistrat --paper-envelope flag + record real exec venue"
```

---

### Task 4: bybit paper-envelope helper + venue in book-state

**Files:**
- Modify: `src/fxhnt/application/bybit_exec_record.py` (`build_crypto_pos_state` ~18, after `crypto_envelope` ~14, `record_bybit_exec_track` ~77)
- Test: `tests/unit/test_bybit_exec_record.py`

**Interfaces:**
- Produces:
  - `build_crypto_pos_state(positions, marks, envelope, funding_cursor_ts, venue="") -> dict` (adds `"venue"`).
  - `paper_or_b2a_crypto_envelope(repo, equity, paper_envelope, is_paper) -> float`.
  - `record_bybit_exec_track(repo, exec_return, new_positions, today_marks, envelope, funding_cursor_ts, today, at, venue="")`.

- [ ] **Step 1: Write the failing tests** — append to `tests/unit/test_bybit_exec_record.py`:

```python
def test_build_crypto_pos_state_persists_venue():
    from fxhnt.application.bybit_exec_record import build_crypto_pos_state
    st = build_crypto_pos_state({"BTCUSDT": 1.0}, {"BTCUSDT": 60000.0}, 5000.0, 123, venue="bybit-testnet")
    assert st["venue"] == "bybit-testnet" and st["funding_cursor_ts"] == 123


def test_paper_or_b2a_crypto_envelope_paper_mode():
    from fxhnt.application.bybit_exec_record import paper_or_b2a_crypto_envelope
    class _Repo:
        def current_allocation(self):
            raise AssertionError("B2a must not be read in paper mode")
    assert paper_or_b2a_crypto_envelope(_Repo(), equity=8000.0, paper_envelope=10000.0, is_paper=True) == 8000.0


def test_paper_or_b2a_crypto_envelope_refuses_non_testnet():
    import pytest
    from fxhnt.application.bybit_exec_record import paper_or_b2a_crypto_envelope
    with pytest.raises(ValueError, match="paper-envelope refused"):
        paper_or_b2a_crypto_envelope(object(), equity=8000.0, paper_envelope=10000.0, is_paper=False)


def test_paper_or_b2a_crypto_envelope_default_reads_b2a():
    from fxhnt.application.bybit_exec_record import paper_or_b2a_crypto_envelope
    class _Repo:
        def current_allocation(self):
            return {"bybit_4edge": 3000.0}
    assert paper_or_b2a_crypto_envelope(_Repo(), equity=8000.0, paper_envelope=0.0, is_paper=True) == 3000.0
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_bybit_exec_record.py -k "venue or paper_or_b2a" -v`
Expected: FAIL (unexpected kwarg `venue`; cannot import `paper_or_b2a_crypto_envelope`)

- [ ] **Step 3: Implement** — in `src/fxhnt/application/bybit_exec_record.py`:

Replace `build_crypto_pos_state`:
```python
def build_crypto_pos_state(positions: dict[str, float], marks: dict[str, float], envelope: float,
                           funding_cursor_ts: int, venue: str = "") -> dict[str, Any]:
    return {"positions": {s: float(q) for s, q in positions.items()},
            "marks": {s: float(m) for s, m in marks.items()},
            "envelope": float(envelope), "funding_cursor_ts": int(funding_cursor_ts),
            "venue": str(venue)}
```

Add after `crypto_envelope`:
```python
def paper_or_b2a_crypto_envelope(repo: Any, equity: float, paper_envelope: float, is_paper: bool) -> float:
    """PAPER-validation envelope (min(paper, equity)) when `paper_envelope > 0`, bypassing B2a — refused on a
    non-testnet account (never widen real-capital exposure). Otherwise the B2a/B3-gated `crypto_envelope`."""
    if paper_envelope > 0.0:
        if not is_paper:
            raise ValueError("paper-envelope refused: bybit account is not testnet/paper")
        return min(float(paper_envelope), float(equity))
    return crypto_envelope(repo, equity)
```

Replace `record_bybit_exec_track` to thread `venue`:
```python
def record_bybit_exec_track(repo: Any, exec_return: float | None, new_positions: dict[str, float],
                            today_marks: dict[str, float], envelope: float, funding_cursor_ts: int,
                            today: str, at: dt.datetime, venue: str = "") -> None:
    """... (existing docstring) ... `venue` is recorded on the exec book-state."""
    from fxhnt.application.forward_engine import run_track
    run_track(repo, "bybit_4edge_exec", _CryptoExecBookingStrategy(exec_return), today=today, at=at)
    repo.set_book_state("bybit_4edge_exec_pos",
                        build_crypto_pos_state(new_positions, today_marks, envelope, funding_cursor_ts, venue),
                        at=at)
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_bybit_exec_record.py tests/integration/test_record_bybit_exec_track.py -v`
Expected: PASS (all — the integration test's existing `record_bybit_exec_track` calls still work via the `venue=""` default)

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/application/bybit_exec_record.py tests/unit/test_bybit_exec_record.py
git commit -m "feat(exec): bybit paper-envelope helper (testnet-only) + venue in book-state"
```

---

### Task 5: Port the Bybit exec leg into `run_bybit_testnet` (application function)

**Files:**
- Create: `src/fxhnt/application/bybit_testnet_run.py`
- Read (source of the port): `src/fxhnt/adapters/orchestration/assets.py:1106-1365` (the `_testnet_*` seams + `bybit_testnet_reconcile` + `bybit_testnet_record`)
- Test: `tests/unit/test_bybit_testnet_run.py` (create)

**Interfaces:**
- Consumes: `paper_or_b2a_crypto_envelope`, `record_bybit_exec_track(..., venue=)` (Task 4).
- Produces: `run_bybit_testnet(settings, *, paper_envelope: float, execute: bool, log: Callable[[str], None]) -> dict` and the five moved seams `_testnet_store/_testnet_universe/_testnet_unlock_events/_testnet_marks/_testnet_adapter` (module-level, monkeypatchable).

**Port method (faithful relocation — do NOT rewrite the logic):**
1. Copy the five `_testnet_*` helper functions verbatim from `assets.py:1109-1143` into the new module (they already do `from ... import ...` locally; keep as-is).
2. Copy the BODY of `bybit_testnet_reconcile` (`assets.py:1156-1289`) then the BODY of `bybit_testnet_record` (`assets.py:1297-1365`) into `run_bybit_testnet`, concatenated (reconcile computes `recon`-equivalent locals; record consumes them directly — no dict handoff).
3. Mechanical transforms, applied uniformly:
   - `context.log.info(x)` / `context.log.warning(x)` / `context.log.error(x)` → `log(x)`.
   - Replace the reconcile's early returns (`noop`/`killed`/`aborted`) with: `log(...)` then `return {...}` summary (record step is skipped for these — mirror the record asset's `if recon.get("noop")/killed/aborted: return`).
   - Envelope line (`assets.py:1208`): replace
     `envelope = crypto_envelope(fwd_repo, equity)` with
     `envelope = paper_or_b2a_crypto_envelope(fwd_repo, equity, paper_envelope, is_paper=settings.bybit_exec.testnet)`
     and change the import at the top from `crypto_envelope` to `paper_or_b2a_crypto_envelope`.
   - **Dry-run:** guard order placement + all writes on `execute`. When `execute is False`: run `plan_orders`, compute the slippage estimate against `marks` if cheaply available, but place NO orders, write NO fills/NAV, and DON'T advance the funding cursor — return a summary with `"placed": 0, "dry_run": True`. (Mirror `execute-multistrat`'s dry-run: plan only.)
   - The `record_bybit_exec_track(...)` call gains `venue="bybit-testnet"`.
   - `s = get_settings()` becomes the passed-in `settings` param (drop the local `get_settings()` for the top-level; keep it out).
4. Return a summary dict: `{"run_date", "placed", "equity", "envelope", "realized_funding", "realized_fees", "mean_exec_slippage_bps", "mean_timing_drift_bps", "exec_return", "venue": "bybit-testnet", "dry_run"}` (superset is fine).

- [ ] **Step 1: Write the failing tests** — create `tests/unit/test_bybit_testnet_run.py`. Use the monkeypatchable seams + a fake adapter to avoid Timescale/ccxt (mirror how the deleted assets were tested; see `tests/integration/test_bybit_testnet_job_wiring.py` for the seam-swap pattern):

```python
import fxhnt.application.bybit_testnet_run as run_mod


class _FakeAdapter:
    def __init__(self): self.placed = []
    def equity(self): return 8000.0
    def positions(self): return {}
    def instrument_limits(self): return {}
    def open_orders(self): return []
    def mid(self, sym): return 100.0
    def funding_since(self, cursor): return 0.0
    def place_market(self, sym, qty, reduce_only=False):
        self.placed.append((sym, qty)); raise AssertionError("kill/dry-run must not place")


def _patch_seams(monkeypatch, adapter):
    monkeypatch.setattr(run_mod, "_testnet_adapter", lambda s: adapter)
    monkeypatch.setattr(run_mod, "_testnet_store", lambda s: _FakeStore())
    monkeypatch.setattr(run_mod, "_testnet_universe", lambda store: None)
    monkeypatch.setattr(run_mod, "_testnet_unlock_events", lambda s: [])
    monkeypatch.setattr(run_mod, "_testnet_marks", lambda store, universe=None: {"BTCUSDT": 100.0})


class _FakeStore:
    def close(self): ...


def test_kill_switch_places_nothing(monkeypatch, tmp_path):
    from fxhnt.config import Settings
    s = Settings(); s.bybit_exec.kill_switch = True
    adapter = _FakeAdapter(); _patch_seams(monkeypatch, adapter)
    # also stub weights so plan_orders has input; monkeypatch latest_raw_sleeve_weights/combined_symbol_weights
    monkeypatch.setattr(run_mod, "latest_raw_sleeve_weights", lambda *a, **k: {})
    monkeypatch.setattr(run_mod, "combined_symbol_weights", lambda *a, **k: ({}, "2026-07-11"))
    out = run_bybit_testnet_helper(s, adapter, monkeypatch)  # see helper below
    assert out.get("killed") is True and adapter.placed == []
```

(Adjust seam/import names to whatever the port exposes; the essential assertions are: kill-switch ⇒ `placed == 0` and no exception, and a dry-run ⇒ `placed == 0`. Keep the fakes minimal — the point is to prove the guards survived the port, not to re-test slippage math already covered elsewhere.)

Add a dry-run test:
```python
def test_dry_run_places_nothing_and_writes_no_nav(monkeypatch):
    from fxhnt.application.bybit_testnet_run import run_bybit_testnet
    from fxhnt.config import Settings
    s = Settings()
    adapter = _FakeAdapter(); _patch_seams(monkeypatch, adapter)
    monkeypatch.setattr(run_mod, "latest_raw_sleeve_weights", lambda *a, **k: {})
    monkeypatch.setattr(run_mod, "combined_symbol_weights", lambda *a, **k: ({}, "2026-07-11"))
    logs = []
    out = run_bybit_testnet(s, paper_envelope=0.0, execute=False, log=logs.append)
    assert out["placed"] == 0 and out.get("dry_run") is True and adapter.placed == []
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_bybit_testnet_run.py -v`
Expected: FAIL `ModuleNotFoundError: No module named 'fxhnt.application.bybit_testnet_run'`

- [ ] **Step 3: Implement** — create `src/fxhnt/application/bybit_testnet_run.py` following the Port method above. Module skeleton:

```python
"""Bybit testnet execution leg as a plain application function (ported off the Dagster reconcile/record
assets — same guards, now CLI-driven). `run_bybit_testnet` reconciles the bybit_4edge book on TESTNET and
records fills/NAV + the observed bybit_4edge_exec return in ONE process. Dry-run (execute=False) plans only."""
from __future__ import annotations

import datetime as dt
from typing import Any, Callable

# the five monkeypatchable seams (moved verbatim from assets.py) ...
def _testnet_store(s): ...
def _testnet_universe(store): ...
def _testnet_unlock_events(s): ...
def _testnet_marks(store, universe=None): ...
def _testnet_adapter(s): ...


def run_bybit_testnet(settings: Any, *, paper_envelope: float, execute: bool,
                      log: Callable[[str], None]) -> dict[str, Any]:
    ...  # ported reconcile + record body, with the 4 mechanical transforms
```

Copy the reconcile+record bodies and apply the transforms. Keep the module-level `from ... import ...` that the seams and body use INSIDE functions where the assets had them (they import locally). Where the reconcile body imports `latest_raw_sleeve_weights, combined_symbol_weights` etc., keep those imports at module level so the tests can `monkeypatch.setattr(run_mod, "latest_raw_sleeve_weights", ...)`.

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_bybit_testnet_run.py -v`
Expected: PASS

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/application/bybit_testnet_run.py tests/unit/test_bybit_testnet_run.py
git commit -m "feat(exec): port bybit testnet leg to run_bybit_testnet (faithful, +paper-envelope +venue)"
```

---

### Task 6: `execute-bybit` CLI command

**Files:**
- Modify: `src/fxhnt/cli.py` (add `execute_bybit` command near `execute_multistrat`)
- Test: covered by Task 5's `run_bybit_testnet` tests + a smoke test here.

**Interfaces:**
- Consumes: `run_bybit_testnet(settings, paper_envelope, execute, log)` (Task 5); `_resolve_paper_envelope` (Task 3).

- [ ] **Step 1: Write the failing test** — append to `tests/unit/test_cli_execute_helpers.py`:

```python
def test_execute_bybit_command_registered():
    from fxhnt.cli import app
    names = {c.name for c in app.registered_commands}
    assert "execute-bybit" in names
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_cli_execute_helpers.py -k execute_bybit -v`
Expected: FAIL `assert 'execute-bybit' in ...`

- [ ] **Step 3: Implement** — in `src/fxhnt/cli.py`:

```python
@app.command("execute-bybit")
def execute_bybit(
    do_execute: bool = typer.Option(False, "--execute", help="place orders (default: dry-run plan only)"),
    paper_envelope: float = typer.Option(
        0.0, "--paper-envelope",
        help="PAPER-only validation envelope in $ (bypasses B2a; refused on live). 0 = B2a-gated."),
) -> None:
    """Reconcile + record the bybit_4edge book on Bybit TESTNET (ported off Dagster). Dry-run by default;
    same safety guards (kill-switch, NAV-marker idempotency, open-orders guard). Records bybit_4edge_exec."""
    from fxhnt.application.bybit_testnet_run import run_bybit_testnet

    settings = get_settings()
    paper_env = _resolve_paper_envelope(paper_envelope, settings.paper_validation.envelope)
    out = run_bybit_testnet(settings, paper_envelope=paper_env, execute=do_execute, log=typer.echo)
    typer.echo(f"-> placed {out.get('placed', 0)} | envelope ${out.get('envelope', 0.0):,.0f} "
               f"| exec_return {out.get('exec_return')}")
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/unit/test_cli_execute_helpers.py -v`
Expected: PASS

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/cli.py tests/unit/test_cli_execute_helpers.py
git commit -m "feat(cli): execute-bybit command (bybit testnet exec off Dagster, --paper-envelope)"
```

---

### Task 7: Remove the Dagster bybit exec assets + wiring (clean cut)

**Files:**
- Modify: `src/fxhnt/adapters/orchestration/assets.py` (delete `_testnet_*` helpers ~1106-1143, `bybit_testnet_reconcile` ~1146-1289, `bybit_testnet_record` ~1292-1365)
- Modify: `src/fxhnt/adapters/orchestration/definitions.py` (remove imports + job + schedule)
- Modify: `tests/integration/test_orchestration_definitions.py` (assert bybit exec job/schedule GONE)
- Delete: `tests/integration/test_bybit_testnet_job_wiring.py` (the job it wired no longer exists; the exec logic is now tested via Task 5)

**Interfaces:**
- Produces: `defs` with no bybit exec assets/job/schedule; `combined_book_forward_job` untouched.

- [ ] **Step 1: Write/adjust the failing test** — in `tests/integration/test_orchestration_definitions.py`, add:

```python
def test_bybit_testnet_exec_removed_from_definitions():
    from fxhnt.adapters.orchestration.definitions import defs
    job_names = {j.name for j in defs.jobs}
    assert "bybit_testnet_execution_job" not in job_names
    sched_names = {s.name for s in defs.schedules}
    assert "bybit_testnet_daily" not in sched_names
    asset_keys = {k.to_python_identifier() for k in defs.get_asset_graph().all_asset_keys}
    assert "bybit_testnet_reconcile" not in asset_keys
    assert "bybit_testnet_record" not in asset_keys
```

(If `test_orchestration_definitions.py` already asserts these exist, update those assertions to the negative form. If `defs.jobs`/`defs.schedules` iteration differs in this Dagster version, adapt to the file's existing access pattern — mirror how the file already reads `defs`.)

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/integration/test_orchestration_definitions.py -k removed -v`
Expected: FAIL (job/schedule/assets still present)

- [ ] **Step 3: Implement**

In `assets.py`: delete the block from the `# ---- Bybit testnet real-paper execution leg` comment (~1106) through the end of `bybit_testnet_record` (~1365). Do NOT touch `cockpit_forward` (starts ~1368) or `bybit_4edge_nav`/`bybit_paper_book`/other assets.

In `definitions.py`:
- Remove `bybit_testnet_reconcile,` and `bybit_testnet_record,` from the import block (lines 11-12).
- Delete the entire bybit exec block (lines 49-65): the comment, `from fxhnt.config import get_settings`, `bybit_testnet_job`, `bybit_testnet_schedule`.
- In `Definitions(...)`: remove `bybit_testnet_reconcile, bybit_testnet_record,` from `assets=[...]`; remove `bybit_testnet_job` from `jobs=[...]`; remove `bybit_testnet_schedule` from `schedules=[...]`.

Delete `tests/integration/test_bybit_testnet_job_wiring.py`:
```bash
git rm tests/integration/test_bybit_testnet_job_wiring.py
```

- [ ] **Step 4: Run — expect PASS + module still loads**
Run: `SQLX_OFFLINE=true pytest tests/integration/test_orchestration_definitions.py -v`
Expected: PASS
Run: `SQLX_OFFLINE=true python -c "from fxhnt.adapters.orchestration.definitions import defs; print(len(defs.jobs), 'jobs')"`
Expected: prints `1 jobs` (only `combined_book_forward_job`), no ImportError.

- [ ] **Step 5: Commit**
```bash
git add src/fxhnt/adapters/orchestration/assets.py src/fxhnt/adapters/orchestration/definitions.py \
        tests/integration/test_orchestration_definitions.py
git rm tests/integration/test_bybit_testnet_job_wiring.py
git commit -m "refactor(dagster): remove bybit exec assets/job/schedule (leg now CLI-driven)"
```

---

### Task 8: cockpit — surface `exec_venue` in the badge

**Files:**
- Modify: `src/fxhnt/ports/dashboard.py` (`StrategyDetail` ~line 84 — add `exec_venue`)
- Modify: `src/fxhnt/application/dashboard_service.py` (`detail` ~169-192; add `_safe_book_venue`)
- Modify: `src/fxhnt/adapters/web/templates/strategy.html` (line 47 badge)
- Test: `tests/integration/test_dashboard_surfacing.py`

**Interfaces:**
- Consumes: `self._repo.get_book_state(f"{twin}_pos")` → `.get("venue")`.
- Produces: `StrategyDetail.exec_venue: str | None`; template renders `{{ d.exec_venue or d.venue }}`.

- [ ] **Step 1: Write the failing test** — append to `tests/integration/test_dashboard_surfacing.py`, following the file's existing repo-seeding pattern (seed a `multistrat` + `multistrat_exec` summary, then a `multistrat_exec_pos` book-state with `venue`):

```python
def test_detail_exec_venue_from_book_state(tmp_path):
    # ... seed via the same ForwardNavRepo pattern the other tests in this file use:
    #   summaries for "multistrat" and "multistrat_exec"
    #   repo.set_book_state("multistrat_exec_pos", {"positions": {}, "prices": {}, "envelope": 1e6,
    #                        "venue": "ibkr-paper"}, at=<naive utc>)
    svc = DashboardService(repo)
    d = svc.detail("multistrat")
    assert d.exec_twin_sid == "multistrat_exec"
    assert d.exec_venue == "ibkr-paper"


def test_detail_exec_venue_none_when_unrecorded(tmp_path):
    # same seeding but the multistrat_exec_pos book-state has NO "venue" key (pre-C3 anchor)
    svc = DashboardService(repo)
    d = svc.detail("multistrat")
    assert d.exec_venue is None
```

- [ ] **Step 2: Run — expect FAIL**
Run: `SQLX_OFFLINE=true pytest tests/integration/test_dashboard_surfacing.py -k exec_venue -v`
Expected: FAIL `AttributeError: 'StrategyDetail' object has no attribute 'exec_venue'`

- [ ] **Step 3: Implement**

In `src/fxhnt/ports/dashboard.py`, add to `class StrategyDetail`:
```python
    exec_venue: str | None = None
```

In `src/fxhnt/application/dashboard_service.py`, add to `DashboardService`:
```python
    def _safe_book_venue(self, twin_sid: str) -> str | None:
        """Execution venue recorded on the exec twin's book-state; None if absent/unreachable (fallback)."""
        if not twin_sid:
            return None
        try:
            return self._repo.get_book_state(f"{twin_sid}_pos").get("venue") or None
        except Exception:  # noqa: BLE001 — a missing book-state must never break the dashboard
            return None
```

In `detail()`, add `pair = self._exec_pair(strategy_id, summaries)` immediately above `return StrategyDetail(`, and replace the trailing `**self._exec_pair(strategy_id, summaries),` with:
```python
            **pair,
            exec_venue=self._safe_book_venue(pair["exec_twin_sid"]),
```

In `src/fxhnt/adapters/web/templates/strategy.html` line 47:
```html
  <div class="lct">Executed reality <span class="r"><span class="exec-tag">{{ d.exec_venue or d.venue }}</span></span></div>
```

- [ ] **Step 4: Run — expect PASS**
Run: `SQLX_OFFLINE=true pytest tests/integration/test_dashboard_surfacing.py -k exec_venue -v`
Expected: PASS

- [ ] **Step 5: Local cockpit verification (read the numbers)**

```bash
SQLX_OFFLINE=true python - <<'PY'
import datetime as dt
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
from fxhnt.config import get_settings
r = ForwardNavRepo(get_settings().operational_dsn); r.migrate()
st = r.get_book_state("multistrat_exec_pos") or {"positions": {}, "prices": {}, "envelope": 1e6}
st["venue"] = "ibkr-paper"; r.set_book_state("multistrat_exec_pos", st, at=dt.datetime.now(dt.UTC).replace(tzinfo=None))
print("seeded multistrat_exec_pos venue=ibkr-paper")
PY
./scripts/dev-cockpit-local.sh    # 127.0.0.1:8801 (read-only)
```
Playwright → `http://127.0.0.1:8801/strategy/multistrat`, screenshot, **read** that the "Executed reality" badge shows `ibkr-paper`.

- [ ] **Step 6: Commit**
```bash
git add src/fxhnt/ports/dashboard.py src/fxhnt/application/dashboard_service.py \
        src/fxhnt/adapters/web/templates/strategy.html tests/integration/test_dashboard_surfacing.py
git commit -m "feat(cockpit): surface recorded exec venue in Executed-reality badge (fallback to configured)"
```

---

### Task 9: infra — arm alpaca rebalancer + new bybit rebalancer CronJob

**Files:**
- Modify: `infra/k8s/jobs/fxhnt-alpaca-rebalancer.yaml` (args + `suspend`)
- Create: `infra/k8s/jobs/fxhnt-bybit-rebalancer.yaml`

**Interfaces:**
- Consumes: `execute-multistrat --paper-envelope` (Task 3), `execute-bybit --paper-envelope` (Task 6).

- [ ] **Step 1: Alpaca rebalancer** — append `--paper-envelope 1000000` to the container's `fxhnt execute-multistrat --execute` arg string (preserve the rest verbatim), and set `suspend: false`.

- [ ] **Step 2: Bybit rebalancer** — create `infra/k8s/jobs/fxhnt-bybit-rebalancer.yaml` by copying `fxhnt-alpaca-rebalancer.yaml` and changing:
  - `metadata.name` → `fxhnt-bybit-rebalancer`; labels `app.kubernetes.io/name: fxhnt-bybit-rebalancer`.
  - container command → `fxhnt execute-bybit --execute --paper-envelope 1000000`.
  - env: replace `FXHNT_ALPACA_*` (from `alpaca-credentials`) with `FXHNT_BYBIT_EXEC_API_KEY`/`FXHNT_BYBIT_EXEC_API_SECRET` from the bybit exec secret (`secretKeyRef` name `bybit-testnet-credentials`, keys `api-key`/`api-secret`); keep `FXHNT_OPERATIONAL_DSN` + `PGPASSWORD`.
  - NetworkPolicy (if the alpaca job ships one): egress to Postgres 5432 + gitea (git-sync) + external 443 (Bybit API). Bybit is an EXTERNAL host — no in-cluster pod selector needed, just 443 egress.
  - `suspend: true` (arm only after `bybit-testnet-credentials` is populated).

- [ ] **Step 3: Validate manifests**
Run: `kubectl apply --dry-run=client -f infra/k8s/jobs/fxhnt-alpaca-rebalancer.yaml -f infra/k8s/jobs/fxhnt-bybit-rebalancer.yaml`
Expected: both `configured/created (dry run)`, no error.

- [ ] **Step 4: Commit**
```bash
git add infra/k8s/jobs/fxhnt-alpaca-rebalancer.yaml infra/k8s/jobs/fxhnt-bybit-rebalancer.yaml
git commit -m "chore(infra): arm alpaca --paper-envelope + new fxhnt-bybit-rebalancer CronJob"
```

---

## Full-suite gate (before finishing)

- [ ] Ensure no other pytest is running, then run once clean:
```bash
pkill -9 -f pytest 2>/dev/null; sleep 1
SQLX_OFFLINE=true pytest tests/unit tests/integration -q
```
Expected: all pass (watch for exit-144/SIGTERM from concurrent runs — re-run singly if seen).

## Deployment (post-merge, controller-run — NOT TDD tasks)

1. Push `master`; `./scripts/argo-deploy-cockpit.sh` (build + deploy). No schema migration.
2. Revert `strategy_allocation[multistrat]` → `$0` (or let the nightly recompute it). Cockpit then shows the honest B2a state.
3. Cancel the queued IBKR paper orders on DU9600528 via a one-off `ib_async` job.
4. Arm `fxhnt-alpaca-rebalancer` (`suspend: false`); once `bybit-testnet-credentials` is confirmed, arm `fxhnt-bybit-rebalancer`.
5. Verify the cockpit LOCALLY, then confirm on `dashboard.fxhnt.ai` that the multistrat "Executed reality" badge shows the real venue and the allocation reads the honest B2a value.

## Self-Review

- **Spec coverage:** Component 1 (bybit→CLI port) → Tasks 5,6,7; Component 2 (paper envelope) → Tasks 1,2,3,4,5,6; Component 3 (venue) → Tasks 2,4,5,8; Component 4 (cockpit) → Task 8; Component 5 (wiring) → Task 9 + Deployment. Faithful-port guards → Task 5 (kill/dry-run tests) + preserved logic. ✓
- **Types consistent:** `venue` param default `""` at every record site; `exec_venue: str | None` in the DTO; `paper_envelope: float`/`account_is_paper: bool` uniform across `plan_and_record`, `paper_or_b2a_crypto_envelope`, `run_bybit_testnet`, and both CLI commands. ✓
- **Clean cut:** Task 7 removes assets + definitions wiring + the wiring test; `test_orchestration_definitions.py` asserts absence; module-load check confirms no orphaned imports. ✓
- **No placeholders:** every code step shows code; every test step shows the command + expected result. The port task (5) is a faithful relocation with enumerated mechanical transforms + explicit source line ranges (re-transcribing 220 lines verbatim would be error-prone; the transforms are exact). ✓
