From 460875be78429acc7c3127001394c808b7c5c6c8 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 20 Jul 2026 17:32:28 +0300 Subject: [PATCH] fix(ops): bound the stuck-orders and exception-inbox reads, report truncation (S-8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetch_orders_by_status now requires a row cap (SQL LIMIT, deterministic ORDER BY primary key) and the exception inbox's two store reads (dead-letter rows, dead webhook deliveries) accept one — before, a large tenant's worklist read materialised its entire open-orders set and every dead row in memory on a worker thread (security pre-audit S-8). Per product decision the bound is honest pagination, not a silent cut: every capped caller probes with cap+1 and reports 'scan_truncated' in the response (stuck-orders, exceptions list, exceptions stats) when the window overflowed. Caps are env-tunable size safety nets in the journal_scan_limit mould: AGENTFLOW_OPS_ORDERS_SCAN_LIMIT and AGENTFLOW_OPS_INBOX_SCAN_LIMIT, both defaulting to 20000 — demo scale never gets near them. Two correctness guards that fall out of truncation: - a truncated webhook scan skips auto_resolve_missing_triage_findings — an incomplete seen-set cannot prove absence, and a wrongly resolved row whose updated_at never advances could not be reopened by upsert; - R1 journal-vs-store keeps its false-negative-only failure mode: an order outside the bounded window falls into the existing 'missing from the store - tolerated' branch, never a phantom finding. OpenAPI snapshot regenerated (three additive optional fields). Tests: engine LIMIT + determinism unit tests, newest-N store tests for both reads, and integration tests pinning the truncation flag on all three surfaces plus the auto-resolve guard. Co-Authored-By: Claude Fable 5 --- docs/openapi.json | 15 ++++ src/serving/api/routers/ops.py | 83 +++++++++++++++---- src/serving/control_plane/embedded.py | 40 +++++---- src/serving/control_plane/postgres.py | 33 ++++---- src/serving/control_plane/store.py | 15 +++- .../semantic_layer/query/entity_queries.py | 11 ++- src/serving/semantic_layer/reconciliation.py | 29 ++++++- tests/integration/test_exceptions_inbox.py | 70 ++++++++++++++++ tests/integration/test_stuck_orders.py | 26 ++++++ tests/unit/test_control_plane_store.py | 38 ++++++++- tests/unit/test_query_package_logic.py | 21 +++-- 11 files changed, 318 insertions(+), 63 deletions(-) diff --git a/docs/openapi.json b/docs/openapi.json index 018caa1c..7e99787b 100644 --- a/docs/openapi.json +++ b/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", diff --git a/src/serving/api/routers/ops.py b/src/serving/api/routers/ops.py index 79e97857..e33e55d5 100644 --- a/src/serving/api/routers/ops.py +++ b/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,11 +420,12 @@ 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) @@ -394,21 +433,28 @@ def _gather_exception_items(request: Request) -> tuple[list[dict[str, Any]], str 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, } diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index 86a66588..a1c4a1fb 100644 --- a/src/serving/control_plane/embedded.py +++ b/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 " — 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,7 +1376,9 @@ 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 = ( @@ -1389,12 +1386,13 @@ def list_dead_webhook_deliveries(self, tenant_id: str | None = None) -> list[dic "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], diff --git a/src/serving/control_plane/postgres.py b/src/serving/control_plane/postgres.py index c6b17eb6..42eb0c08 100644 --- a/src/serving/control_plane/postgres.py +++ b/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 " — 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], diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index 57bc4f08..2ec4ab20 100644 --- a/src/serving/control_plane/store.py +++ b/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) ---------- diff --git a/src/serving/semantic_layer/query/entity_queries.py b/src/serving/semantic_layer/query/entity_queries.py index 4a08acce..12846331 100644 --- a/src/serving/semantic_layer/query/entity_queries.py +++ b/src/serving/semantic_layer/query/entity_queries.py @@ -132,6 +132,8 @@ def fetch_orders_by_status( self: QueryExecutionHost, statuses: list[str], tenant_id: str | None = None, + *, + limit: int, ) -> list[dict]: """Bulk read for the stuck-orders worklist (ops-surfaces-spec.md §3.2). @@ -141,6 +143,12 @@ def fetch_orders_by_status( order's latest ``orders.status`` row) happens in the caller via ``fetch_pipeline_events(topic="orders.status")``, the same port method the Order 360 timeline already uses. + + ``limit`` is required (security pre-audit S-8): without it this read + materialises a large tenant's entire open-orders set on a worker + thread. Truncation is deterministic (ORDER BY primary key), and the + callers probe with ``cap + 1`` to *detect* it rather than cut + silently. """ entity_def = self.catalog.entities.get("order") if entity_def is None or not statuses: @@ -162,7 +170,8 @@ def render(value: str) -> str: # caller-supplied catalog ladder, never a literal here f"SELECT * FROM {table_name} " # nosec B608 f"WHERE status IN ({status_placeholders}) " - f"ORDER BY {self._quote_identifier(entity_def.primary_key)}" + f"ORDER BY {self._quote_identifier(entity_def.primary_key)} " + f"LIMIT {int(limit)}" ) try: rows = ( diff --git a/src/serving/semantic_layer/reconciliation.py b/src/serving/semantic_layer/reconciliation.py index 1721f9c3..814b019b 100644 --- a/src/serving/semantic_layer/reconciliation.py +++ b/src/serving/semantic_layer/reconciliation.py @@ -23,6 +23,26 @@ _STATUS_EVENT_PREFIX = "order.status." _DEFAULT_JOURNAL_SCAN_LIMIT = 20_000 +_DEFAULT_ORDERS_SCAN_LIMIT = 20_000 + + +def orders_scan_limit() -> int: + """Row cap for the ops-surfaces open-orders read (security pre-audit S-8): + ``fetch_orders_by_status`` with no bound materialises a large tenant's + entire open-orders set on a worker thread. Same "size safety net" contract + as ``journal_scan_limit`` — at demo scale the read never gets near the + cap, and the stuck-orders route probes with ``cap + 1`` so hitting it is + *reported* (``scan_truncated``), never a silent cut. Env-tunable via + ``AGENTFLOW_OPS_ORDERS_SCAN_LIMIT``. + """ + raw = (os.getenv("AGENTFLOW_OPS_ORDERS_SCAN_LIMIT") or "").strip() + if not raw: + return _DEFAULT_ORDERS_SCAN_LIMIT + try: + value = int(raw) + except ValueError: + return _DEFAULT_ORDERS_SCAN_LIMIT + return value if value > 0 else _DEFAULT_ORDERS_SCAN_LIMIT def journal_scan_limit() -> int: @@ -110,7 +130,14 @@ def check_journal_vs_store( if not latest_by_entity: return [] - order_rows = engine.fetch_orders_by_status([*ladder, *terminal_names], tenant_id=tenant_id) + # Bounded like the journal scan above (S-8). Truncation here can only + # *miss* findings, never invent them: an order absent from the bounded + # window falls into the "missing from the store — tolerated, not flagged" + # branch below, the same false-negative-only failure mode the bounded + # journal scan already accepts. + order_rows = engine.fetch_orders_by_status( + [*ladder, *terminal_names], tenant_id=tenant_id, limit=orders_scan_limit() + ) store_status_by_id = {str(row.get("order_id")): row.get("status") for row in order_rows} findings: list[ReconciliationFinding] = [] diff --git a/tests/integration/test_exceptions_inbox.py b/tests/integration/test_exceptions_inbox.py index 1c59198a..5a31739e 100644 --- a/tests/integration/test_exceptions_inbox.py +++ b/tests/integration/test_exceptions_inbox.py @@ -310,6 +310,76 @@ def test_acknowledge_then_resolve_lifecycle(authed_client: TestClient, auth_head assert stats["manual_resolutions"] == 1 +def test_scan_cap_reports_truncation_in_list_and_stats( + client: TestClient, monkeypatch: pytest.MonkeyPatch +): + # S-8: the demo seed has two dead-letter rows; a cap of 1 truncates the + # inbox's native source, which both surfaces report — never a silent cut. + data = client.get("/v1/ops/exceptions").json() + assert data["scan_truncated"] is False + assert client.get("/v1/ops/exceptions/stats").json()["scan_truncated"] is False + + monkeypatch.setenv("AGENTFLOW_OPS_INBOX_SCAN_LIMIT", "1") + + data = client.get("/v1/ops/exceptions").json() + assert data["scan_truncated"] is True + assert len(data["items"]) == 1 + assert client.get("/v1/ops/exceptions/stats").json()["scan_truncated"] is True + + +def test_truncated_webhook_scan_never_auto_resolves_out_of_window_items( + client: TestClient, monkeypatch: pytest.MonkeyPatch +): + # S-8 guard: a truncated scan cannot prove absence. Without the guard, + # the capped gather would pass an incomplete seen-set to auto-resolve and + # mark the still-dead out-of-window delivery resolved — and since its + # `updated_at` never advances, the later upsert could not reopen it. + from datetime import UTC, datetime, timedelta + + from src.serving.control_plane import get_control_plane_store + from src.serving.control_plane.embedded import ensure_webhook_delivery_queue_table + + store = get_control_plane_store(client.app) + conn = store._conn + ensure_webhook_delivery_queue_table(conn) + now = datetime.now(UTC) + for webhook_id, event_id, updated_at in ( + ("wh-old", "evt-old", now - timedelta(minutes=10)), + ("wh-new", "evt-new", now), + ): + conn.execute( + """ + INSERT INTO webhook_delivery_queue + (webhook_id, event_id, tenant, event_type, body, status, attempts, + last_error, created_at, updated_at) + VALUES (?, ?, 'default', 'order.created', '{}', 'dead', 5, + 'connection refused', ?, ?) + """, + [webhook_id, event_id, updated_at, updated_at], + ) + + # Both enter the overlay while the scan is unbounded. + data = client.get("/v1/ops/exceptions", params={"source": "webhook_delivery"}).json() + assert {item["item_id"] for item in data["items"]} == { + "wh:wh-old:evt-old", + "wh:wh-new:evt-new", + } + + # Capped to the newest row: truncated, and wh-old is out of the window. + monkeypatch.setenv("AGENTFLOW_OPS_INBOX_SCAN_LIMIT", "1") + data = client.get("/v1/ops/exceptions", params={"source": "webhook_delivery"}).json() + assert data["scan_truncated"] is True + assert [item["item_id"] for item in data["items"]] == ["wh:wh-new:evt-new"] + + # Uncapped again: wh-old is still open — the truncated scan resolved + # nothing behind the operator's back. + monkeypatch.delenv("AGENTFLOW_OPS_INBOX_SCAN_LIMIT") + data = client.get("/v1/ops/exceptions", params={"source": "webhook_delivery"}).json() + status_by_id = {item["item_id"]: item["status"] for item in data["items"]} + assert status_by_id["wh:wh-old:evt-old"] == "open" + assert status_by_id["wh:wh-new:evt-new"] == "open" + + def test_auto_resolve_when_the_finding_no_longer_reproduces( authed_client: TestClient, auth_headers ): diff --git a/tests/integration/test_stuck_orders.py b/tests/integration/test_stuck_orders.py index 8c7df632..bc6b924a 100644 --- a/tests/integration/test_stuck_orders.py +++ b/tests/integration/test_stuck_orders.py @@ -113,6 +113,32 @@ def test_pagination_shape(client: TestClient): assert data["pagination"] == {"page": 1, "page_size": 2, "total": 5, "pages": 3} +def test_scan_cap_reports_truncation_instead_of_cutting_silently( + client: TestClient, monkeypatch: pytest.MonkeyPatch +): + # S-8: the open-orders read is bounded, and hitting the bound is visible + # (`scan_truncated`), never a silent cut of the worklist. The window is + # deterministic — the engine read orders by primary key, so with a cap of + # 2 the first two open orders by id are the whole scanned picture. + monkeypatch.setenv("AGENTFLOW_OPS_ORDERS_SCAN_LIMIT", "2") + + response = client.get("/v1/ops/stuck-orders", params={"include_within_sla": "true"}) + + assert response.status_code == 200 + data = response.json() + assert data["scan_truncated"] is True + assert {item["order_id"] for item in data["items"]} == { + "ORD-20260404-1002", + "ORD-20260404-1003", + } + assert data["pagination"]["total"] == 2 + + # Uncapped, the flag reports an untruncated scan. + monkeypatch.delenv("AGENTFLOW_OPS_ORDERS_SCAN_LIMIT") + data = client.get("/v1/ops/stuck-orders").json() + assert data["scan_truncated"] is False + + def test_order_without_stage_rows_reports_fallback_clock(client: TestClient): # I12: an order written outside the stage-row writer degrades honestly # to the created_at fallback instead of pretending to have a journal diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index 5ce2a246..f353823d 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -552,6 +552,7 @@ def _seed_dead_letter( event_id: str, tenant_id: str = "acme", status: str = "failed", + received_at: datetime | None = None, ) -> None: conn.execute( """ @@ -560,7 +561,7 @@ def _seed_dead_letter( failure_detail, received_at, retry_count, last_retried_at, status ) VALUES (?, ?, 'order.created', '{"a": 1}', 'semantic', 'x', ?, 0, NULL, ?) """, - [event_id, tenant_id, datetime.now(UTC), status], + [event_id, tenant_id, received_at or datetime.now(UTC), status], ) @@ -983,6 +984,41 @@ def test_list_dead_letter_events_for_inbox_returns_every_status( assert {row["event_id"] for row in rows} == {"evt-failed", "evt-dismissed"} +def test_list_dead_webhook_deliveries_limit_keeps_the_newest_rows( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + # S-8: the bound must be the newest N (ORDER BY updated_at DESC), so the + # inbox's cap+1 probe sees fresh items first and truncation is honest. + base = datetime.now(UTC) + for index in range(3): + _seed_webhook_dead( + conn, + webhook_id=f"wh-{index}", + event_id=f"evt-{index}", + updated_at=base - timedelta(minutes=3 - index), + ) + + rows = store.list_dead_webhook_deliveries("acme", limit=2) + + assert [row["webhook_id"] for row in rows] == ["wh-2", "wh-1"] + + +def test_list_dead_letter_events_for_inbox_limit_keeps_the_newest_rows( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + base = datetime.now(UTC) + for index in range(3): + _seed_dead_letter( + conn, + event_id=f"evt-{index}", + received_at=base - timedelta(minutes=3 - index), + ) + + rows = outbox_store.list_dead_letter_events_for_inbox("acme", limit=2) + + assert [row["event_id"] for row in rows] == ["evt-2", "evt-1"] + + def test_list_stuck_replay_dead_letter_events_filters_by_age_and_status( outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection ) -> None: diff --git a/tests/unit/test_query_package_logic.py b/tests/unit/test_query_package_logic.py index 9693beaf..790d2367 100644 --- a/tests/unit/test_query_package_logic.py +++ b/tests/unit/test_query_package_logic.py @@ -324,14 +324,14 @@ def test_scan_entity_rows_by_ids_quotes_every_id(host: _Host) -> None: def test_fetch_orders_by_status_empty_ladder_returns_empty(host: _Host) -> None: - assert host.fetch_orders_by_status([]) == [] + assert host.fetch_orders_by_status([], limit=50) == [] host._backend.execute.assert_not_called() def test_fetch_orders_by_status_param_backend_binds_statuses(host: _Host) -> None: host._backend.execute.return_value = [{"order_id": "ORD-1", "status": "pending"}] - rows = host.fetch_orders_by_status(["pending", "confirmed"]) + rows = host.fetch_orders_by_status(["pending", "confirmed"], limit=50) assert rows == [{"order_id": "ORD-1", "status": "pending"}] sql, params = host._backend.execute.call_args.args @@ -339,10 +339,21 @@ def test_fetch_orders_by_status_param_backend_binds_statuses(host: _Host) -> Non assert params == ["pending", "confirmed"] +def test_fetch_orders_by_status_is_bounded_and_deterministic(host: _Host) -> None: + # S-8: the read must carry the caller's row cap, and truncation must be + # deterministic — ORDER BY primary key, then LIMIT. + host._backend.execute.return_value = [] + + host.fetch_orders_by_status(["pending"], limit=101) + + sql = host._backend.execute.call_args.args[0] + assert sql.endswith('ORDER BY "order_id" LIMIT 101') + + def test_fetch_orders_by_status_literal_backend_quotes_statuses(literal_host: _Host) -> None: literal_host._backend.execute.return_value = [] - assert literal_host.fetch_orders_by_status(["pend'ing"]) == [] + assert literal_host.fetch_orders_by_status(["pend'ing"], limit=50) == [] sql = literal_host._backend.execute.call_args.args[0] assert "WHERE status IN ('pend''ing')" in sql assert len(literal_host._backend.execute.call_args.args) == 1 @@ -352,14 +363,14 @@ def test_fetch_orders_by_status_missing_table_maps_to_value_error(host: _Host) - host._backend.execute.side_effect = BackendMissingTableError("no table") with pytest.raises(ValueError, match="not materialized yet"): - host.fetch_orders_by_status(["pending"]) + host.fetch_orders_by_status(["pending"], limit=50) def test_fetch_orders_by_status_execution_error_maps_to_value_error(host: _Host) -> None: host._backend.execute.side_effect = BackendExecutionError("boom") with pytest.raises(ValueError, match="Open-orders lookup failed"): - host.fetch_orders_by_status(["pending"]) + host.fetch_orders_by_status(["pending"], limit=50) # ---------------------------------------------------------------------------