fix(ops): bound the stuck-orders and exception-inbox reads, report truncation (S-8)

docs/openapi.json

@@ -3233,6 +3233,11 @@
             },
             "type": "object",
             "title": "Pagination"
+          },
+          "scan_truncated": {
+            "type": "boolean",
+            "title": "Scan Truncated",
+            "default": false
           }
         },
         "type": "object",
@@ -3261,6 +3266,11 @@
           "manual_resolutions": {
             "type": "integer",
             "title": "Manual Resolutions"
+          },
+          "scan_truncated": {
+            "type": "boolean",
+            "title": "Scan Truncated",
+            "default": false
           }
         },
         "type": "object",
@@ -4451,6 +4461,11 @@
             },
             "type": "object",
             "title": "Pagination"
+          },
+          "scan_truncated": {
+            "type": "boolean",
+            "title": "Scan Truncated",
+            "default": false
           }
         },
         "type": "object",

src/serving/api/routers/ops.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import math
+import os
 from datetime import UTC, datetime, timedelta
 from typing import Any, Literal, cast
 
@@ -30,6 +31,7 @@
     check_journal_vs_store,
     check_stuck_replay,
     journal_scan_limit,
+    orders_scan_limit,
 )
 from src.serving.semantic_layer.stage_clock import (
     coerce_dt,
@@ -48,6 +50,27 @@
 }
 _SEVERITY_RANK = {"high": 0, "medium": 1, "low": 2}
 
+_DEFAULT_INBOX_SCAN_LIMIT = 20_000
+
+
+def inbox_scan_limit() -> int:
+    """Per-source row cap for the exception-inbox store reads (security
+    pre-audit S-8): the inbox materialises every dead-letter row and dead
+    webhook delivery for the tenant in memory on a worker thread. Same "size
+    safety net" contract as ``journal_scan_limit``/``orders_scan_limit`` —
+    the gather probes with ``cap + 1``, so hitting the cap is reported
+    (``scan_truncated``), never a silent cut. Env-tunable via
+    ``AGENTFLOW_OPS_INBOX_SCAN_LIMIT``.
+    """
+    raw = (os.getenv("AGENTFLOW_OPS_INBOX_SCAN_LIMIT") or "").strip()
+    if not raw:
+        return _DEFAULT_INBOX_SCAN_LIMIT
+    try:
+        value = int(raw)
+    except ValueError:
+        return _DEFAULT_INBOX_SCAN_LIMIT
+    return value if value > 0 else _DEFAULT_INBOX_SCAN_LIMIT
+
 
 class StuckOrderItem(BaseModel):
     order_id: str
@@ -71,6 +94,9 @@ class StuckOrdersResponse(BaseModel):
     items: list[StuckOrderItem]
     summary: StuckOrdersSummary
     pagination: dict[str, int]
+    # True when the open-orders read hit its scan cap (S-8): the worklist,
+    # summary counts, and total then cover the scanned window only.
+    scan_truncated: bool = False
 
 
 def _resolve_tenant_id(request: Request) -> str | None:
@@ -144,7 +170,14 @@ def _build_stuck_orders_payload(
     stage_budgets = (getattr(order_def, "stages", None) or []) if order_def else []
     ladder = ladder_stage_names(stage_budgets)
 
-    order_rows = engine.fetch_orders_by_status(ladder, tenant_id=tenant_id)
+    # cap+1 probe: hitting the cap is reported as `scan_truncated`, never a
+    # silent cut of the worklist (S-8). Truncation is deterministic — the
+    # engine read orders by primary key.
+    scan_cap = orders_scan_limit()
+    order_rows = engine.fetch_orders_by_status(ladder, tenant_id=tenant_id, limit=scan_cap + 1)
+    scan_truncated = len(order_rows) > scan_cap
+    if scan_truncated:
+        order_rows = order_rows[:scan_cap]
     stage_rows = engine.fetch_pipeline_events(
         tenant_id=tenant_id,
         topic="orders.status",
@@ -217,6 +250,7 @@ def _sort_key(item: dict[str, Any]) -> float:
             "total": total,
             "pages": math.ceil(total / page_size) if total else 0,
         },
+        "scan_truncated": scan_truncated,
     }
 
 
@@ -265,12 +299,16 @@ class ExceptionItem(BaseModel):
 class ExceptionsListResponse(BaseModel):
     items: list[ExceptionItem]
     pagination: dict[str, int]
