diff --git a/app/scheduler.py b/app/scheduler.py
index 400b53f..4f7f9a3 100644
--- a/app/scheduler.py
+++ b/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
-    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
+    When run as a pipeline step it acts only on the scan that stamped *this
+    pipeline's* run id (``expected_run_id``): if the pipeline's own scan was
+    disabled or failed, the stored run id is some other scan's — including a
+    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:
diff --git a/app/services/pipeline_run.py b/app/services/pipeline_run.py
new file mode 100644
index 0000000..f38fc66
--- /dev/null
+++ b/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)
diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py
index 2139fc9..cc56ce3 100644
--- a/app/services/rr_scanner_service.py
+++ b/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
-# scan_all_tickers completes, so a consumer can tell a scan actually ran this
-# pipeline pass (freshness of COMPLETED) and which setups belong to it
-# (detected_at >= STARTED). The shadow book relies on both.
+# Boundary of the most recent *successful* scan. Written together, only when
+# scan_all_tickers completes: STARTED bounds which setups belong to the run
+# (detected_at >= STARTED), COMPLETED gives its freshness, and RUN_ID identifies
+# 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
-    # book refuses to trade unless COMPLETED is fresh (proving a scan ran in this
-    # pipeline pass, not a prior session) and selects only setups from this run
-    # (detected_at >= STARTED). Written after the loop so a hard failure above
-    # leaves the previous, now-stale, marker in place.
+    # Record the run boundary only now that the scan has completed, stamped with
+    # the run id of the pipeline this scan ran inside (or a fresh id when run
+    # standalone — a manual scan then can never match a pipeline's expected id).
+    # All three markers are written in one commit so a reader sees a consistent
+    # (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
diff --git a/app/services/shadow_book_service.py b/app/services/shadow_book_service.py
index 53b8a9f..9ff00d2 100644
--- a/app/services/shadow_book_service.py
+++ b/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
-    passes the applicable check:
+    * ``expected_run_id`` set (pipeline step): the stored run id must match it
+      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
-      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.
+    The three markers are written in one commit, so the returned STARTED belongs
+    to the same run as the matched RUN_ID and correctly bounds its setups.
     """
     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 completed < require_scan_after:
+    if expected_run_id is not None:
+        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
-    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``.
+    ``expected_run_id`` binds this run to the scan that stamped that exact id
+    (the pipeline's own scan), so a scan that failed in this pipeline — or a
+    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:
diff --git a/tests/unit/test_pipeline_run.py b/tests/unit/test_pipeline_run.py
new file mode 100644
index 0000000..ca32618
--- /dev/null
+++ b/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
diff --git a/tests/unit/test_shadow_book_service.py b/tests/unit/test_shadow_book_service.py
index 4e0f791..41edcb5 100644
--- a/tests/unit/test_shadow_book_service.py
+++ b/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):
-        """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."""
+    async def test_pipeline_scan_run_id_match_is_accepted(self, session):
+        """The scan that stamped this pipeline's run id is the one to act on."""
         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))
+        now = datetime.now(timezone.utc)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
         await session.commit()
-        await _mark_scan(session, started=manual_scan - timedelta(minutes=5),
-                         completed=manual_scan)
+        await _mark_scan(session, started=now - timedelta(minutes=1),
+                         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):
-        """The pipeline's own scan completes just after the pipeline began."""
+    async def test_pipeline_scan_failed_leaves_prior_run_id(self, session):
+        """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)
-        scan_completed = pipeline_start + timedelta(minutes=1)
-        session.add(_setup(ids["AAA"], rank=0.9, detected=scan_completed))
+        now = datetime.now(timezone.utc)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
         await session.commit()
-        await _mark_scan(session, started=pipeline_start + timedelta(seconds=1),
-                         completed=scan_completed)
+        await _mark_scan(session, started=now - timedelta(minutes=1),
+                         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: