diff --git a/README.md b/README.md index 8969886c..13ef218d 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ One public interface: - `/api/v1/commercial_purchase_approval_packets/latest` exposes the KRW 2,000,000,000 commercial purchase approval packet across proposal, close, procurement, contract, value, security, onboarding, operations, analytics truthfulness, Figma review, review-process policy, packaging decision, and buyer signature/budget authority follow-ups. - `/api/v1/commercial_due_diligence_rooms/latest` exposes the KRW 2,000,000,000 commercial due diligence room across purchase approval, runtime API evidence, admin trace/access evidence, security, commercial terms, value analytics, implementation readiness, Figma review, review-process policy, packaging decision, and buyer/external missing artifacts. - `/api/v1/commercial_investment_committee_memos/latest` exposes the KRW 2,000,000,000 commercial investment committee memo across due diligence, purchase approval, financial case, risk/security, commercial terms, implementation readiness, Figma review, review-process policy, packaging decision, and buyer/external approval conditions. +- `/api/v1/operational_alert_classifications` classifies external gateway alerts before incident routing. The first production rule suppresses the complete LiteLLM Prisma P2028 non-blocking spend-log signature from outage/page routing while keeping inference health, latency, and 5xx alerts active. One fused orchestration loop: @@ -150,6 +151,7 @@ See [docs/architecture.md](docs/architecture.md) for the source-backed analysis. - [Commercial due diligence room](docs/commercial_due_diligence_room.md) - [Commercial investment committee memo](docs/commercial_investment_committee_memo.md) - [Commercial plugin operating model](docs/commercial_plugin_operating_model.md) +- [Operational alert classification](docs/operational_alerts.md) - [Figma artifacts](docs/figma_artifacts.md) - [Plugin-driven implementation plan](docs/superpowers/plans/2026-07-02-plugin-driven-product-design.md) - [Commercial plugin readiness plan](docs/superpowers/plans/2026-07-02-commercial-plugin-readiness.md) diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 20f5f83a..252e3fed 100644 --- a/contextual_orchestrator/__init__.py +++ b/contextual_orchestrator/__init__.py @@ -2,6 +2,7 @@ from .credentials import NotConfigured, get_credential, register_credential from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents +from .operational_alerts import classify_operational_alert __all__ = [ "ModelAgent", @@ -11,4 +12,5 @@ "get_credential", "register_credential", "NotConfigured", + "classify_operational_alert", ] diff --git a/contextual_orchestrator/api_contract.py b/contextual_orchestrator/api_contract.py index ca9d3b5b..7eef5850 100644 --- a/contextual_orchestrator/api_contract.py +++ b/contextual_orchestrator/api_contract.py @@ -279,6 +279,26 @@ "responses": {"200": {"description": "Commercial investment committee memo"}}, } }, + "/api/v1/operational_alert_classifications": { + "post": { + "operationId": "create_operational_alert_classification", + "summary": "Classify external gateway alerts before incident routing", + "security": [{"admin_bearer_auth": []}], + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": True, + "description": "External alert payload from monitoring, log, or webhook systems.", + }, + } + }, + }, + "responses": {"201": {"description": "Operational alert classification"}}, + } + }, "/api/v1/workflow_runs": { "get": { "operationId": "list_workflow_runs", diff --git a/contextual_orchestrator/operational_alerts.py b/contextual_orchestrator/operational_alerts.py new file mode 100644 index 00000000..0c2b94ba --- /dev/null +++ b/contextual_orchestrator/operational_alerts.py @@ -0,0 +1,139 @@ +"""Operational alert classification for gateway incident routing.""" + +from __future__ import annotations + +from collections.abc import Mapping +import re +from typing import Any + + +_LITELLM_PATTERN = re.compile(r"\blitellm\b", re.IGNORECASE) +_PRISMA_PATTERN = re.compile(r"\bprisma\b", re.IGNORECASE) +_P2028_PATTERN = re.compile(r"(?i)(?:error_code[\"']?\s*[:=]\s*[\"']?)?P2028\b") +_TRANSACTION_START_PATTERN = re.compile(r"unable to start a transaction in the given time", re.IGNORECASE) +_NON_BLOCKING_PATTERN = re.compile(r"\[?non[- ]blocking\]?", re.IGNORECASE) +_SPEND_LOG_PATTERN = re.compile(r"update\s+spend\s+logs?|spend[-_ ]?logs?", re.IGNORECASE) +_DB_EXCEPTION_PATTERN = re.compile(r"db[_ -]?exceptions?|db read/write call failed", re.IGNORECASE) + + +_SUPPRESSED_ACTIONS = [ + "Drop from outage/page routing while recording a suppressed-alert audit event.", + "Keep inference health, latency, and user-facing HTTP error-rate alerts active.", + "Review LiteLLM, PgCat, and PostgreSQL pool budgets if suppressed counts trend upward.", +] + +_ESCALATION_ACTIONS = [ + "Route to normal operator review or incident policy.", + "Check inference health, request error rate, and customer-facing latency before downgrading.", + "Only suppress after a specific non-blocking persistence signature is added and tested.", +] + + +def classify_operational_alert(alert: Mapping[str, Any]) -> dict[str, Any]: + """Classify an external gateway alert before it becomes a page or outage. + + The classifier intentionally returns a derived decision, not the raw alert + body. Alert payloads can contain URLs, headers, or request fragments, so the + response keeps only signal names and operator actions. + """ + text = _flatten_alert_text(alert) + matched_signals = _matched_signals(text) + is_litellm_p2028_spend_log = { + "litellm_proxy", + "prisma_client", + "p2028_transaction_start_timeout", + "transaction_start_timeout", + "non_blocking_context", + "spend_log_update", + }.issubset(matched_signals) + + if is_litellm_p2028_spend_log: + return { + "classification_id": "litellm_prisma_p2028_spend_log_non_blocking", + "source_component": "external_litellm_proxy", + "classification_status": "suppressed", + "incident_routing": "drop", + "page_required": False, + "service_impact": "none_observed", + "normalized_severity": "info", + "reason_code": "non_blocking_spend_log_transaction_timeout", + "matched_signals": sorted(matched_signals), + "rationale": ( + "LiteLLM marked the Prisma spend-log update as non-blocking; " + "P2028 transaction-start pressure is a persistence backlog signal, " + "not standalone proof that inference is unavailable." + ), + "operator_actions": list(_SUPPRESSED_ACTIONS), + "audit_event_name": "operational_alert_suppressed", + } + + return { + "classification_id": "generic_gateway_alert_review", + "source_component": "external_gateway", + "classification_status": "escalate", + "incident_routing": "normal_policy", + "page_required": _page_required(alert, text), + "service_impact": "unknown", + "normalized_severity": _normalized_severity(alert, text), + "reason_code": "no_suppression_signature_match", + "matched_signals": sorted(matched_signals), + "rationale": ( + "The alert did not match the complete non-blocking LiteLLM Prisma " + "spend-log P2028 signature, so it must not be suppressed automatically." + ), + "operator_actions": list(_ESCALATION_ACTIONS), + "audit_event_name": "operational_alert_escalated", + } + + +def _matched_signals(text: str) -> set[str]: + signals: set[str] = set() + if _LITELLM_PATTERN.search(text): + signals.add("litellm_proxy") + if _PRISMA_PATTERN.search(text): + signals.add("prisma_client") + if _P2028_PATTERN.search(text): + signals.add("p2028_transaction_start_timeout") + if _TRANSACTION_START_PATTERN.search(text): + signals.add("transaction_start_timeout") + if _NON_BLOCKING_PATTERN.search(text): + signals.add("non_blocking_context") + if _SPEND_LOG_PATTERN.search(text): + signals.add("spend_log_update") + if _DB_EXCEPTION_PATTERN.search(text): + signals.add("db_exception_alert") + return signals + + +def _flatten_alert_text(value: Any) -> str: + if isinstance(value, Mapping): + parts: list[str] = [] + for key, child in value.items(): + parts.append(str(key)) + parts.append(_flatten_alert_text(child)) + return "\n".join(parts) + if isinstance(value, list | tuple | set): + return "\n".join(_flatten_alert_text(item) for item in value) + if value is None: + return "" + return str(value) + + +def _normalized_severity(alert: Mapping[str, Any], text: str) -> str: + level = str(alert.get("level") or alert.get("severity") or "").strip().lower() + if not level: + match = re.search(r"(?im)^\s*(level|severity)\s*:\s*([a-z]+)\s*$", text) + level = match.group(2).lower() if match else "" + if level in {"critical", "fatal"}: + return "critical" + if level in {"high", "error"}: + return "high" + if level in {"medium", "warning", "warn"}: + return "warning" + if level in {"low", "info", "informational"}: + return "info" + return "unknown" + + +def _page_required(alert: Mapping[str, Any], text: str) -> bool: + return _normalized_severity(alert, text) in {"critical", "high"} diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index cb9c5a34..4f06b2d3 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -21,6 +21,7 @@ redact_value, sse_stream_body, ) +from .operational_alerts import classify_operational_alert ALLOWED_CHAT_KEYS = {"model", "messages", "orchestration", "orchestration_mode", "mode", "include_orchestration_trace", "stream"} @@ -484,7 +485,8 @@ def do_PATCH(self) -> None: # noqa: N802 def do_POST(self) -> None: # noqa: N802 try: path = urllib.parse.urlparse(self.path).path - scope = "admin" if path == "/admin/simulate" else "inference" + admin_post_paths = {"/admin/simulate", "/api/v1/operational_alert_classifications"} + scope = "admin" if path in admin_post_paths else "inference" self._authorize(scope) body = self._read_json() @@ -549,6 +551,22 @@ def do_POST(self) -> None: # noqa: N802 evaluation_run = self._run(lambda: orchestrator.run_evaluation([str(item) for item in prompts], mode=mode)) self._send(_response_payload(evaluation_run, include_trace), 201) return + if path == "/api/v1/operational_alert_classifications": + classification = classify_operational_alert(body) + orchestrator.record_analytics_event( + "operational_alert_classified", + { + "endpoint_path": "/api/v1/operational_alert_classifications", + "actor_scope": "admin", + "status_code": 201, + "classification_status": classification["classification_status"], + "reason_code": classification["reason_code"], + "incident_routing": classification["incident_routing"], + "page_required": classification["page_required"], + }, + ) + self._send(classification, 201) + return self._send_error(404, "route_not_found", "not found") except json.JSONDecodeError: self._send_error(400, "invalid_json", "request body is not valid JSON") diff --git a/docs/operational_alerts.md b/docs/operational_alerts.md new file mode 100644 index 00000000..c4464db0 --- /dev/null +++ b/docs/operational_alerts.md @@ -0,0 +1,73 @@ +# Operational Alert Classification + +## Purpose + +`contextual-orchestrator` can sit in front of model-gateway operations as a +small incident-routing guard. The guard classifies external alerts before they +become outage tickets or pages. It does not mutate LiteLLM, PostgreSQL, PgCat, +or the monitoring backend; it returns a deterministic routing decision that an +alert pipeline can apply. + +## LiteLLM Prisma P2028 Spend-Log Rule + +The first supported suppression rule targets this complete signature: + +- `LiteLLM` +- `Prisma` +- `P2028` +- `Unable to start a transaction in the given time` +- `[Non-Blocking]` +- `update spend logs` + +When all signals are present, the classifier returns: + +- `classification_status: suppressed` +- `incident_routing: drop` +- `page_required: false` +- `reason_code: non_blocking_spend_log_transaction_timeout` + +Rationale: LiteLLM has already marked the spend-log write path as +non-blocking. A transaction-start timeout here is a persistence backlog or pool +pressure signal. By itself, it is not proof that inference is unavailable or +that users are receiving failed responses. + +## Non-Suppression Cases + +The classifier must not suppress similar-looking alerts when the complete +signature is missing. Examples: + +- P2028 during key lookup, request authorization, model routing, or any + user-facing request path. +- LiteLLM or Prisma DB errors without `[Non-Blocking]`. +- Spend or ledger errors paired with elevated 5xx, latency, health-check, or + customer-facing SLO alerts. + +Those return `classification_status: escalate` and follow normal incident +policy. + +## API + +```bash +curl -s http://127.0.0.1:8000/api/v1/operational_alert_classifications \ + -H "authorization: Bearer $CONTEXTUAL_ORCHESTRATOR_ADMIN_TOKEN" \ + -H "content-type: application/json" \ + -d @alert.json | jq . +``` + +The endpoint is admin-scoped because alert payloads can contain operational +metadata. The response does not echo the raw alert body; it exposes only matched +signal names, the routing decision, a reason code, and operator actions. + +## Wiring + +1. Send the monitoring webhook payload to + `/api/v1/operational_alert_classifications`. +2. If `incident_routing == "drop"` and `page_required == false`, do not create + an outage/page. Record the classifier response as an audit event. +3. Keep separate inference health, latency, 5xx, and customer-SLO alerts active. +4. Trend the suppressed count. If it rises, review LiteLLM/PgCat/PostgreSQL pool + budget and spend-log backlog during maintenance. + +Project coordination follows ContextualWisdomLab/.github#363: read Project #1 +first, keep the active work item visible, and link the PR or issue back to the +Project item. diff --git a/docs/rest_api_design.md b/docs/rest_api_design.md index a2da681c..113c0adf 100644 --- a/docs/rest_api_design.md +++ b/docs/rest_api_design.md @@ -43,6 +43,7 @@ | `GET` | `/api/v1/commercial_purchase_approval_packets/latest` | Read KRW 2,000,000,000 commercial purchase approval packet | | `GET` | `/api/v1/commercial_due_diligence_rooms/latest` | Read KRW 2,000,000,000 commercial due diligence room | | `GET` | `/api/v1/commercial_investment_committee_memos/latest` | Read KRW 2,000,000,000 commercial investment committee memo | +| `POST` | `/api/v1/operational_alert_classifications` | Classify external gateway alerts before outage/page routing | | `POST` | `/api/v1/workflow_runs` | Create a route/conduct run | | `GET` | `/api/v1/workflow_runs` | List recent workflow runs | | `GET` | `/api/v1/workflow_runs?page_number=1&page_size=20` | Paginate workflow run history with deterministic page metadata | @@ -90,6 +91,7 @@ These product surfaces are now implemented in this prototype: | `GET` | `/api/v1/commercial_purchase_approval_packets/latest` | Produce the buyer purchase approval packet that ties proposal, close, procurement, contract, value, security, onboarding, operations, analytics, Figma, review-process policy, packaging decision, and buyer authority follow-ups into one runtime approval artifact. | Fugu API adoption; TRINITY verification; Conductor trace/access evidence; buyer finance, procurement, legal, security, and implementation approval. | | `GET` | `/api/v1/commercial_due_diligence_rooms/latest` | Produce the buyer due diligence room that ties purchase approval, runtime API evidence, admin trace/access evidence, security, commercial terms, value analytics, implementation readiness, Figma, review-process policy, packaging decision, and buyer/external missing artifacts into one runtime diligence artifact. | Fugu API adoption; TRINITY verification; Conductor trace/access evidence; buyer diligence committee review. | | `GET` | `/api/v1/commercial_investment_committee_memos/latest` | Produce the investment committee memo that ties due diligence, purchase approval, financial case, risk/security, commercial terms, implementation readiness, Figma, review-process policy, packaging decision, and buyer/external approval conditions into one executive recommendation artifact. | Fugu API adoption; TRINITY verification; Conductor trace/access evidence; executive investment committee review. | +| `POST` | `/api/v1/operational_alert_classifications` | Classify external gateway alerts into suppressed or normal-policy routing before they become outage tickets or pages. | Fugu-style hidden gateway operations need operator-visible evidence and safe routing controls; production operations readiness requires incident evidence to distinguish service impact from non-blocking persistence noise. | ## Production Library Target diff --git a/tests/test_operational_alerts.py b/tests/test_operational_alerts.py new file mode 100644 index 00000000..64bbbd9f --- /dev/null +++ b/tests/test_operational_alerts.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sys +import threading +import urllib.error +import urllib.request + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator, classify_operational_alert # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + + +LITELLM_P2028_ALERT = { + "alert_type": "db_exceptions", + "level": "High", + "timestamp": "01:15:03", + "message": ( + 'DB read/write call failed: 504: {"is_panic":false,"message":"Transaction API error: ' + 'Unable to start a transaction in the given time.","meta":{"error":"Unable to start a ' + 'transaction in the given time."},"error_code":"P2028"}' + "[Non-Blocking]LiteLLM Prisma Client Exception - update spend logs" + ), + "traceback": "File /usr/lib/python3.13/site-packages/litellm/proxy/db/db_spend_update_writer.py", + "proxy_url": "https://llm-gateway.hyosung.com", +} + + +def test_litellm_p2028_spend_log_alert_is_suppressed_from_incident_routing() -> None: + classification = classify_operational_alert(LITELLM_P2028_ALERT) + + assert classification["classification_status"] == "suppressed" + assert classification["incident_routing"] == "drop" + assert classification["page_required"] is False + assert classification["normalized_severity"] == "info" + assert classification["reason_code"] == "non_blocking_spend_log_transaction_timeout" + assert { + "litellm_proxy", + "prisma_client", + "p2028_transaction_start_timeout", + "transaction_start_timeout", + "non_blocking_context", + "spend_log_update", + "db_exception_alert", + }.issubset(set(classification["matched_signals"])) + + rendered = json.dumps(classification) + assert "llm-gateway.hyosung.com" not in rendered + assert "db_spend_update_writer.py" not in rendered + + +def test_p2028_without_non_blocking_spend_log_signature_is_not_suppressed() -> None: + classification = classify_operational_alert( + { + "alert_type": "db_exceptions", + "level": "High", + "message": "LiteLLM Prisma failed with P2028 during key lookup before request authorization.", + } + ) + + assert classification["classification_status"] == "escalate" + assert classification["incident_routing"] == "normal_policy" + assert classification["page_required"] is True + assert classification["reason_code"] == "no_suppression_signature_match" + + +def test_unknown_low_signal_alert_stays_on_normal_policy_without_page() -> None: + classification = classify_operational_alert({"level": "Info", "message": "background cache refresh skipped"}) + + assert classification["classification_status"] == "escalate" + assert classification["incident_routing"] == "normal_policy" + assert classification["page_required"] is False + assert classification["normalized_severity"] == "info" + + +def _post_json(url: str, payload: dict[str, object], token: str) -> tuple[int, dict[str, object]]: + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "authorization": f"Bearer {token}", + "content-type": "application/json", + "connection": "close", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def _serve() -> tuple[object, int]: + orchestrator = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(admin_token="admin_token", inference_token="inference_token"), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, server.server_address[1] + + +def test_operational_alert_classification_endpoint_is_admin_scoped() -> None: + server, port = _serve() + url = f"http://127.0.0.1:{port}/api/v1/operational_alert_classifications" + try: + denied_status, denied = _post_json(url, LITELLM_P2028_ALERT, "inference_token") + status, body = _post_json(url, LITELLM_P2028_ALERT, "admin_token") + finally: + server.shutdown() + + assert denied_status == 401 + assert denied["error"]["code"] == "unauthorized" + assert status == 201 + assert body["classification_status"] == "suppressed" + assert body["incident_routing"] == "drop" + + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print(f"ok {name}") + print("ok")