+    # True when a source read hit its scan cap (S-8): the inbox and its
+    # counts then cover the scanned window only.
+    scan_truncated: bool = False
 
 
 class ExceptionsStatsResponse(BaseModel):
     by_source: dict[str, dict[str, int]] = Field(default_factory=dict)
     last_24h: int
     manual_resolutions: int
+    scan_truncated: bool = False
 
 
 class TriageActionRequest(BaseModel):
@@ -382,33 +420,41 @@ def _reconciliation_finding_to_item(
     }
 
 
-def _gather_exception_items(request: Request) -> tuple[list[dict[str, Any]], str]:
+def _gather_exception_items(request: Request) -> tuple[list[dict[str, Any]], str, bool]:
     """Run R1/R2, upsert/auto-resolve the overlay, and assemble every current
     item across all three sources (§4.1), unfiltered — the list and stats
     endpoints both start from this same picture, so counts never drift
-    between them within one request."""
+    between them within one request. The third element reports whether any
+    source read hit its scan cap (S-8)."""
     store = get_control_plane_store(request.app)
     engine = request.app.state.query_engine
     tenant_id = _tenant_id(request)
     catalog = request.app.state.catalog
     order_def = catalog.entities.get("order")
     stage_budgets = (getattr(order_def, "stages", None) or []) if order_def else []
     now = datetime.now(UTC)
+    scan_cap = inbox_scan_limit()
 
     # Source 2: webhook dead deliveries — overlay-backed (§4.1 #2).
-    dead_deliveries = store.list_dead_webhook_deliveries(tenant_id)
+    dead_deliveries = store.list_dead_webhook_deliveries(tenant_id, limit=scan_cap + 1)
+    webhook_truncated = len(dead_deliveries) > scan_cap
+    if webhook_truncated:
+        dead_deliveries = dead_deliveries[:scan_cap]
     webhook_seen_ids = [f"wh:{row['webhook_id']}:{row['event_id']}" for row in dead_deliveries]
     for row, item_id in zip(dead_deliveries, webhook_seen_ids, strict=True):
         seen_at = coerce_dt(row.get("updated_at")) or now
         store.upsert_triage_finding(
             item_id=item_id, tenant_id=tenant_id, source="webhook_delivery", seen_at=seen_at
         )
-    store.auto_resolve_missing_triage_findings(
-        tenant_id=tenant_id,
-        source="webhook_delivery",
-        seen_item_ids=webhook_seen_ids,
-        resolved_at=now,
-    )
+    if not webhook_truncated:
+        # A truncated scan cannot prove absence: auto-resolving against an
+        # incomplete seen-set would mark still-dead deliveries resolved.
+        store.auto_resolve_missing_triage_findings(
+            tenant_id=tenant_id,
+            source="webhook_delivery",
+            seen_item_ids=webhook_seen_ids,
+            resolved_at=now,
+        )
 
     # Source 3: reconciliation findings — overlay-backed (§4.1 #3).
     findings = [
@@ -434,9 +480,12 @@ def _gather_exception_items(request: Request) -> tuple[list[dict[str, Any]], str
         state.item_id: state for state in store.list_triage_states(tenant_id=tenant_id)
     }
 
-    items: list[dict[str, Any]] = [
-        _deadletter_row_to_item(row) for row in store.list_dead_letter_events_for_inbox(tenant_id)
-    ]
+    deadletter_rows = store.list_dead_letter_events_for_inbox(tenant_id, limit=scan_cap + 1)
+    deadletter_truncated = len(deadletter_rows) > scan_cap
+    if deadletter_truncated:
+        deadletter_rows = deadletter_rows[:scan_cap]
+
+    items: list[dict[str, Any]] = [_deadletter_row_to_item(row) for row in deadletter_rows]
     items.extend(
         _webhook_delivery_row_to_item(row, overlay_states.get(item_id), now)
         for row, item_id in zip(dead_deliveries, webhook_seen_ids, strict=True)
@@ -445,7 +494,7 @@ def _gather_exception_items(request: Request) -> tuple[list[dict[str, Any]], str
         _reconciliation_finding_to_item(finding, overlay_states.get(item_id))
         for finding, item_id in zip(findings, reconciliation_seen_ids, strict=True)
     )
-    return items, tenant_id
+    return items, tenant_id, webhook_truncated or deadletter_truncated
 
 
 def _build_exceptions_list_payload(
@@ -455,7 +504,7 @@ def _build_exceptions_list_payload(
     page: int,
     page_size: int,
 ) -> dict[str, Any]:
-    items, _tenant = _gather_exception_items(request)
+    items, _tenant, scan_truncated = _gather_exception_items(request)
 
     if source is not None:
         items = [item for item in items if item["source"] == source]
@@ -484,11 +533,12 @@ def _build_exceptions_list_payload(
             "total": total,
             "pages": math.ceil(total / page_size) if total else 0,
         },
+        "scan_truncated": scan_truncated,
     }
 
 
 def _build_exceptions_stats_payload(request: Request) -> dict[str, Any]:
-    items, tenant_id = _gather_exception_items(request)
+    items, tenant_id, scan_truncated = _gather_exception_items(request)
     store = get_control_plane_store(request.app)
     now = datetime.now(UTC)
 
@@ -506,6 +556,7 @@ def _build_exceptions_stats_payload(request: Request) -> dict[str, Any]:
         "by_source": by_source,
         "last_24h": last_24h,
         "manual_resolutions": manual_resolutions,
+        "scan_truncated": scan_truncated,
     }
 
 

src/serving/control_plane/embedded.py

@@ -1140,27 +1140,22 @@ def get_dead_letter_stats(self, tenant_id: str) -> dict:
             ],
         }
 
-    def list_dead_letter_events_for_inbox(self, tenant_id: str) -> list[dict]:
+    def list_dead_letter_events_for_inbox(
+        self, tenant_id: str, *, limit: int | None = None
+    ) -> list[dict]:
         cursor = self._conn.cursor()
         try:
             ensure_dead_letter_table(cursor)
-            rows = cursor.execute(
-                """
-                SELECT
-                    event_id,
-                    event_type,
-                    failure_reason,
-                    failure_detail,
-                    received_at,
-                    retry_count,
-                    last_retried_at,
-                    status
-                FROM dead_letter_events
-                WHERE COALESCE(tenant_id, 'default') = ?
-                ORDER BY received_at DESC
-                """,
-                [tenant_id],
-            ).fetchall()
+            select = (
+                "SELECT event_id, event_type, failure_reason, failure_detail, "
+                "received_at, retry_count, last_retried_at, status "
+                "FROM dead_letter_events "
+                "WHERE COALESCE(tenant_id, 'default') = ? "
+                "ORDER BY received_at DESC"
+            )
+            # suffix is empty or "LIMIT <int>" — never caller-shaped text
+            suffix = f" LIMIT {int(limit)}" if limit is not None else ""
+            rows = cursor.execute(select + suffix, [tenant_id]).fetchall()
         finally:
             cursor.close()
         return [
@@ -1381,20 +1376,23 @@ def count_triage_manual_actions(self, tenant_id: str) -> int:
 
     # --- webhook dead deliveries for the exception inbox ----------------------
 
-    def list_dead_webhook_deliveries(self, tenant_id: str | None = None) -> list[dict]:
+    def list_dead_webhook_deliveries(
+        self, tenant_id: str | None = None, *, limit: int | None = None
+    ) -> list[dict]:
         conn = self._conn
         ensure_webhook_delivery_queue_table(conn)
         select = (
             "SELECT webhook_id, event_id, tenant, event_type, body, attempts, "
             "last_status_code, last_error, created_at, updated_at "
             "FROM webhook_delivery_queue WHERE status = 'dead'"
         )
+        suffix = f" LIMIT {int(limit)}" if limit is not None else ""
         if tenant_id is not None:
             rows = conn.execute(
-                select + " AND tenant = ? ORDER BY updated_at DESC", [tenant_id]
+                select + " AND tenant = ? ORDER BY updated_at DESC" + suffix, [tenant_id]
             ).fetchall()
         else:
-            rows = conn.execute(select + " ORDER BY updated_at DESC").fetchall()
+            rows = conn.execute(select + " ORDER BY updated_at DESC" + suffix).fetchall()
         return [
             {
                 "webhook_id": row[0],

src/serving/control_plane/postgres.py

@@ -1224,18 +1224,20 @@ def get_dead_letter_stats(self, tenant_id: str) -> dict:
             ],
         }
 
-    def list_dead_letter_events_for_inbox(self, tenant_id: str) -> list[dict]:
+    def list_dead_letter_events_for_inbox(
+        self, tenant_id: str, *, limit: int | None = None
+    ) -> list[dict]:
+        select = (
+            "SELECT event_id, event_type, failure_reason, failure_detail, "
+            "received_at, retry_count, last_retried_at, status "
+            "FROM dead_letter_events "
+            "WHERE COALESCE(tenant_id, 'default') = %s "
+            "ORDER BY received_at DESC"
+        )
+        # suffix is empty or "LIMIT <int>" — never caller-shaped text
+        suffix = f" LIMIT {int(limit)}" if limit is not None else ""
         with self._connect() as conn:
-            rows = conn.execute(
-                """
-                SELECT event_id, event_type, failure_reason, failure_detail,
-                       received_at, retry_count, last_retried_at, status
-                FROM dead_letter_events
-                WHERE COALESCE(tenant_id, 'default') = %s
-                ORDER BY received_at DESC
-                """,
-                (tenant_id,),
-            ).fetchall()
+            rows = conn.execute(select + suffix, (tenant_id,)).fetchall()
         return [
             {
                 "event_id": row[0],
@@ -1430,19 +1432,22 @@ def count_triage_manual_actions(self, tenant_id: str) -> int:
 
     # --- webhook dead deliveries for the exception inbox ----------------------
 
-    def list_dead_webhook_deliveries(self, tenant_id: str | None = None) -> list[dict]:
+    def list_dead_webhook_deliveries(
+        self, tenant_id: str | None = None, *, limit: int | None = None
+    ) -> list[dict]:
         select = (
             "SELECT webhook_id, event_id, tenant, event_type, body, attempts, "
             "last_status_code, last_error, created_at, updated_at "
             "FROM webhook_delivery_queue WHERE status = 'dead'"
         )
+        suffix = f" LIMIT {int(limit)}" if limit is not None else ""
         with self._connect() as conn:
             if tenant_id is not None:
                 rows = conn.execute(
-                    select + " AND tenant = %s ORDER BY updated_at DESC", (tenant_id,)
+                    select + " AND tenant = %s ORDER BY updated_at DESC" + suffix, (tenant_id,)
                 ).fetchall()
             else:
-                rows = conn.execute(select + " ORDER BY updated_at DESC").fetchall()
+                rows = conn.execute(select + " ORDER BY updated_at DESC" + suffix).fetchall()
         return [
             {
                 "webhook_id": row[0],

src/serving/control_plane/store.py

@@ -455,12 +455,16 @@ def get_dead_letter_stats(self, tenant_id: str) -> dict:
         for one tenant's active (``failed``) dead-letter events."""
 
     @abstractmethod
