fix: bind shadow book to its own pipeline's scan, not wall-clock freshness

A 6-hour freshness window proves only that some scan ran recently, which a
manual mid-day scan satisfies. Scenario: a manual scan succeeds at 13:00;
the 15:30 near-close pipeline's scan step is disabled or fails; at 15:30
the 13:00 completion is still 'fresh', so the shadow step trades that
earlier batch despite no successful scan in the current pipeline.

_run_pipeline now records its start in a per-task contextvar, visible to
the steps it awaits. run_shadow_book reads it and requires the scan
completion marker to be at/after the pipeline start, so a scan that failed
or was disabled in this pass (marker left at a prior run, before the
pipeline began) cannot be substituted by an earlier manual scan. A direct
Admin trigger has no pipeline context and falls back to the freshness
window -- an explicit operator action, not an automated one.

Tests pin the reported case: a fresh manual scan predating the pipeline
start is refused; the pipeline's own post-start scan is accepted.

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

app/scheduler.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import json
 import logging
 import asyncio
+import contextvars
 from datetime import date, datetime, timedelta, timezone
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
@@ -625,9 +626,16 @@ async def run_shadow_book() -> None:
     prices the discretionary book sees, leaving *selection* as the only
     difference between the two books.
 
+    When run as a pipeline step it only acts on a scan that completed *inside
+    this pipeline pass* (``require_scan_after``): if the pipeline's own scan was
+    disabled or failed, an earlier still-fresh manual scan must not stand in for
+    it. Triggered directly from Admin (no pipeline context) it falls back to the
+    scan-freshness window — an explicit operator action.
+
     Opt-in (``shadow_book_enabled``) because it writes live trades.
     """
     job_name = "shadow_book"
+    require_scan_after = _pipeline_started_at.get()
     _log_event(logging.INFO, "job_start", job=job_name)
     _runtime_start(job_name, total=1)
 
@@ -646,7 +654,9 @@ async def run_shadow_book() -> None:
 
             activation_config = await get_activation_config(db)
             summary = await shadow_book_service.open_shadow_positions(
-                db, activation_config=activation_config
+                db,
+                activation_config=activation_config,
+                require_scan_after=require_scan_after,
             )
             symbols = await shadow_book_service.symbols_for(db, summary["symbols"])
 
@@ -1319,6 +1329,15 @@ _INTRADAY_PIPELINE_STEPS = [
 _NEAR_CLOSE_DURATION_WARN_SECONDS = 600
 
 
+# Wall-clock start of the pipeline invocation currently running, visible to its
+# steps via the shared task context. The shadow book uses it to require that the
+# scan it acts on completed *inside this pipeline pass* — a fresh but earlier
+# manual scan must not stand in for a scan that failed or was disabled here.
+_pipeline_started_at: contextvars.ContextVar[datetime | None] = contextvars.ContextVar(
+    "pipeline_started_at", default=None
+)
+
+
 async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
     """Run an ordered list of (step_name, coroutine_name) steps.
 
@@ -1337,6 +1356,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
 
     funcs = globals()
     done = 0
+    token = _pipeline_started_at.set(datetime.now(timezone.utc))
     try:
         for step_name, func_name in steps:
             _runtime_progress(job_name, processed=done, total=total, current_ticker=step_name)
@@ -1350,6 +1370,8 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
     except Exception as exc:
         _runtime_finish(job_name, "error", processed=done, total=total, message=str(exc))
         _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
+    finally:
+        _pipeline_started_at.reset(token)
 
 
 async def run_daily_pipeline() -> None:

app/services/shadow_book_service.py

@@ -169,14 +169,28 @@ async def _shadow_user_id(db: AsyncSession) -> int | None:
     return int(row[0]) if row else None
 
 
-async def _last_scan_start(db: AsyncSession, *, now: datetime) -> datetime | None:
-    """Start of the last successful scan, if it ran in this pipeline pass.
+async def _last_scan_start(
+    db: AsyncSession,
+    *,
+    now: datetime,
+    require_scan_after: datetime | None = None,
+) -> datetime | None:
+    """Start of the last successful scan, if it is the one we may act on.
 
-    Returns None — meaning "no scan to act on" — unless the scanner's COMPLETED
-    marker is fresh. Pipeline steps fail independently, so a scan that was
-    disabled, errored, or produced nothing leaves a stale marker; trading on the
-    newest stored setups then would enter a previous session's picks at stale
-    prices. Freshness is proven by the marker, not by setup age.
+    Returns None — "no scan to act on" — unless the scanner's COMPLETED marker
+    passes the applicable check:
+
+    * ``require_scan_after`` set (pipeline step): the scan must have completed
+      at/after the pipeline began. This is the airtight guarantee — a scan that
+      was disabled or failed in *this* pipeline leaves the marker at a previous
+      run, and a fresh but earlier manual scan completed before the pipeline
+      started, so neither can stand in for the pipeline's own scan.
+    * ``require_scan_after`` None (direct Admin trigger): fall back to the
+      freshness window. There is no pipeline scan to bind to, so acting on a
+      recent scan is the operator's explicit choice.
+
+    Either way the returned value is the scan's start, the lower bound for the
+    setups belonging to that run.
     """
     from app.services import rr_scanner_service as rr
 
@@ -186,7 +200,10 @@ async def _last_scan_start(db: AsyncSession, *, now: datetime) -> datetime | Non
     )
     if started is None or completed is None:
         return None
-    if now - completed > MAX_SCAN_AGE:
+    if require_scan_after is not None:
+        if completed < require_scan_after:
+            return None
+    elif now - completed > MAX_SCAN_AGE:
         return None
     return started
 
@@ -201,7 +218,11 @@ def _parse_dt(raw: str | None) -> datetime | None:
 
 
 async def _todays_qualified_setups(
-    db: AsyncSession, config: dict, *, now: datetime
+    db: AsyncSession,
+    config: dict,
+    *,
+    now: datetime,
+    require_scan_after: datetime | None = None,
 ) -> list[TradeSetup]:
     """Long-only qualified setups from the scan that just ran, best rank first.
 
@@ -219,7 +240,9 @@ async def _todays_qualified_setups(
        the reverse.
     4. Qualify, then rank by ``strategy_rank`` (unranked sort last).
     """
-    run_start = await _last_scan_start(db, now=now)
+    run_start = await _last_scan_start(
+        db, now=now, require_scan_after=require_scan_after
+    )
     if run_start is None:
         return []
 
@@ -252,12 +275,17 @@ async def open_shadow_positions(
     *,
     activation_config: dict,
     opened_at: datetime | None = None,
+    require_scan_after: datetime | None = None,
 ) -> dict:
     """Fill free capacity with the top-ranked qualified setups.
 
     Mirrors the backtest: rank the qualified cross-section, walk it top-down,
     skip anything already held or locked out by post-stop gate-reset, and stop
     at capacity. Returns a summary for the job log.
+
+    ``require_scan_after`` binds this run to a scan that completed at/after that
+    instant (the pipeline's start), so a stale or manual scan cannot substitute
+    for a scan that failed in this pipeline pass. See ``_last_scan_start``.
     """
     summary = {
         "opened": 0,
@@ -283,7 +311,10 @@ async def open_shadow_positions(
     equity, cash = await equity_and_cash(db, config["start_equity"], positions)
     timestamp = opened_at or datetime.now(timezone.utc)
 
-    for setup in await _todays_qualified_setups(db, activation_config, now=timestamp):
+    candidates = await _todays_qualified_setups(
+        db, activation_config, now=timestamp, require_scan_after=require_scan_after
+    )
+    for setup in candidates:
         if free_slots <= 0:
             break
         if setup.ticker_id in held:

tests/unit/test_shadow_book_service.py

@@ -283,6 +283,44 @@ class TestScanFreshness:
         assert summary["opened"] == 0
 
 
+class TestPipelineScanBinding:
+    @pytest.mark.asyncio
+    async def test_fresh_manual_scan_before_pipeline_is_refused(self, session):
+        """A manual scan at 13:00 is still 'fresh' at 15:30, but the 15:30
+        pipeline's own scan failed. Binding to the pipeline start rejects the
+        13:00 batch — no successful scan happened in *this* pipeline pass."""
+        ids = await _seed(session, ["AAA"])
+        pipeline_start = datetime.now(timezone.utc)
+        manual_scan = pipeline_start - timedelta(hours=2, minutes=30)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=manual_scan))
+        await session.commit()
+        await _mark_scan(session, started=manual_scan - timedelta(minutes=5),
+                         completed=manual_scan)
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG, require_scan_after=pipeline_start
+        )
+
+        assert summary["opened"] == 0
+
+    @pytest.mark.asyncio
+    async def test_pipeline_scan_after_start_is_accepted(self, session):
+        """The pipeline's own scan completes just after the pipeline began."""
+        ids = await _seed(session, ["AAA"])
+        pipeline_start = datetime.now(timezone.utc)
+        scan_completed = pipeline_start + timedelta(minutes=1)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=scan_completed))
+        await session.commit()
+        await _mark_scan(session, started=pipeline_start + timedelta(seconds=1),
+                         completed=scan_completed)
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG, require_scan_after=pipeline_start
+        )
+
+        assert summary["opened"] == 1
+
+
 class TestLongOnly:
     @pytest.mark.asyncio
     async def test_shorts_are_never_taken_even_with_gate_disabled(self, session):