fix(market-data): integrate tested consumer recovery patches

DATA_LAYER_SERVICE_ACCESS_GUIDE.md

@@ -99,6 +99,22 @@ execution or risk authority except a separately declared fresh
 `MARK_INDEX_PRICE` read. A new alpha must add its own manifest requirement
 before calling a reference metric; it must not reuse another alpha's identity.
 
+### Execution MARK/INDEX Read Contract
+
+Trading System's `latest_execution_feed(..., MARK_INDEX_PRICE)` uses
+`POST /v2/market-data/reference:batch` with EXECUTION grade, one complete row,
+`limit=page_size=max_pages=1`, and the identity's signed freshness/session policy.
+The Kafka-native backend reads its verified market cache; this is not venue REST.
+Component/session evidence may admit unchanged values only within the signed
+component cadence. Disconnect, gap, generation/config mismatch or expired
+component evidence remains blocked. Original source timestamps are preserved.
+
+Generic `snapshot(MARK_INDEX_PRICE)` is a different read contract. Its event-age
+rejection does not establish a failure of the component-aware execution reference
+path. Tests must name the endpoint they actually exercised and retain both
+outcomes; never silently substitute one to turn a failing gate green. TRADE
+session liveness likewise never makes an old last-trade price execution-eligible.
+
 ## 1. Connection Architecture
 
 Production services should **never** connect directly to external exchanges (Binance, DNSE, etc.) if they are running within the `bobby_network`. Instead, they must use the `data_layer` as a unified gateway.

qdl/adapters/binance/bar_edge.py

@@ -7,7 +7,10 @@
 from typing import Callable
 
 from app.providers.binance.rest import BinanceProviderError, fetch_klines
-from qdl.adapters.intervals import BarHistoryGapError, canonical_interval_ms, latest_closed_boundary_ms
+from qdl.adapters.intervals import (
+    BarHistoryGapError, BarHistoryOverlapError, canonical_interval_ms,
+    is_valid_bar_open_ms, latest_closed_boundary_ms,
+)
 from qdl.provider.v1 import raw_provider_pb2
 from qdl.raw.capture import capture_exact_frame
 
@@ -236,6 +239,43 @@ def fetch_closed_bar_history_raw_envelopes(
             observed_ms=observed_ms,
             interval_ms=interval_ms,
         )