-    def list_dead_letter_events_for_inbox(self, tenant_id: str) -> list[dict]:
+    def list_dead_letter_events_for_inbox(
+        self, tenant_id: str, *, limit: int | None = None
+    ) -> list[dict]:
         """Every dead-letter row for one tenant, any status, newest first —
         the exception inbox's native source (§4.1 #1). Unlike
         ``list_dead_letter_events`` (the public ``/v1/deadletter`` route:
         ``status='failed'`` only, paginated), the inbox aggregates and
-        paginates across three heterogeneous sources itself."""
+        paginates across three heterogeneous sources itself. ``limit`` bounds
+        the read to the newest N rows (S-8) — the inbox probes with
+        ``cap + 1`` to detect truncation instead of cutting silently."""
 
     @abstractmethod
     def list_stuck_replay_dead_letter_events(
@@ -535,10 +539,13 @@ def count_triage_manual_actions(self, tenant_id: str) -> int:
     # --- webhook dead deliveries for the exception inbox ----------------------
 
     @abstractmethod
-    def list_dead_webhook_deliveries(self, tenant_id: str | None = None) -> list[dict]:
+    def list_dead_webhook_deliveries(
+        self, tenant_id: str | None = None, *, limit: int | None = None
+    ) -> list[dict]:
         """Every ``webhook_delivery_queue`` row parked ``dead``, optionally
         scoped to one tenant — the exception inbox's overlay source #2
-        (§4.1)."""
+        (§4.1). ``limit`` bounds the read to the newest N rows (S-8), same
+        ``cap + 1`` probe contract as the dead-letter inbox read."""
 
     # --- API usage accounting (per-tenant/per-key request counters) ----------