diff --git a/docs/agent-tools/claude-tools.json b/docs/agent-tools/claude-tools.json index 17a5beb0..f0208e11 100644 --- a/docs/agent-tools/claude-tools.json +++ b/docs/agent-tools/claude-tools.json @@ -106,6 +106,22 @@ ] } }, + { + "name": "get_order_timeline_v1_entity_order__order_id__timeline_get", + "description": "Get Order Timeline\n\nOrder 360: order state, stage history, pipeline trail, customer block,\nand linked exceptions \u2014 one composed read (ops-surfaces-spec.md \u00a72).\n\nSame tenant scoping and 404 semantics as GET /entity/order/{order_id}.\nNot cached: this is the \"now\" surface (ADR 0011 constraint 3, spec \u00a71.8).\n\nExample:\n GET /v1/entity/order/ORD-20260404-1001/timeline", + "input_schema": { + "type": "object", + "properties": { + "order_id": { + "type": "string", + "title": "Order Id" + } + }, + "required": [ + "order_id" + ] + } + }, { "name": "get_metric_v1_metrics__metric_name__get", "description": "Get Metric\n\nGet a real-time metric value.\n\nSupported metrics: revenue, order_count, avg_order_value, conversion_rate,\nactive_sessions, error_rate.\n\nWindow options: 5m, 15m, 1h, 6h, 24h.\n\nExample:\n GET /v1/metrics/revenue?window=1h", diff --git a/docs/agent-tools/openai-tools.json b/docs/agent-tools/openai-tools.json index 77d2606c..5ba5ac6e 100644 --- a/docs/agent-tools/openai-tools.json +++ b/docs/agent-tools/openai-tools.json @@ -115,6 +115,25 @@ } } }, + { + "type": "function", + "function": { + "name": "get_order_timeline_v1_entity_order__order_id__timeline_get", + "description": "Get Order Timeline\n\nOrder 360: order state, stage history, pipeline trail, customer block,\nand linked exceptions \u2014 one composed read (ops-surfaces-spec.md \u00a72).\n\nSame tenant scoping and 404 semantics as GET /entity/order/{order_id}.\nNot cached: this is the \"now\" surface (ADR 0011 constraint 3, spec \u00a71.8).\n\nExample:\n GET /v1/entity/order/ORD-20260404-1001/timeline", + "parameters": { + "type": "object", + "properties": { + "order_id": { + "type": "string", + "title": "Order Id" + } + }, + "required": [ + "order_id" + ] + } + } + }, { "type": "function", "function": { diff --git a/docs/openapi.json b/docs/openapi.json index 4a3c047e..76bf67cd 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -161,6 +161,49 @@ } } }, + "/v1/entity/order/{order_id}/timeline": { + "get": { + "tags": [ + "agent" + ], + "summary": "Get Order Timeline", + "description": "Order 360: order state, stage history, pipeline trail, customer block,\nand linked exceptions \u2014 one composed read (ops-surfaces-spec.md \u00a72).\n\nSame tenant scoping and 404 semantics as GET /entity/order/{order_id}.\nNot cached: this is the \"now\" surface (ADR 0011 constraint 3, spec \u00a71.8).\n\nExample:\n GET /v1/entity/order/ORD-20260404-1001/timeline", + "operationId": "get_order_timeline_v1_entity_order__order_id__timeline_get", + "parameters": [ + { + "name": "order_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Order Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/OrderTimelineResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/v1/metrics/{metric_name}": { "get": { "tags": [ @@ -3017,6 +3060,415 @@ ], "title": "NLQueryRequest" }, + "OrderTimelineCustomer": { + "properties": { + "user_id": { + "type": "string", + "title": "User Id" + }, + "total_orders": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Total Orders" + }, + "total_spent": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Total Spent" + }, + "first_order_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "First Order At" + }, + "last_order_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Last Order At" + }, + "preferred_category": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Preferred Category" + } + }, + "type": "object", + "required": [ + "user_id" + ], + "title": "OrderTimelineCustomer" + }, + "OrderTimelineException": { + "properties": { + "event_id": { + "type": "string", + "title": "Event Id" + }, + "failure_reason": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Failure Reason" + }, + "status": { + "type": "string", + "title": "Status" + }, + "occurred_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Occurred At" + }, + "actions": { + "$ref": "#/components/schemas/OrderTimelineExceptionActions" + } + }, + "type": "object", + "required": [ + "event_id", + "status", + "actions" + ], + "title": "OrderTimelineException" + }, + "OrderTimelineExceptionActions": { + "properties": { + "replay": { + "type": "string", + "title": "Replay" + }, + "dismiss": { + "type": "string", + "title": "Dismiss" + } + }, + "type": "object", + "required": [ + "replay", + "dismiss" + ], + "title": "OrderTimelineExceptionActions" + }, + "OrderTimelineOrder": { + "properties": { + "order_id": { + "type": "string", + "title": "Order Id" + }, + "user_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "User Id" + }, + "status": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Status" + }, + "total_amount": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Total Amount" + }, + "currency": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Currency" + }, + "created_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Created At" + } + }, + "type": "object", + "required": [ + "order_id" + ], + "title": "OrderTimelineOrder" + }, + "OrderTimelinePipelineTrailItem": { + "properties": { + "event_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Event Id" + }, + "topic": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Topic" + }, + "event_type": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Event Type" + }, + "latency_ms": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Latency Ms" + }, + "processed_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Processed At" + } + }, + "type": "object", + "title": "OrderTimelinePipelineTrailItem" + }, + "OrderTimelineResponse": { + "properties": { + "order": { + "$ref": "#/components/schemas/OrderTimelineOrder" + }, + "stage": { + "$ref": "#/components/schemas/OrderTimelineStage" + }, + "stage_history": { + "items": { + "$ref": "#/components/schemas/OrderTimelineStageHistoryItem" + }, + "type": "array", + "title": "Stage History" + }, + "pipeline_trail": { + "items": { + "$ref": "#/components/schemas/OrderTimelinePipelineTrailItem" + }, + "type": "array", + "title": "Pipeline Trail" + }, + "customer": { + "anyOf": [ + { + "$ref": "#/components/schemas/OrderTimelineCustomer" + }, + { + "type": "null" + } + ] + }, + "exceptions": { + "items": { + "$ref": "#/components/schemas/OrderTimelineException" + }, + "type": "array", + "title": "Exceptions" + } + }, + "type": "object", + "required": [ + "order", + "stage" + ], + "title": "OrderTimelineResponse" + }, + "OrderTimelineStage": { + "properties": { + "current": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Current" + }, + "entered_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Entered At" + }, + "in_stage_seconds": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "In Stage Seconds" + }, + "sla_minutes": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Sla Minutes" + }, + "breached": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Breached" + }, + "clock": { + "type": "string", + "enum": [ + "journal", + "fallback" + ], + "title": "Clock" + } + }, + "type": "object", + "required": [ + "clock" + ], + "title": "OrderTimelineStage" + }, + "OrderTimelineStageHistoryItem": { + "properties": { + "status": { + "type": "string", + "title": "Status" + }, + "at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "At" + } + }, + "type": "object", + "required": [ + "status" + ], + "title": "OrderTimelineStageHistoryItem" + }, "QueryResponse": { "properties": { "answer": { diff --git a/src/processing/clickhouse_sink.py b/src/processing/clickhouse_sink.py index 67117d6c..14e8f1d9 100644 --- a/src/processing/clickhouse_sink.py +++ b/src/processing/clickhouse_sink.py @@ -82,7 +82,8 @@ def record_pipeline_event( topic: str, tenant_id: str, event_type: str, - latency_ms: int, + latency_ms: int | None, + entity_id: str | None = None, processed_at: datetime | None = None, ) -> None: self._backend.insert_rows( @@ -92,7 +93,7 @@ def record_pipeline_event( "event_id": event_id, "topic": topic, "tenant_id": tenant_id, - "entity_id": None, + "entity_id": entity_id, "event_type": event_type, "latency_ms": latency_ms, "processed_at": processed_at or datetime.now(UTC), diff --git a/src/processing/local_pipeline.py b/src/processing/local_pipeline.py index f3453114..4fa630a0 100644 --- a/src/processing/local_pipeline.py +++ b/src/processing/local_pipeline.py @@ -99,6 +99,7 @@ def _ensure_tables(conn: duckdb.DuckDBPyConnection) -> None: conn.execute( "ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS tenant_id VARCHAR DEFAULT 'default'" ) + conn.execute("ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS entity_id VARCHAR") def _event_tenant(event: dict) -> str: @@ -108,6 +109,25 @@ def _event_tenant(event: dict) -> str: return str(tenant) if tenant else "default" +# ops-surfaces-spec.md §1.3: the entity_id axis of the pipeline_events journal +# is what Order 360 (and lineage) key off of. NULL when the id isn't +# derivable from the payload — never synthesized. +_ENTITY_ID_FIELD_BY_PREFIX = ( + ("order.", "order_id"), + ("user.", "user_id"), + ("product.", "product_id"), + ("session.", "session_id"), +) + + +def _derive_entity_id(event: dict, event_type: str) -> str | None: + for prefix, field_name in _ENTITY_ID_FIELD_BY_PREFIX: + if event_type.startswith(prefix): + value = event.get(field_name) + return str(value) if value is not None else None + return None + + def _process_event( conn: duckdb.DuckDBPyConnection, event: dict, @@ -118,6 +138,7 @@ def _process_event( event_type = event.get("event_type", "") event_id = event.get("event_id", "unknown") tenant_id = _event_tenant(event) + entity_id = _derive_entity_id(event, event_type) conn.execute("BEGIN") try: @@ -127,11 +148,11 @@ def _process_event( conn.execute( """ INSERT INTO pipeline_events ( - event_id, topic, tenant_id, event_type, latency_ms, processed_at + event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at ) - VALUES (?, 'events.deadletter', ?, ?, 0, ?) + VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?) """, - [event_id, tenant_id, event_type, datetime.now(UTC)], + [event_id, tenant_id, entity_id, event_type, datetime.now(UTC)], ) if iceberg_sink is not None: iceberg_sink.write_batch( @@ -157,6 +178,7 @@ def _process_event( event_id=str(event_id), topic="events.deadletter", tenant_id=tenant_id, + entity_id=entity_id, event_type=event_type, latency_ms=0, ) @@ -169,11 +191,11 @@ def _process_event( conn.execute( """ INSERT INTO pipeline_events ( - event_id, topic, tenant_id, event_type, latency_ms, processed_at + event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at ) - VALUES (?, 'events.deadletter', ?, ?, 0, ?) + VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?) """, - [event_id, tenant_id, event_type, datetime.now(UTC)], + [event_id, tenant_id, entity_id, event_type, datetime.now(UTC)], ) if iceberg_sink is not None: iceberg_sink.write_batch( @@ -195,6 +217,7 @@ def _process_event( event_id=str(event_id), topic="events.deadletter", tenant_id=tenant_id, + entity_id=entity_id, event_type=event_type, latency_ms=0, ) @@ -204,6 +227,7 @@ def _process_event( if event_type.startswith("order."): event = enrich_order(event) _upsert_order(conn, event) + _record_order_status(conn, event, event_id, tenant_id) if iceberg_sink is not None: iceberg_sink.write_batch("orders", [event]) elif event_type in ("click", "page_view", "add_to_cart"): @@ -233,16 +257,25 @@ def _process_event( conn.execute( """ INSERT INTO pipeline_events ( - event_id, topic, tenant_id, event_type, latency_ms, processed_at + event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at ) - VALUES (?, 'events.validated', ?, ?, ?, ?) + VALUES (?, 'events.validated', ?, ?, ?, ?, ?) """, - [event_id, tenant_id, event_type, latency_ms, datetime.now(UTC)], + [event_id, tenant_id, entity_id, event_type, latency_ms, datetime.now(UTC)], ) conn.execute("COMMIT") if clickhouse_sink is not None: if event_type.startswith("order."): clickhouse_sink.upsert_order(event) + clickhouse_sink.record_pipeline_event( + event_id=f"{event_id}-status", + topic="orders.status", + tenant_id=tenant_id, + entity_id=str(event["order_id"]), + event_type=f"order.status.{event['status']}", + latency_ms=None, + processed_at=datetime.now(UTC), + ) elif event_type in ("click", "page_view", "add_to_cart"): clickhouse_sink.upsert_session(event) elif event_type.startswith("product."): @@ -251,6 +284,7 @@ def _process_event( event_id=str(event_id), topic="events.validated", tenant_id=tenant_id, + entity_id=entity_id, event_type=event_type, latency_ms=latency_ms, ) @@ -299,6 +333,30 @@ def _upsert_order(conn: duckdb.DuckDBPyConnection, event: dict) -> None: ) +def _record_order_status( + conn: duckdb.DuckDBPyConnection, event: dict, event_id: str, tenant_id: str +) -> None: + """Stage-entry journal row (ops-surfaces-spec.md §1.2) — the stage clock + for Order 360 / stuck-orders. ``topic='orders.status'`` is deliberately + disjoint from the ingestion vocabulary (``order.created``, ...) so it + never gets picked up by scans that filter on ingestion event types.""" + conn.execute( + """ + INSERT INTO pipeline_events ( + event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at + ) + VALUES (?, 'orders.status', ?, ?, ?, NULL, ?) + """, + [ + f"{event_id}-status", + tenant_id, + str(event["order_id"]), + f"order.status.{event['status']}", + datetime.now(UTC), + ], + ) + + def _upsert_product(conn: duckdb.DuckDBPyConnection, event: dict) -> None: conn.execute( """ diff --git a/src/serving/api/routers/agent_query.py b/src/serving/api/routers/agent_query.py index af2242ad..5d2ce341 100644 --- a/src/serving/api/routers/agent_query.py +++ b/src/serving/api/routers/agent_query.py @@ -24,6 +24,7 @@ resolve_request_version, ) from src.serving.cache import ENTITY_TTL_SECONDS, QueryCache, cache_entity_key +from src.serving.control_plane import get_control_plane_store logger = structlog.get_logger() tracer = trace.get_tracer("agentflow.api") @@ -113,6 +114,29 @@ def _as_of_iso_text(as_of: datetime | None) -> str | None: return as_of.replace(microsecond=0).isoformat().replace("+00:00", "Z") +def _coerce_dt(value: object) -> datetime | None: + # Naive DuckDB timestamps are local wall-clock (DuckDB's NOW()), not UTC — + # same local_tz convention as EntityQueryMixin.get_entity's _last_updated. + local_tz = datetime.now().astimezone().tzinfo or UTC + if isinstance(value, datetime): + return ( + value.astimezone(UTC) + if value.tzinfo is not None + else value.replace(tzinfo=local_tz).astimezone(UTC) + ) + if isinstance(value, str): + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + return ( + parsed.astimezone(UTC) + if parsed.tzinfo is not None + else parsed.replace(tzinfo=local_tz).astimezone(UTC) + ) + return None + + def _transform_payload_for_requested_version( req: Request, payload: dict[str, Any] ) -> dict[str, Any]: @@ -245,6 +269,225 @@ class MetricResponse(BaseModel): meta: dict = Field(default_factory=dict) +# ── Order 360 timeline models (ops-surfaces-spec.md §2.2) ──────── +# The customer block is a fixed field allow-list — the users_enriched +# columns, PII-free by construction (spec invariant I3). No literal stage +# name or SLA budget appears below (invariant I2): sla_minutes/breached +# come only from the catalog's optional `stages` field, null until that +# field exists (D3, ops-surfaces-spec.md §1.5). + + +class OrderTimelineOrder(BaseModel): + order_id: str + user_id: str | None = None + status: str | None = None + total_amount: float | None = None + currency: str | None = None + created_at: datetime | None = None + + +class OrderTimelineStage(BaseModel): + current: str | None = None + entered_at: datetime | None = None + in_stage_seconds: float | None = None + sla_minutes: int | None = None + breached: bool | None = None + clock: Literal["journal", "fallback"] + + +class OrderTimelineStageHistoryItem(BaseModel): + status: str + at: datetime | None = None + + +class OrderTimelinePipelineTrailItem(BaseModel): + event_id: str | None = None + topic: str | None = None + event_type: str | None = None + latency_ms: float | None = None + processed_at: datetime | None = None + + +class OrderTimelineCustomer(BaseModel): + user_id: str + total_orders: int | None = None + total_spent: float | None = None + first_order_at: datetime | None = None + last_order_at: datetime | None = None + preferred_category: str | None = None + + +class OrderTimelineExceptionActions(BaseModel): + replay: str + dismiss: str + + +class OrderTimelineException(BaseModel): + event_id: str + failure_reason: str | None = None + status: str + occurred_at: datetime | None = None + actions: OrderTimelineExceptionActions + + +class OrderTimelineResponse(BaseModel): + order: OrderTimelineOrder + stage: OrderTimelineStage + stage_history: list[OrderTimelineStageHistoryItem] = Field(default_factory=list) + pipeline_trail: list[OrderTimelinePipelineTrailItem] = Field(default_factory=list) + customer: OrderTimelineCustomer | None = None + exceptions: list[OrderTimelineException] = Field(default_factory=list) + + +_ORDER_TIMELINE_ORDER_FIELDS = ( + "order_id", + "user_id", + "status", + "total_amount", + "currency", + "created_at", +) +_ORDER_TIMELINE_CUSTOMER_FIELDS = ( + "user_id", + "total_orders", + "total_spent", + "first_order_at", + "last_order_at", + "preferred_category", +) + + +def _build_order_timeline(request: Request, order_id: str) -> dict[str, Any] | None: + """Sync composition for GET /entity/order/{order_id}/timeline. + + Runs on a worker thread (the route offloads it, matching lineage.py / + deadletter.py). Composes exactly the two ops-layer ports per ADR 0011: + QueryEngine for the order row, the journal, and the customer projection; + ControlPlaneStore for dead-letter exception detail. No raw connection, no + vault DSN (invariant I1). + """ + engine = request.app.state.query_engine + tenant_id = _resolve_tenant_id(request) + store_tenant_id = tenant_id or "default" + + order_row = engine.get_entity("order", order_id, tenant_id=tenant_id) + if order_row is None: + return None + order_row = dict(order_row) + + journal_rows = engine.fetch_pipeline_events( + tenant_id=tenant_id, entity_id=order_id, newest_first=False + ) + stage_rows = [row for row in journal_rows if row.get("topic") == "orders.status"] + trail_rows = [row for row in journal_rows if row.get("topic") != "orders.status"] + + stage_history = [ + { + "status": str(row["event_type"]).removeprefix("order.status."), + "at": row.get("processed_at"), + } + for row in stage_rows + if row.get("event_type") + ] + + current_status = order_row.get("status") + catalog = request.app.state.catalog + order_def = catalog.entities.get("order") + stage_budgets = (getattr(order_def, "stages", None) or []) if order_def else [] + budget = next( + ( + entry + for entry in stage_budgets + if isinstance(entry, dict) and entry.get("name") == current_status + ), + None, + ) + + entered_at = None + clock = "fallback" + target_event_type = f"order.status.{current_status}" + for row in reversed(stage_rows): + if row.get("event_type") == target_event_type: + entered_at = _coerce_dt(row.get("processed_at")) + clock = "journal" + break + if entered_at is None: + entered_at = _coerce_dt(order_row.get("created_at")) + + in_stage_seconds = ( + (datetime.now(UTC) - entered_at).total_seconds() if entered_at is not None else None + ) + sla_minutes = budget.get("sla_minutes") if budget else None + is_terminal = bool(budget.get("terminal")) if budget else False + breached = ( + None + if is_terminal or sla_minutes is None or in_stage_seconds is None + else in_stage_seconds > sla_minutes * 60 + ) + + customer = None + user_id = order_row.get("user_id") + if user_id: + try: + user_row = engine.get_entity("user", str(user_id), tenant_id=tenant_id) + except ValueError: + # users_enriched not materialized for this tenant/profile — the + # customer block is "null when absent" (spec §2.1), not a 503 for + # the whole timeline; the order/stage/trail data is still good. + user_row = None + if user_row is not None: + customer = {field: user_row.get(field) for field in _ORDER_TIMELINE_CUSTOMER_FIELDS} + + store = get_control_plane_store(request.app) + exceptions = [] + for row in trail_rows: + if row.get("topic") != "events.deadletter": + continue + event_id = row.get("event_id") + if not event_id: + continue + detail = store.get_dead_letter_event(str(event_id), store_tenant_id) + if detail is None: + continue + exceptions.append( + { + "event_id": detail["event_id"], + "failure_reason": detail.get("failure_reason"), + "status": detail["status"], + "occurred_at": detail.get("received_at"), + "actions": { + "replay": f"/v1/deadletter/{detail['event_id']}/replay", + "dismiss": f"/v1/deadletter/{detail['event_id']}/dismiss", + }, + } + ) + + return { + "order": {field: order_row.get(field) for field in _ORDER_TIMELINE_ORDER_FIELDS}, + "stage": { + "current": current_status, + "entered_at": entered_at, + "in_stage_seconds": in_stage_seconds, + "sla_minutes": sla_minutes, + "breached": breached, + "clock": clock, + }, + "stage_history": stage_history, + "pipeline_trail": [ + { + "event_id": row.get("event_id"), + "topic": row.get("topic"), + "event_type": row.get("event_type"), + "latency_ms": row.get("latency_ms"), + "processed_at": row.get("processed_at"), + } + for row in trail_rows + ], + "customer": customer, + "exceptions": exceptions, + } + + # ── Endpoints ─────────────────────────────────────────────────── @@ -449,6 +692,32 @@ async def get_entity( return EntityResponse.model_validate(transformed_payload) +@router.get("/entity/order/{order_id}/timeline", response_model=OrderTimelineResponse) +async def get_order_timeline(order_id: str, req: Request) -> OrderTimelineResponse: + """Order 360: order state, stage history, pipeline trail, customer block, + and linked exceptions — one composed read (ops-surfaces-spec.md §2). + + Same tenant scoping and 404 semantics as GET /entity/order/{order_id}. + Not cached: this is the "now" surface (ADR 0011 constraint 3, spec §1.8). + + Example: + GET /v1/entity/order/ORD-20260404-1001/timeline + """ + catalog = req.app.state.catalog + if "order" not in catalog.entities: + raise HTTPException(status_code=404, detail="Unknown entity type: order") + + try: + payload = await run_in_threadpool(_build_order_timeline, req, order_id) + except ValueError as e: + raise HTTPException(status_code=503, detail=str(e)) from None + + if payload is None: + raise HTTPException(status_code=404, detail=f"order/{order_id} not found") + + return OrderTimelineResponse.model_validate(payload) + + @router.get("/metrics/{metric_name}", response_model=MetricResponse) async def get_metric( metric_name: str, diff --git a/src/serving/api/routers/lineage.py b/src/serving/api/routers/lineage.py index 8972dcd1..c9c82633 100644 --- a/src/serving/api/routers/lineage.py +++ b/src/serving/api/routers/lineage.py @@ -104,6 +104,12 @@ def _fetch_matching_events(request: Request, entity_type: str, entity_id: str) - if tenant_id is not None and "tenant_id" in columns: where_clauses.append("COALESCE(tenant_id, 'default') = ?") params.append(str(tenant_id)) + if "topic" in columns: + # ops-surfaces-spec.md §1.1: orders.status rows are the warehouse + # stage clock, not the ingestion pipeline trail this endpoint + # reconstructs — excluded so they don't hijack source_topic / + # earliest_at (Order 360's timeline shows stage history instead). + where_clauses.append("topic != 'orders.status'") cursor.execute( ( diff --git a/src/serving/backends/clickhouse_backend.py b/src/serving/backends/clickhouse_backend.py index 99befffc..750a05a1 100644 --- a/src/serving/backends/clickhouse_backend.py +++ b/src/serving/backends/clickhouse_backend.py @@ -544,6 +544,58 @@ def ts(delta: timedelta) -> str: expect_json=False, translate=False, ) + # Stage-entry trails (ops-surfaces-spec.md §1.6), mirroring the DuckDB + # seed row-for-row so the Order 360 demo story matches on either + # backend. ORD-20260404-1004's single pending entry at created_at is + # the demo's sole SLA breach once D3 wires the stages: contract block. + self._request( + "\n".join( + [ + # demo seed data uses trusted config and generated timestamps + f"INSERT INTO {self._database}.pipeline_events (event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at) VALUES", # nosec B608 + "('evt-ord-1001-status-pending', 'orders.status', 'default', 'ORD-20260404-1001'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=120))}'),", + "('evt-ord-1001-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1001'," + f" 'order.status.confirmed', NULL, '{ts(timedelta(minutes=100))}'),", + "('evt-ord-1001-status-shipped', 'orders.status', 'default', 'ORD-20260404-1001'," + f" 'order.status.shipped', NULL, '{ts(timedelta(minutes=60))}'),", + "('evt-ord-1001-status-delivered', 'orders.status', 'default', 'ORD-20260404-1001'," + f" 'order.status.delivered', NULL, '{ts(timedelta(minutes=10))}'),", + "('evt-ord-1002-status-pending', 'orders.status', 'default', 'ORD-20260404-1002'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=90))}'),", + "('evt-ord-1002-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1002'," + f" 'order.status.confirmed', NULL, '{ts(timedelta(minutes=80))}'),", + "('evt-ord-1002-status-shipped', 'orders.status', 'default', 'ORD-20260404-1002'," + f" 'order.status.shipped', NULL, '{ts(timedelta(minutes=70))}'),", + "('evt-ord-1003-status-pending', 'orders.status', 'default', 'ORD-20260404-1003'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=60))}'),", + "('evt-ord-1003-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1003'," + f" 'order.status.confirmed', NULL, '{ts(timedelta(minutes=50))}'),", + "('evt-ord-1004-status-pending', 'orders.status', 'default', 'ORD-20260404-1004'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=45))}'),", + "('evt-ord-1005-status-pending', 'orders.status', 'default', 'ORD-20260404-1005'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=30))}'),", + "('evt-ord-1005-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1005'," + f" 'order.status.confirmed', NULL, '{ts(timedelta(minutes=25))}'),", + "('evt-ord-1005-status-shipped', 'orders.status', 'default', 'ORD-20260404-1005'," + f" 'order.status.shipped', NULL, '{ts(timedelta(minutes=15))}'),", + "('evt-ord-1005-status-delivered', 'orders.status', 'default', 'ORD-20260404-1005'," + f" 'order.status.delivered', NULL, '{ts(timedelta(minutes=5))}'),", + "('evt-ord-1006-status-pending', 'orders.status', 'default', 'ORD-20260404-1006'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=20))}'),", + "('evt-ord-1006-status-cancelled', 'orders.status', 'default', 'ORD-20260404-1006'," + f" 'order.status.cancelled', NULL, '{ts(timedelta(minutes=10))}'),", + "('evt-ord-1007-status-pending', 'orders.status', 'default', 'ORD-20260404-1007'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=15))}'),", + "('evt-ord-1007-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1007'," + f" 'order.status.confirmed', NULL, '{ts(timedelta(minutes=8))}'),", + "('evt-ord-1008-status-pending', 'orders.status', 'default', 'ORD-20260404-1008'," + f" 'order.status.pending', NULL, '{ts(timedelta(minutes=5))}')", + ] + ), + expect_json=False, + translate=False, + ) def health(self) -> dict: try: diff --git a/src/serving/backends/duckdb_backend.py b/src/serving/backends/duckdb_backend.py index 1bbaaf8e..f6e10523 100644 --- a/src/serving/backends/duckdb_backend.py +++ b/src/serving/backends/duckdb_backend.py @@ -291,6 +291,56 @@ def initialize_demo_data(self) -> None: 'order.served', 4, NOW() - INTERVAL '1 minute') """) + # Stage-entry trails (ops-surfaces-spec.md §1.6): topic='orders.status', + # one row per stage transition, back-dated between each order's + # created_at and now. ORD-20260404-1004's single pending entry at + # created_at is the demo's sole SLA breach (45min vs a 30min budget, + # once D3 wires the stages: contract block). ORD-20260404-1001 carries + # the full ladder for the Order 360 story (I7). + self._conn.execute(""" + INSERT INTO pipeline_events + (event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at) + VALUES + ('evt-ord-1001-status-pending', 'orders.status', 'default', 'ORD-20260404-1001', + 'order.status.pending', NULL, NOW() - INTERVAL '120 minutes'), + ('evt-ord-1001-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1001', + 'order.status.confirmed', NULL, NOW() - INTERVAL '100 minutes'), + ('evt-ord-1001-status-shipped', 'orders.status', 'default', 'ORD-20260404-1001', + 'order.status.shipped', NULL, NOW() - INTERVAL '60 minutes'), + ('evt-ord-1001-status-delivered', 'orders.status', 'default', 'ORD-20260404-1001', + 'order.status.delivered', NULL, NOW() - INTERVAL '10 minutes'), + ('evt-ord-1002-status-pending', 'orders.status', 'default', 'ORD-20260404-1002', + 'order.status.pending', NULL, NOW() - INTERVAL '90 minutes'), + ('evt-ord-1002-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1002', + 'order.status.confirmed', NULL, NOW() - INTERVAL '80 minutes'), + ('evt-ord-1002-status-shipped', 'orders.status', 'default', 'ORD-20260404-1002', + 'order.status.shipped', NULL, NOW() - INTERVAL '70 minutes'), + ('evt-ord-1003-status-pending', 'orders.status', 'default', 'ORD-20260404-1003', + 'order.status.pending', NULL, NOW() - INTERVAL '60 minutes'), + ('evt-ord-1003-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1003', + 'order.status.confirmed', NULL, NOW() - INTERVAL '50 minutes'), + ('evt-ord-1004-status-pending', 'orders.status', 'default', 'ORD-20260404-1004', + 'order.status.pending', NULL, NOW() - INTERVAL '45 minutes'), + ('evt-ord-1005-status-pending', 'orders.status', 'default', 'ORD-20260404-1005', + 'order.status.pending', NULL, NOW() - INTERVAL '30 minutes'), + ('evt-ord-1005-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1005', + 'order.status.confirmed', NULL, NOW() - INTERVAL '25 minutes'), + ('evt-ord-1005-status-shipped', 'orders.status', 'default', 'ORD-20260404-1005', + 'order.status.shipped', NULL, NOW() - INTERVAL '15 minutes'), + ('evt-ord-1005-status-delivered', 'orders.status', 'default', 'ORD-20260404-1005', + 'order.status.delivered', NULL, NOW() - INTERVAL '5 minutes'), + ('evt-ord-1006-status-pending', 'orders.status', 'default', 'ORD-20260404-1006', + 'order.status.pending', NULL, NOW() - INTERVAL '20 minutes'), + ('evt-ord-1006-status-cancelled', 'orders.status', 'default', 'ORD-20260404-1006', + 'order.status.cancelled', NULL, NOW() - INTERVAL '10 minutes'), + ('evt-ord-1007-status-pending', 'orders.status', 'default', 'ORD-20260404-1007', + 'order.status.pending', NULL, NOW() - INTERVAL '15 minutes'), + ('evt-ord-1007-status-confirmed', 'orders.status', 'default', 'ORD-20260404-1007', + 'order.status.confirmed', NULL, NOW() - INTERVAL '8 minutes'), + ('evt-ord-1008-status-pending', 'orders.status', 'default', 'ORD-20260404-1008', + 'order.status.pending', NULL, NOW() - INTERVAL '5 minutes') + """) + def health(self) -> dict: try: value = self.scalar("SELECT 1") diff --git a/tests/integration/test_order_timeline.py b/tests/integration/test_order_timeline.py new file mode 100644 index 00000000..d4351e1a --- /dev/null +++ b/tests/integration/test_order_timeline.py @@ -0,0 +1,159 @@ +"""Integration tests for Order 360 — GET /v1/entity/order/{order_id}/timeline +(ops-surfaces-spec.md §2, D2). Exercises the demo story pins (I7), the PII-free +customer allow-list (I3), the fallback-clock honesty (I12), and the +error_rate re-pin by arithmetic (I9). Tenant scoping (I8) and the +no-third-path ratchet (I1) live in test_tenant_isolation.py and +test_control_plane_store.py respectively — same pattern as the other ops +surfaces. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from src.serving.api.main import app + +pytestmark = pytest.mark.integration + +_PII_FIELD_NAMES = ("first_name", "last_name", "email", "phone", "birth_date") + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + db_path = tmp_path / "order-timeline.duckdb" + monkeypatch.setenv("DUCKDB_PATH", str(db_path)) + monkeypatch.setenv("SERVING_BACKEND", "duckdb") + monkeypatch.setenv("AGENTFLOW_AUTH_DISABLED", "true") + + with TestClient(app) as c: + yield c + + +def test_unknown_order_returns_404(client: TestClient): + response = client.get("/v1/entity/order/ORD-DOES-NOT-EXIST/timeline") + + assert response.status_code == 404 + assert response.json() == {"detail": "order/ORD-DOES-NOT-EXIST not found"} + + +def test_delivered_order_carries_full_story(client: TestClient): + # I7: ORD-20260404-1001 shows the full stage history, a >=3-row pipeline + # trail, and the USR-10001 customer block. + response = client.get("/v1/entity/order/ORD-20260404-1001/timeline") + + assert response.status_code == 200 + data = response.json() + + assert data["order"]["order_id"] == "ORD-20260404-1001" + assert data["order"]["status"] == "delivered" + + assert data["stage"]["current"] == "delivered" + assert data["stage"]["clock"] == "journal" + assert data["stage"]["in_stage_seconds"] > 0 + + statuses = [row["status"] for row in data["stage_history"]] + assert statuses == ["pending", "confirmed", "shipped", "delivered"] + # Ascending order — each stage entered later than the last. + timestamps = [row["at"] for row in data["stage_history"]] + assert timestamps == sorted(timestamps) + + assert len(data["pipeline_trail"]) >= 3 + trail_topics = {row["topic"] for row in data["pipeline_trail"]} + assert "orders.status" not in trail_topics + + assert data["customer"] is not None + assert data["customer"]["user_id"] == "USR-10001" + assert data["customer"]["total_orders"] == 34 + + +def test_pending_order_is_the_sole_breach_candidate(client: TestClient): + # I7: ORD-20260404-1004's single pending stage entry sits at created_at, + # 45 minutes ago — the demo's sole SLA breach once D3 wires budgets. + response = client.get("/v1/entity/order/ORD-20260404-1004/timeline") + + assert response.status_code == 200 + data = response.json() + + assert data["stage"]["current"] == "pending" + assert data["stage"]["clock"] == "journal" + # ~45 minutes, allow for wall-clock drift while the test runs. + assert 2600 < data["stage"]["in_stage_seconds"] < 2900 + assert [row["status"] for row in data["stage_history"]] == ["pending"] + + +def test_customer_block_is_a_pii_free_allowlist(client: TestClient): + response = client.get("/v1/entity/order/ORD-20260404-1001/timeline") + + assert response.status_code == 200 + data = response.json() + + assert set(data["customer"].keys()) == { + "user_id", + "total_orders", + "total_spent", + "first_order_at", + "last_order_at", + "preferred_category", + } + # Structural belt-and-braces: no PII field name anywhere in the payload. + for field_name in _PII_FIELD_NAMES: + assert field_name not in response.text + + +def test_order_without_stage_rows_reports_fallback_clock(client: TestClient): + # I12: an order written outside the stage-row writer (e.g. a bypass + # insert) degrades honestly to the created_at fallback instead of + # pretending to have a journal clock. + conn = client.app.state.query_engine._conn + conn.execute( + """ + INSERT INTO orders_v2 (order_id, user_id, status, total_amount, currency, created_at) + VALUES ('ORD-BYPASS-1', 'USR-10001', 'confirmed', 500.0, 'RUB', + NOW() - INTERVAL '10 minutes') + """ + ) + + response = client.get("/v1/entity/order/ORD-BYPASS-1/timeline") + + assert response.status_code == 200 + data = response.json() + assert data["stage"]["current"] == "confirmed" + assert data["stage"]["clock"] == "fallback" + assert data["stage_history"] == [] + assert 0 < data["stage"]["in_stage_seconds"] < 900 + + +def test_stage_vocabulary_outside_the_ladder_never_crashes(client: TestClient): + # I4 (D2's honest-degradation slice): a status the pipeline never wrote a + # stage row for — here because it isn't even in the contract vocabulary + # yet — still resolves cleanly rather than 500ing. No budget exists + # pre-D3 either way, so sla_minutes/breached stay null. + conn = client.app.state.query_engine._conn + conn.execute( + """ + INSERT INTO orders_v2 (order_id, user_id, status, total_amount, currency, created_at) + VALUES ('ORD-WEIRD-STATUS', 'USR-10001', 'on_hold', 500.0, 'RUB', + NOW() - INTERVAL '10 minutes') + """ + ) + + response = client.get("/v1/entity/order/ORD-WEIRD-STATUS/timeline") + + assert response.status_code == 200 + data = response.json() + assert data["stage"]["current"] == "on_hold" + assert data["stage"]["sla_minutes"] is None + assert data["stage"]["breached"] is None + + +def test_error_rate_repins_by_arithmetic_after_seeded_stage_rows(client: TestClient): + # I9: seeding 19 stage rows (spec §1.6) moves the demo error_rate + # denominator. 2 dead-letter rows over 13 ambient/lineage + 19 stage + # rows = 32 total in the 24h window. + response = client.get("/v1/metrics/error_rate?window=24h") + + assert response.status_code == 200 + assert response.json()["value"] == pytest.approx(2 / 32) diff --git a/tests/integration/test_tenant_isolation.py b/tests/integration/test_tenant_isolation.py index 812eabf1..78caf12f 100644 --- a/tests/integration/test_tenant_isolation.py +++ b/tests/integration/test_tenant_isolation.py @@ -187,6 +187,24 @@ def test_cross_tenant_entity_lookup_returns_404(client: TestClient): assert response.json() == {"detail": "order/ORD-ACME not found"} +def test_cross_tenant_order_timeline_returns_404(client: TestClient): + # ops-surfaces-spec.md §1.7 / invariant I8: Order 360 timeline scopes + # both ports' reads by the request tenant, same as the entity route. + response = client.get("/v1/entity/order/ORD-ACME/timeline", headers={"X-API-Key": "demo-key"}) + + assert response.status_code == 404 + assert response.json() == {"detail": "order/ORD-ACME not found"} + + +def test_tenant_api_key_reads_own_order_timeline(client: TestClient): + response = client.get("/v1/entity/order/ORD-ACME/timeline", headers={"X-API-Key": "acme-key"}) + + assert response.status_code == 200 + data = response.json() + assert data["order"]["order_id"] == "ORD-ACME" + assert data["order"]["user_id"] == "USR-ACME-2" + + def test_metric_cache_does_not_leak_across_tenants(client: TestClient): class FakeRedis: def __init__(self) -> None: diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index 38f871d8..04e31b32 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -851,3 +851,13 @@ def test_webhook_path_does_not_reach_into_the_engine_connection() -> None: ): source = (PROJECT_ROOT / relative).read_text(encoding="utf-8") assert "query_engine._conn" not in source, relative + + +def test_ops_timeline_path_does_not_reach_into_the_engine_connection_or_vault() -> None: + """ADR 0011 invariant I1: the ops surfaces (Order 360 timeline first, + D2) compose exactly the QueryEngine/ServingBackend and ControlPlaneStore + ports — no raw ``query_engine._conn`` reach, no vault DSN, ever.""" + source = (PROJECT_ROOT / "src/serving/api/routers/agent_query.py").read_text(encoding="utf-8") + assert "query_engine._conn" not in source + assert "_conn." not in source + assert "VAULT_DSN" not in source diff --git a/tests/unit/test_local_pipeline_clickhouse_mirror.py b/tests/unit/test_local_pipeline_clickhouse_mirror.py index 3aee9f0a..64563b89 100644 --- a/tests/unit/test_local_pipeline_clickhouse_mirror.py +++ b/tests/unit/test_local_pipeline_clickhouse_mirror.py @@ -59,10 +59,18 @@ def test_valid_order_mirrors_upsert_and_validated_journal_row(conn) -> None: assert (success, reason) == (True, "ok") kinds = [kind for kind, _ in sink.calls] - assert kinds == ["upsert_order", "record_pipeline_event"] - journal = sink.calls[1][1] + # ops-surfaces-spec.md §1.2: the orders.status stage-entry row is mirrored + # right after the domain upsert, ahead of the events.validated journal row. + assert kinds == ["upsert_order", "record_pipeline_event", "record_pipeline_event"] + stage_row = sink.calls[1][1] + assert stage_row["topic"] == "orders.status" + assert stage_row["entity_id"] == str(event["order_id"]) + assert stage_row["event_type"] == f"order.status.{event['status']}" + assert stage_row["latency_ms"] is None + journal = sink.calls[2][1] assert journal["topic"] == "events.validated" assert journal["event_id"] == str(event["event_id"]) + assert journal["entity_id"] == str(event["order_id"]) # DuckDB stays the canonical local store — the mirror is additive. row = conn.execute("SELECT COUNT(*) FROM orders_v2").fetchone() assert row is not None diff --git a/tests/unit/test_local_pipeline_entity_axis.py b/tests/unit/test_local_pipeline_entity_axis.py new file mode 100644 index 00000000..e42d00e9 --- /dev/null +++ b/tests/unit/test_local_pipeline_entity_axis.py @@ -0,0 +1,167 @@ +"""Unit coverage for the pipeline_events journal's entity_id axis and the +orders.status stage-entry writer (ops-surfaces-spec.md §1.2, §1.3 — D2). + +Before this, only the demo seed populated entity_id; the live write sites +(both DuckDB inserts here, plus clickhouse_sink.py's mirror, covered +separately in test_local_pipeline_clickhouse_mirror.py) left it NULL. +""" + +from __future__ import annotations + +import json +from collections.abc import Iterator + +import duckdb +import pytest + +from src.ingestion.producers.event_producer import generate_click, generate_order +from src.processing.local_pipeline import _derive_entity_id, _ensure_tables, _process_event + + +@pytest.fixture +def conn() -> Iterator[duckdb.DuckDBPyConnection]: + connection = duckdb.connect(":memory:") + _ensure_tables(connection) + try: + yield connection + finally: + connection.close() + + +def _order_event() -> dict: + _, event = generate_order() + return json.loads(event.model_dump_json()) + + +class TestDeriveEntityId: + def test_order_event_uses_order_id(self) -> None: + assert _derive_entity_id({"order_id": "ORD-1"}, "order.created") == "ORD-1" + + def test_user_event_uses_user_id(self) -> None: + assert _derive_entity_id({"user_id": "USR-1"}, "user.updated") == "USR-1" + + def test_product_event_uses_product_id(self) -> None: + assert _derive_entity_id({"product_id": "PROD-1"}, "product.updated") == "PROD-1" + + def test_session_event_uses_session_id(self) -> None: + assert _derive_entity_id({"session_id": "SES-1"}, "session.ended") == "SES-1" + + def test_unrelated_event_type_is_not_derivable(self) -> None: + assert _derive_entity_id({"order_id": "ORD-1"}, "payment.completed") is None + + def test_missing_field_is_not_synthesized(self) -> None: + # order.* but no order_id in the payload — NULL, never a made-up id. + assert _derive_entity_id({}, "order.created") is None + + +def test_validated_order_row_carries_entity_id(conn) -> None: + event = _order_event() + + success, _ = _process_event(conn, event) + + assert success + row = conn.execute( + "SELECT entity_id FROM pipeline_events WHERE topic = 'events.validated'" + ).fetchone() + assert row is not None + assert row[0] == str(event["order_id"]) + + +def test_deadletter_row_derives_entity_id_when_order_id_present(conn) -> None: + # order_id is present but the payload is otherwise schema-invalid. + malformed = {"event_type": "order.created", "order_id": "ORD-BAD-1"} + + success, reason = _process_event(conn, malformed) + + assert not success + assert reason.startswith("schema:") + row = conn.execute( + "SELECT entity_id FROM pipeline_events WHERE topic = 'events.deadletter'" + ).fetchone() + assert row is not None + assert row[0] == "ORD-BAD-1" + + +def test_deadletter_row_entity_id_null_when_not_derivable(conn) -> None: + success, reason = _process_event(conn, {"event_type": "order.created"}) + + assert not success + assert reason.startswith("schema:") + row = conn.execute( + "SELECT entity_id FROM pipeline_events WHERE topic = 'events.deadletter'" + ).fetchone() + assert row is not None + assert row[0] is None + + +def test_click_event_journal_row_entity_id_null(conn) -> None: + # Clickstream events carry no entity_id prefix mapping (session.* is the + # session family; raw click/page_view/add_to_cart events are unmapped). + _, click = generate_click() + event = json.loads(click.model_dump_json()) + + success, _ = _process_event(conn, event) + + assert success + row = conn.execute( + "SELECT entity_id FROM pipeline_events WHERE topic = 'events.validated'" + ).fetchone() + assert row is not None + assert row[0] is None + + +class TestOrderStatusStageRow: + def test_order_event_writes_a_stage_row(self, conn) -> None: + event = _order_event() + + success, _ = _process_event(conn, event) + + assert success + rows = conn.execute( + "SELECT event_type, entity_id, latency_ms, tenant_id " + "FROM pipeline_events WHERE topic = 'orders.status'" + ).fetchall() + assert len(rows) == 1 + event_type, entity_id, latency_ms, tenant_id = rows[0] + assert event_type == f"order.status.{event['status']}" + assert entity_id == str(event["order_id"]) + assert latency_ms is None + assert tenant_id == "default" + + def test_stage_row_topic_is_disjoint_from_validated_row(self, conn) -> None: + event = _order_event() + + success, _ = _process_event(conn, event) + + assert success + topics = { + row[0] + for row in conn.execute( + "SELECT DISTINCT topic FROM pipeline_events WHERE entity_id = ?", + [str(event["order_id"])], + ).fetchall() + } + assert topics == {"events.validated", "orders.status"} + + def test_non_order_event_writes_no_stage_row(self, conn) -> None: + _, click = generate_click() + event = json.loads(click.model_dump_json()) + + success, _ = _process_event(conn, event) + + assert success + row = conn.execute( + "SELECT COUNT(*) FROM pipeline_events WHERE topic = 'orders.status'" + ).fetchone() + assert row is not None + assert row[0] == 0 + + def test_deadletter_event_writes_no_stage_row(self, conn) -> None: + success, _ = _process_event(conn, {"event_type": "order.created", "order_id": "ORD-X"}) + + assert not success + row = conn.execute( + "SELECT COUNT(*) FROM pipeline_events WHERE topic = 'orders.status'" + ).fetchone() + assert row is not None + assert row[0] == 0 diff --git a/tests/unit/test_security_tooling_policy.py b/tests/unit/test_security_tooling_policy.py index a3e248c2..780cfc7d 100644 --- a/tests/unit/test_security_tooling_policy.py +++ b/tests/unit/test_security_tooling_policy.py @@ -65,7 +65,12 @@ def test_sql_injection_checks_are_not_globally_suppressed() -> None: # %s. All other adapter SQL is literal (the lease fragment is inlined and # the tenant/reason filters branch into full literal statements). "src/serving/control_plane/postgres.py": 3, - "src/serving/backends/clickhouse_backend.py": 7, + # D2 (reviewed 2026-07-04): the new orders.status stage-trail seed INSERT + # in initialize_demo_data follows the same pattern as the file's other six + # seed-block sites — a static f-string of hardcoded demo ids and + # ts()-formatted (trusted, generated) timestamps, no request-derived + # input. + "src/serving/backends/clickhouse_backend.py": 8, "src/serving/backends/duckdb_backend.py": 2, "src/serving/semantic_layer/nl_engine.py": 6, "src/serving/semantic_layer/query/engine.py": 1,