+        if len(closed) > page_limit or any(int(row[0]) > page_end_ms for row in closed):
+            raise RuntimeError("Binance closed-bar history page exceeds requested end boundary")
+        for row in closed:
+            open_time = int(row[0])
+            previous = selected_by_open.get(open_time)
+            if previous is not None and previous != row:
+                raise RuntimeError("Binance history pages conflict for one open time")
+            selected_by_open[open_time] = row
+        selected = tuple(selected_by_open[key] for key in sorted(selected_by_open))
+        # Walk newest first: only the suffix after the newest discontinuity can
+        # be certified, including when that discontinuity crosses REST pages.
+        for index in range(len(selected) - 1, 0, -1):
+            previous, current = int(selected[index - 1][0]), int(selected[index][0])
+            if current - previous == interval_ms:
+                continue
+            error = BarHistoryGapError("Binance", binding.native_symbol, binding.interval,
+                                       previous, current)
+            suffix = selected[index:]
+            latest_open = latest_closed_boundary_ms(
+                binding.interval, observed_ms, provider="BINANCE"
+            ) - interval_ms
+            if (error.kind == "OVERLAPPING_PROVIDER_WINDOWS" and binding.interval == "3d"
+                    and int(suffix[-1][0]) == latest_open
+                    and all(is_valid_bar_open_ms(binding.interval, int(row[0]), provider="BINANCE")
+                            for row in suffix)):
+                received_at_ns = time.time_ns()
+                envelopes = tuple(_capture_row(
+                    binding, row, origin="BACKFILLED", received_at_ns=received_at_ns + offset,
+                    test_provenance=test_provenance,
+                ) for offset, row in enumerate(suffix))
+                raise BarHistoryOverlapError(
+                    "Binance", binding.native_symbol, binding.interval, previous, current,
+                    requested_rows=limit, recent_envelopes=envelopes,
+                )
+            raise error
+        if any(int(row[6]) > page_end_ms for row in closed):
+            raise RuntimeError("Binance closed-bar history page exceeds requested end boundary")
         exhausted = False
         if len(closed) != page_limit:
             if not (allow_short and len(closed) < page_limit and _provider_exhausted_before(
@@ -247,18 +287,6 @@ def fetch_closed_bar_history_raw_envelopes(
                     f"requested={page_limit} observed={len(closed)}"
                 )
             exhausted = True
-        if any(int(row[6]) > page_end_ms for row in closed):
-            raise RuntimeError(
-                "Binance closed-bar history page exceeds requested end boundary"
-            )
-        for row in closed:
-            open_time = int(row[0])
-            previous = selected_by_open.get(open_time)
-            if previous is not None and previous != row:
-                raise RuntimeError(
-                    "Binance history pages conflict for one open time"
-                )
-            selected_by_open[open_time] = row
         if exhausted or not closed:
             break
         earliest_open_ms = min(int(row[0]) for row in closed)

qdl/adapters/intervals.py

@@ -23,6 +23,30 @@ def __init__(self, provider: str, symbol: str, interval: str, previous: int, cur
         super().__init__(f"{provider} closed-bar history discontinuity kind={self.kind} "
                          f"symbol={symbol} interval={interval} previous_ms={previous} current_ms={current}")
 
+    def retry_delay_seconds(self, attempt: int) -> int:
+        # Overlapping native windows cannot be repaired by fetching a missing
+        # bar. Recheck for provider corrections daily, retaining failed coverage.
+        if self.kind == "OVERLAPPING_PROVIDER_WINDOWS":
+            return 86_400
+        return min(2 ** min(attempt, 6), 30)
+
+
+class BarHistoryOverlapError(BarHistoryGapError):
+    """Incomplete requested history with an independently validated recent suffix.
+
+    Only history materialization may use these envelopes. Request fetchers still
+    raise: this is never a successful short response or proof of a listing floor.
+    """
+
+    def __init__(self, provider: str, symbol: str, interval: str, previous: int,
+                 current: int, *, requested_rows: int, recent_envelopes: tuple):
+        super().__init__(provider, symbol, interval, previous, current)
+        if self.kind != "OVERLAPPING_PROVIDER_WINDOWS" or not recent_envelopes:
+            raise ValueError("history overlap outcome requires a nonempty recent suffix")
+        self.requested_rows = requested_rows
+        self.recent_envelopes = recent_envelopes
+        self.older_prefix_status = "UNCERTIFIED_PROVIDER_OVERLAP"
+
 
 _UNIT_MS = {
     "s": 1_000,

qdl/consumer/manifest.py

@@ -102,6 +102,8 @@ def __post_init__(self) -> None:
             raise ValueError("consumer manifest contains an unknown data-plane permission")
         if self.execution_dependency not in {"FORBIDDEN", "PAPER_ONLY", "ALLOWED"}:
             raise ValueError("consumer execution dependency policy is invalid")
+        if self.environment in {"sandbox", "live"} and self.execution_dependency == "PAPER_ONLY":
+            raise ValueError("sandbox/live consumer cannot declare PAPER_ONLY execution dependency")
         if not self.requirements:
             raise ValueError("consumer manifest requires at least one data requirement")
         if self.rollback_contract not in {"V1", "V2"}:

qdl/reference/execution_live.py

@@ -208,7 +208,8 @@ async def fetch(
             request,
             capability,
             code,
-            "execution MARK/INDEX live view did not return a current active record",
+            "execution MARK/INDEX live view did not return a current active record: "
+            + ",".join(dict.fromkeys(errors)),
         )
 
     def stats(self) -> dict[str, int]:
@@ -236,8 +237,13 @@ def _bounded_reason(response: httpx.Response) -> str:
         except (ValueError, AttributeError):
             return "UNAVAILABLE"
         normalized = str(detail).upper()
-        if normalized.endswith(":STALE"):
-            return "STALE"
+        # Component cadence and session age are freshness failures, not an
+        # absent provider. Preserve the bounded reason for consumer diagnosis.
+        reason = normalized.rsplit(":", 1)[-1]
+        if reason in {"STALE", "COMPONENT_STALE", "SESSION_LIVENESS"}:
+            return reason
+        if reason in {"SESSION_STATE", "QUIET_POLICY_UNAVAILABLE", "LINEAGE_INVALID"}:
+            return reason
         if normalized.endswith(":GAP_OR_RESYNC"):
             return "GAPPED"
         if normalized.endswith(":IDENTITY_MISMATCH"):
@@ -250,7 +256,7 @@ def _bounded_reason(response: httpx.Response) -> str:
 
     @staticmethod
     def _failure_code(errors: list[str]) -> str:
-        if "STALE" in errors:
+        if any(reason in {"STALE", "COMPONENT_STALE", "SESSION_LIVENESS"} for reason in errors):
             return "LIVE_VIEW_STALE"
         if "GAPPED" in errors:
             return "LIVE_VIEW_GAPPED"

qdl/runtime/kn_query_backend.py

@@ -47,7 +47,7 @@
 from qdl.adapters.intervals import canonical_interval_ms
 from qdl.common.v1 import common_pb2
 from qdl.marketdata.v2 import market_data_pb2
-from qdl.query import DataRequirement, FeedType, GapRecord, HistoryResult, MarketDataItem, RecoveryPolicy
+from qdl.query import CoverageStatus, DataRequirement, FeedType, GapRecord, HistoryResult, MarketDataItem, RecoveryPolicy
 from qdl.query.contracts import CanonicalErrorCode, QueryProblem
 from qdl.query.results import NON_REPLAYABLE_STREAM_CURSOR, QueryBackendError, GapScanResult
 from qdl.query.row_cache import BoundedRowCache
@@ -368,6 +368,12 @@ def history_with_envelopes(
         )
         if result is None:
             return None, ()
+        # Native Binance 3d has overlapping historical grids. A retained recent
+        # suffix does not prove listing exhaustion or certify an older prefix.
+        if (binding.feed is FeedType.BAR and binding.interval == "3d"
+                and binding.instrument.identity.venue == "BINANCE"
+                and start_ns is None and len(result.items) < requested):
+            result = replace(result, coverage=CoverageStatus.PARTIAL)
         selected = records
         if start_ns is not None:
             selected = tuple(

qdl/runtime/stable.py

@@ -526,7 +526,12 @@ def stable_grpc_server_credentials(
 
 def load_stable_manifests(config: StableRuntimeConfig) -> ConsumerManifestRegistry:
     manifests = tuple(ConsumerManifestLoader.load(path) for path in config.manifest_paths)
-    if any(item.environment != config.environment for item in manifests):
+    # Public market storage/cursors keep their realm. Consumer realms require
+    # explicit, separately pinned JWT key/subject bindings.
+    raw_realms = os.environ.get("QDL_DATA_JWT_KEY_ENVIRONMENTS_JSON")
+    realms = (DataPlaneSecurityConfig.from_environment().consumer_environments
+              if raw_realms else frozenset({config.environment}))
+    if any(item.environment not in realms for item in manifests):
         raise ValueError("stable consumer manifest environment mismatch")
     return ConsumerManifestRegistry(manifests)