fix: match shadow book to its pipeline's scan by run id, not timestamp

A manually triggered rr_scanner and the scheduled near-close pipeline are
separate APScheduler jobs; max_instances=1 serialises a job only against
itself, so they can overlap. A manual scan starting just before the
pipeline can finish just after it began and overwrite the scan markers.
Its completion timestamp is then later than the pipeline start, so the
previous 'completed >= pipeline_start' check accepted its batch as though
it were the pipeline's own -- exactly when the pipeline's scan may have
failed.

Replace the timestamp comparison with an exact run-id match. A new
pipeline_run module holds a per-task run-id contextvar (separate module so
the scanner and scheduler import it without a cycle). _run_pipeline binds a
fresh id per invocation; scan_all_tickers stamps that id -- or a fresh one
when run standalone -- into the scan markers, written with started/completed
in a single commit. The shadow step requires the stored run id to equal its
pipeline's id exactly, so a concurrent manual scan (its own id) or a failed
pipeline scan (a prior run's id) can never be mistaken for it. Direct Admin
triggers have no pipeline context and keep the freshness fallback.

Known residual: the id match governs whether shadow proceeds; setup
selection remains detected_at >= scan start, so a fully per-run setup
isolation would need a run_id column on trade_setups (not required here).

Tests cover the reported race (manual scan finishing last is refused), a
failed pipeline scan, the id-match accept path, and contextvar propagation
and non-leakage across tasks.

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

app/scheduler.py

@@ -16,7 +16,6 @@ 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
@@ -37,6 +36,7 @@ from app.providers.protocol import SentimentData
 from app.services import (
     fundamental_service,
     ingestion_service,
+    pipeline_run,
     sentiment_service,
     settings_store,
     shadow_book_service,
@@ -626,16 +626,17 @@ 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
+    When run as a pipeline step it acts only on the scan that stamped *this
-    this pipeline pass* (``require_scan_after``): if the pipeline's own scan was
+    pipeline's* run id (``expected_run_id``): if the pipeline's own scan was
-    disabled or failed, an earlier still-fresh manual scan must not stand in for
+    disabled or failed, the stored run id is some other scan's — including a
-    it. Triggered directly from Admin (no pipeline context) it falls back to the
+    manual scan that overlapped and finished last — and shadow refuses.
+    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()
+    expected_run_id = pipeline_run.current()
     _log_event(logging.INFO, "job_start", job=job_name)
     _runtime_start(job_name, total=1)
 
@@ -656,7 +657,7 @@ async def run_shadow_book() -> None:
             summary = await shadow_book_service.open_shadow_positions(
                 db,
                 activation_config=activation_config,
-                require_scan_after=require_scan_after,
+                expected_run_id=expected_run_id,
             )
             symbols = await shadow_book_service.symbols_for(db, summary["symbols"])
 
@@ -1329,20 +1330,16 @@ _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.
 
     Each step respects its own enable flag and manages its own runtime status; a
     failing step is logged and the pipeline continues with the next one.
+
+    A unique run id is bound for the invocation and visible to every step via the
+    shared task context: the scan step stamps it into its completion markers and
+    the shadow step requires an exact match, so only a scan that ran inside this
+    pipeline can drive the shadow book.
     """
     _log_event(logging.INFO, "job_start", job=job_name)
     async with async_session_factory() as db:
@@ -1356,7 +1353,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))
+    token = pipeline_run.bind(pipeline_run.new_run_id())
     try:
         for step_name, func_name in steps:
             _runtime_progress(job_name, processed=done, total=total, current_ticker=step_name)
@@ -1371,7 +1368,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
         _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)
+        pipeline_run.release(token)
 
 
 async def run_daily_pipeline() -> None:

app/services/pipeline_run.py

@@ -0,0 +1,47 @@
+"""Per-invocation identity for pipeline runs.
+
+A pipeline invocation stamps a unique run id into the task context. The scan it
+runs records that id alongside its completion markers, and the shadow book
+requires an *exact* match before acting on the scan's batch.
+
+This is what timestamp comparison cannot provide. A manually triggered scan and
+the scheduled near-close pipeline are separate APScheduler jobs, and
+``max_instances=1`` only serialises a job against itself — not two different
+jobs. So a manual scan can start just before the pipeline and finish just after
+it began, leaving a completion timestamp later than the pipeline's start even
+though its batch is unrelated. Matching on a run id generated by the pipeline,
+and stamped only by the scan running inside that pipeline, removes the ambiguity.
+
+Lives in its own module so the scheduler (which sets the id), the scanner (which
+stamps it), and the shadow book (which checks it) can all import it without an
+import cycle.
+"""
+
+from __future__ import annotations
+
+import contextvars
+import uuid
+
+_run_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
+    "pipeline_run_id", default=None
+)
+
+
+def new_run_id() -> str:
+    """A fresh, collision-free run id."""
+    return uuid.uuid4().hex
+
+
+def current() -> str | None:
+    """Run id of the pipeline invocation on the current task, if any."""
+    return _run_id.get()
+
+
+def bind(run_id: str) -> contextvars.Token:
+    """Set the current run id; pass the returned token to ``release``."""
+    return _run_id.set(run_id)
+
+
+def release(token: contextvars.Token) -> None:
+    """Restore the previous run id (call in a finally)."""
+    _run_id.reset(token)

app/services/rr_scanner_service.py

@@ -47,12 +47,15 @@ from app.services.recommendation_service import (
 
 logger = logging.getLogger(__name__)
 
-# Boundary of the most recent *successful* scan. Written only when
+# Boundary of the most recent *successful* scan. Written together, only when
-# scan_all_tickers completes, so a consumer can tell a scan actually ran this
+# scan_all_tickers completes: STARTED bounds which setups belong to the run
-# pipeline pass (freshness of COMPLETED) and which setups belong to it
+# (detected_at >= STARTED), COMPLETED gives its freshness, and RUN_ID identifies
-# (detected_at >= STARTED). The shadow book relies on both.
+# the pipeline invocation that produced it (or a fresh id for a manual scan).
+# The shadow book matches RUN_ID exactly rather than trusting timestamps, so a
+# concurrent manual scan cannot be mistaken for the pipeline's own.
 KEY_LAST_SCAN_STARTED = "last_scan_run_started_at"
 KEY_LAST_SCAN_COMPLETED = "last_scan_run_completed_at"
+KEY_LAST_SCAN_RUN_ID = "last_scan_run_id"
 
 STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1"
 
@@ -820,17 +823,22 @@ async def scan_all_tickers(
     if progress_callback is not None and total:
         progress_callback(total, total, "")
 
-    # Record the run boundary only now that the scan has completed. The shadow
+    # Record the run boundary only now that the scan has completed, stamped with
-    # book refuses to trade unless COMPLETED is fresh (proving a scan ran in this
+    # the run id of the pipeline this scan ran inside (or a fresh id when run
-    # pipeline pass, not a prior session) and selects only setups from this run
+    # standalone — a manual scan then can never match a pipeline's expected id).
-    # (detected_at >= STARTED). Written after the loop so a hard failure above
+    # All three markers are written in one commit so a reader sees a consistent
-    # leaves the previous, now-stale, marker in place.
+    # (started, completed, run_id) triple, and a hard failure above leaves the
+    # previous, now-superseded, markers in place.
+    from app.services import pipeline_run
+
+    run_id = pipeline_run.current() or pipeline_run.new_run_id()
     await settings_store.upsert_setting(
         db, KEY_LAST_SCAN_STARTED, gate_observation_started_at.isoformat()
     )
     await settings_store.upsert_setting(
         db, KEY_LAST_SCAN_COMPLETED, datetime.now(timezone.utc).isoformat()
     )
+    await settings_store.upsert_setting(db, KEY_LAST_SCAN_RUN_ID, run_id)
     await db.commit()
 
     return all_setups

app/services/shadow_book_service.py

@@ -173,24 +173,22 @@ async def _last_scan_start(
     db: AsyncSession,
     *,
     now: datetime,
-    require_scan_after: datetime | None = None,
+    expected_run_id: str | None = None,
 ) -> datetime | None:
-    """Start of the last successful scan, if it is the one we may act on.
+    """Start of the scan we may act on, or None if there is none.
 
-    Returns None — "no scan to act on" — unless the scanner's COMPLETED marker
+    * ``expected_run_id`` set (pipeline step): the stored run id must match it
-    passes the applicable check:
+      exactly. This is the airtight guarantee — a scan that was disabled or
+      failed in *this* pipeline never stamped this id, and a concurrent manual
+      scan (a separate APScheduler job, not serialised against the pipeline)
+      stamps its own id even when it finishes last, so neither can be mistaken
+      for the pipeline's own scan. Timestamp order alone cannot tell them apart.
+    * ``expected_run_id`` 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.
 
-    * ``require_scan_after`` set (pipeline step): the scan must have completed
+    The three markers are written in one commit, so the returned STARTED belongs
-      at/after the pipeline began. This is the airtight guarantee — a scan that
+    to the same run as the matched RUN_ID and correctly bounds its setups.
-      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
 
@@ -198,10 +196,11 @@ async def _last_scan_start(
     completed = _parse_dt(
         await settings_store.get_value(db, rr.KEY_LAST_SCAN_COMPLETED)
     )
+    run_id = await settings_store.get_value(db, rr.KEY_LAST_SCAN_RUN_ID)
     if started is None or completed is None:
         return None
-    if require_scan_after is not None:
+    if expected_run_id is not None:
-        if completed < require_scan_after:
+        if not run_id or run_id != expected_run_id:
             return None
     elif now - completed > MAX_SCAN_AGE:
         return None
@@ -222,7 +221,7 @@ async def _todays_qualified_setups(
     config: dict,
     *,
     now: datetime,
-    require_scan_after: datetime | None = None,
+    expected_run_id: str | None = None,
 ) -> list[TradeSetup]:
     """Long-only qualified setups from the scan that just ran, best rank first.
 
@@ -241,7 +240,7 @@ async def _todays_qualified_setups(
     4. Qualify, then rank by ``strategy_rank`` (unranked sort last).
     """
     run_start = await _last_scan_start(
-        db, now=now, require_scan_after=require_scan_after
+        db, now=now, expected_run_id=expected_run_id
     )
     if run_start is None:
         return []
@@ -275,7 +274,7 @@ async def open_shadow_positions(
     *,
     activation_config: dict,
     opened_at: datetime | None = None,
-    require_scan_after: datetime | None = None,
+    expected_run_id: str | None = None,
 ) -> dict:
     """Fill free capacity with the top-ranked qualified setups.
 
@@ -283,9 +282,10 @@ async def open_shadow_positions(
     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
+    ``expected_run_id`` binds this run to the scan that stamped that exact id
-    instant (the pipeline's start), so a stale or manual scan cannot substitute
+    (the pipeline's own scan), so a scan that failed in this pipeline — or a
-    for a scan that failed in this pipeline pass. See ``_last_scan_start``.
+    concurrent manual scan that finished last — cannot substitute for it. See
+    ``_last_scan_start``.
     """
     summary = {
         "opened": 0,
@@ -312,7 +312,7 @@ async def open_shadow_positions(
     timestamp = opened_at or datetime.now(timezone.utc)
 
     candidates = await _todays_qualified_setups(
-        db, activation_config, now=timestamp, require_scan_after=require_scan_after
+        db, activation_config, now=timestamp, expected_run_id=expected_run_id
     )
     for setup in candidates:
         if free_slots <= 0:

tests/unit/test_pipeline_run.py

@@ -0,0 +1,72 @@
+"""Pipeline run-id context and the scanner stamping it into scan markers."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.services import pipeline_run
+
+
+def test_no_run_id_by_default():
+    assert pipeline_run.current() is None
+
+
+def test_bind_and_release_restore_previous():
+    assert pipeline_run.current() is None
+    token = pipeline_run.bind("run-1")
+    try:
+        assert pipeline_run.current() == "run-1"
+    finally:
+        pipeline_run.release(token)
+    assert pipeline_run.current() is None
+
+
+def test_new_run_ids_are_unique():
+    ids = {pipeline_run.new_run_id() for _ in range(100)}
+    assert len(ids) == 100
+
+
+@pytest.mark.asyncio
+async def test_run_id_propagates_to_awaited_coroutines():
+    """The scan and shadow steps are awaited inside the pipeline's task, so they
+    must observe the id the pipeline bound."""
+
+    async def step() -> str | None:
+        return pipeline_run.current()
+
+    token = pipeline_run.bind("run-42")
+    try:
+        assert await step() == "run-42"
+    finally:
+        pipeline_run.release(token)
+
+
+@pytest.mark.asyncio
+async def test_run_id_does_not_leak_into_an_independent_task():
+    """A manual scan is a separate APScheduler job, started independently of the
+    pipeline. Modelled here as a task created before the bind: it captures its
+    own context and never observes the id the pipeline binds afterwards."""
+    seen: dict[str, str | None] = {}
+    manual_started = asyncio.Event()
+    let_manual_finish = asyncio.Event()
+
+    async def manual_job() -> None:
+        manual_started.set()
+        await let_manual_finish.wait()
+        seen["manual"] = pipeline_run.current()
+
+    # Created with no id in context — the manual job predates the pipeline bind.
+    task = asyncio.create_task(manual_job())
+    await manual_started.wait()
+
+    token = pipeline_run.bind("pipeline")
+    try:
+        assert pipeline_run.current() == "pipeline"
+        let_manual_finish.set()
+        await task
+    finally:
+        pipeline_run.release(token)
+
+    assert seen["manual"] is None

tests/unit/test_shadow_book_service.py

@@ -79,7 +79,13 @@ def _setup(
     )
 
 
-async def _mark_scan(session, *, started: datetime, completed: datetime | None = None):
+async def _mark_scan(
+    session,
+    *,
+    started: datetime,
+    completed: datetime | None = None,
+    run_id: str = "scan-run",
+):
     """Record a successful scan run so the shadow book has something to act on."""
     from app.services import rr_scanner_service as rr
 
@@ -90,6 +96,9 @@ async def _mark_scan(session, *, started: datetime, completed: datetime | None =
     await shadow_book_service.settings_store.upsert_setting(
         session, rr.KEY_LAST_SCAN_COMPLETED, completed.isoformat()
     )
+    await shadow_book_service.settings_store.upsert_setting(
+        session, rr.KEY_LAST_SCAN_RUN_ID, run_id
+    )
     await session.commit()
 
 
@@ -285,40 +294,56 @@ class TestScanFreshness:
 
 class TestPipelineScanBinding:
     @pytest.mark.asyncio
-    async def test_fresh_manual_scan_before_pipeline_is_refused(self, session):
+    async def test_pipeline_scan_run_id_match_is_accepted(self, session):
-        """A manual scan at 13:00 is still 'fresh' at 15:30, but the 15:30
+        """The scan that stamped this pipeline's run id is the one to act on."""
-        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)
+        now = datetime.now(timezone.utc)
-        manual_scan = pipeline_start - timedelta(hours=2, minutes=30)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
-        session.add(_setup(ids["AAA"], rank=0.9, detected=manual_scan))
         await session.commit()
-        await _mark_scan(session, started=manual_scan - timedelta(minutes=5),
+        await _mark_scan(session, started=now - timedelta(minutes=1),
-                         completed=manual_scan)
+                         completed=now, run_id="pipeline-A")
 
         summary = await shadow_book_service.open_shadow_positions(
-            session, activation_config=_CONFIG, require_scan_after=pipeline_start
+            session, activation_config=_CONFIG, expected_run_id="pipeline-A"
+        )
+
+        assert summary["opened"] == 1
+
+    @pytest.mark.asyncio
+    async def test_concurrent_manual_scan_finishing_last_is_refused(self, session):
+        """The reported race: a manual rr_scanner overlaps the near-close
+        pipeline and writes the markers last. Its completion timestamp is later
+        than the pipeline start, but its run id is not the pipeline's — so the
+        shadow book must refuse, even though a timestamp check would accept."""
+        ids = await _seed(session, ["AAA"])
+        now = datetime.now(timezone.utc)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
+        await session.commit()
+        # Pipeline expects "pipeline-A"; the manual scan's id won the last write.
+        await _mark_scan(session, started=now - timedelta(minutes=1),
+                         completed=now, run_id="manual-999")
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG, expected_run_id="pipeline-A"
         )
 
         assert summary["opened"] == 0
 
     @pytest.mark.asyncio
-    async def test_pipeline_scan_after_start_is_accepted(self, session):
+    async def test_pipeline_scan_failed_leaves_prior_run_id(self, session):
-        """The pipeline's own scan completes just after the pipeline began."""
+        """If the pipeline's own scan failed, the stored id is a prior run's."""
         ids = await _seed(session, ["AAA"])
-        pipeline_start = datetime.now(timezone.utc)
+        now = datetime.now(timezone.utc)
-        scan_completed = pipeline_start + timedelta(minutes=1)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
-        session.add(_setup(ids["AAA"], rank=0.9, detected=scan_completed))
         await session.commit()
-        await _mark_scan(session, started=pipeline_start + timedelta(seconds=1),
+        await _mark_scan(session, started=now - timedelta(minutes=1),
-                         completed=scan_completed)
+                         completed=now, run_id="yesterday")
 
         summary = await shadow_book_service.open_shadow_positions(
-            session, activation_config=_CONFIG, require_scan_after=pipeline_start
+            session, activation_config=_CONFIG, expected_run_id="pipeline-today"
         )
 
-        assert summary["opened"] == 1
+        assert summary["opened"] == 0
 
 
 class TestLongOnly: