Hardening: durable dedup + stuck-incident recovery, signed alerts, drop dead fields

.env.example

@@ -7,3 +7,8 @@ SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
 # Optional — docker-compose Postgres password (defaults to "sentinel")
 POSTGRES_PASSWORD=sentinel
 ANTHROPIC_WORKSPACE_ID=   # only needed if the key is not workspace-scoped
+# Recommended — shared secret: core rejects /alert requests without a matching HMAC signature, and the
+# sandbox senders sign with it. Unset = unsigned alerts accepted (local dev). e.g. `openssl rand -hex 32`
+SENTINEL_ALERT_SECRET=
+# Optional — comma-separated dashboard origins allowed by CORS (default http://localhost:3000)
+SENTINEL_CORS_ORIGINS=

.github/workflows/ci.yml

@@ -8,9 +8,20 @@ jobs:
   backend:
     runs-on: ubuntu-latest
     env:
-      # tests mock all external boundaries; these only satisfy import-time env lookups
+      # unit tests mock all external boundaries; test_db_integration runs its SQL against the service below
       DATABASE_URL: postgresql://ci:ci@localhost:5432/ci
+      SENTINEL_DB_TESTS: "1"
       ANTHROPIC_API_KEY: ci-dummy
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: ci
+          POSTGRES_PASSWORD: ci
+          POSTGRES_DB: ci
+        ports: ["5432:5432"]
+        options: >-
+          --health-cmd pg_isready --health-interval 5s --health-timeout 5s --health-retries 10
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5

claude.md

@@ -46,7 +46,7 @@ Instead of connecting to high-risk production infrastructure, Sentinel operates
 ### How It Works:
 
 1. **Trigger:** A mock deployment script injects a faulty commit into the sandbox repository, spiking the traffic generator's error rates and triggering a simulated critical alert webhook.
-2. **Diagnose:** Sentinel intercepts the alert, queries local Postgres logs, pulls recent git diffs, ranks suspect commits using Claude Sonnet structured JSON outputs, and fetches matching runbooks using vector similarity.
+2. **Diagnose:** Sentinel intercepts the alert, reads its stack trace, pulls recent git diffs, ranks suspect commits using Claude Sonnet structured JSON outputs, and fetches matching runbooks using vector similarity.
 3. **Communicate:** The system constructs an incredibly concise operational summary and dispatches an asynchronous structured diagnostic card to an active Slack channel via an Incoming Webhook.
 4. **Document:** Upon simulated incident resolution, the orchestration loop synthesizes the entire timeline into an perfectly structured postmortem markdown document ready for human export to Notion/Confluence.
 
@@ -154,8 +154,7 @@ Communication across state machines utilizes strict, structured schemas. The sta
       }
     ],
     "impact_assessment": {
-      "error_rate_delta_pct": 340.5,
-      "estimated_affected_users": 142
+      "error_rate_delta_pct": 340.5
     }
   },
   "postmortem_draft_url": null

contract/pipeline_schema.json

@@ -48,8 +48,7 @@
         "impact_assessment": {
           "type": "object",
           "properties": {
-            "error_rate_delta_pct": { "type": ["number", "null"] },
-            "estimated_affected_users": { "type": ["integer", "null"] }
+            "error_rate_delta_pct": { "type": ["number", "null"] }
           }
         },
         "degraded": { "type": "boolean" },

core/db.py

@@ -27,39 +27,64 @@ def init_schema():
                     created_at   TIMESTAMPTZ DEFAULT NOW(),
                     updated_at   TIMESTAMPTZ DEFAULT NOW()
                 );
-                CREATE TABLE IF NOT EXISTS app_logs (
-                    id          SERIAL PRIMARY KEY,
-                    incident_id TEXT REFERENCES incidents(id),
-                    ts          TIMESTAMPTZ DEFAULT NOW(),
-                    level       TEXT,
-                    message     TEXT,
-                    endpoint    TEXT,
-                    raw         JSONB
-                );
+                -- dedup key. Rows created before this column existed stay NULL and never dedup.
+                ALTER TABLE incidents ADD COLUMN IF NOT EXISTS signature TEXT;
+                -- at most one open incident per error signature, across workers and restarts
+                CREATE UNIQUE INDEX IF NOT EXISTS incidents_one_open_per_signature
+                    ON incidents (signature) WHERE status IN ('triggered', 'triaging');
             """)
     conn.close()
 
 
-def save_incident(incident: dict):
+_OPEN = "status IN ('triggered', 'triaging')"  # must match the partial index predicate
+
+
+def open_incident(incident: dict, signature: str) -> tuple[str, bool]:
+    """Insert the incident unless one is already open for this signature.
+    Returns (incident_id, created); when not created, the id is the open incident's."""
     conn = _conn()
-    with conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                """INSERT INTO incidents (id, status, trigger_data, diagnostics, postmortem)
-                   VALUES (%s, %s, %s, %s, %s)
-                   ON CONFLICT (id) DO UPDATE SET
-                     status = EXCLUDED.status,
-                     diagnostics = EXCLUDED.diagnostics,
-                     updated_at = NOW()""",
-                (
-                    incident["incident_id"],
-                    incident["status"],
-                    json.dumps(incident["trigger"]),
-                    json.dumps(incident.get("diagnostics")),
-                    incident.get("postmortem_draft_url"),
-                ),
-            )
+    try:
+        for _ in range(3):  # the open incident can close between the conflict and the lookup
+            with conn:
+                with conn.cursor() as cur:
+                    cur.execute(
+                        f"""INSERT INTO incidents (id, status, trigger_data, signature)
+                            VALUES (%s, %s, %s, %s)
+                            ON CONFLICT (signature) WHERE {_OPEN} DO NOTHING
+                            RETURNING id""",
+                        (
+                            incident["incident_id"],
+                            incident["status"],
+                            json.dumps(incident["trigger"]),
+                            signature,
+                        ),
+                    )
+                    if cur.fetchone():
+                        return incident["incident_id"], True
+                    cur.execute(f"SELECT id FROM incidents WHERE signature = %s AND {_OPEN}", (signature,))
+                    row = cur.fetchone()
+                    if row:
+                        return row[0], False
+        raise RuntimeError(f"could not open or find an incident for signature {signature!r}")
+    finally:
+        conn.close()
+
+
+def stuck_incidents(older_than_s: float) -> list[dict]:
+    """Open incidents with no diagnostics whose last update is older than older_than_s:
+    the process running their triage died."""
+    conn = _conn()
+    with conn.cursor(cursor_factory=RealDictCursor) as cur:
+        cur.execute(
+            f"""SELECT id, trigger_data FROM incidents
+                WHERE {_OPEN} AND (diagnostics IS NULL OR diagnostics = 'null'::jsonb)
+                  AND updated_at < NOW() - make_interval(secs => %s)
+                ORDER BY created_at""",
+            (older_than_s,),
+        )
+        rows = cur.fetchall()
     conn.close()
+    return [dict(r) for r in rows]
 
 
 def get_incident(incident_id: str) -> dict | None:

core/main.py

@@ -1,32 +1,60 @@
+import hashlib
+import hmac
+import json
+import os
+import threading
 from contextlib import asynccontextmanager
-from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi import FastAPI, HTTPException, BackgroundTasks, Request
+from starlette.concurrency import run_in_threadpool
 from fastapi.middleware.cors import CORSMiddleware
 from . import db, orchestrator
 from .services import vector_store, git_client
 
+# Shared secret for alert senders. Unset = unsigned alerts accepted, with a startup warning (local dev).
+_ALERT_SECRET = os.getenv("SENTINEL_ALERT_SECRET", "")
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     db.init_schema()
     n = vector_store.ingest_runbooks(orchestrator.RUNBOOKS_DIR)
     print(f"[core] runbooks ingested: {n}")
+    if not _ALERT_SECRET:
+        print("[core] WARNING: SENTINEL_ALERT_SECRET unset, /alert accepts unsigned alerts (dev only)")
+    # off the startup path: re-triage can take ~15s per orphaned incident
+    threading.Thread(target=orchestrator.recover_stuck_incidents, daemon=True).start()
     yield
 
 
 app = FastAPI(title="Sentinel Core", lifespan=lifespan)
 
-# ponytail: dev-only origin; lock down to the deployed dashboard origin before shipping
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["http://localhost:3000"],
+    # set-but-empty (as in .env.example) must not mean "no origins"
+    allow_origins=(os.getenv("SENTINEL_CORS_ORIGINS") or "http://localhost:3000").split(","),
     allow_methods=["GET", "POST", "PATCH"],
     allow_headers=["*"],
 )
 
 
+def signature_ok(body: bytes, header: str | None, secret: str) -> bool:
+    """X-Sentinel-Signature: sha256=<hex HMAC-SHA256 of the raw body>, the scheme GitHub and Sentry use."""
+    expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(expected, header or "")
+
+
 @app.post("/alert")
-def receive_alert(payload: dict, background_tasks: BackgroundTasks) -> dict:
-    incident = orchestrator.handle_alert(payload)
+async def receive_alert(request: Request, background_tasks: BackgroundTasks) -> dict:
+    body = await request.body()
+    if _ALERT_SECRET and not signature_ok(body, request.headers.get("x-sentinel-signature"), _ALERT_SECRET):
+        raise HTTPException(401, "bad or missing X-Sentinel-Signature")
+    try:
+        payload = json.loads(body)
+    except ValueError:
+        raise HTTPException(400, "body must be JSON")
+    if not isinstance(payload, dict):
+        raise HTTPException(400, "body must be a JSON object")
+    incident = await run_in_threadpool(orchestrator.handle_alert, payload)  # blocking DB call
     if not incident.get("suppressed"):
         background_tasks.add_task(orchestrator.run_diagnostics, incident["incident_id"], payload)
     return incident

core/orchestrator.py

@@ -8,42 +8,28 @@
 REPO_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  # Sentinel root
 RUNBOOKS_DIR = os.path.join(REPO_PATH, "sandbox", "runbooks")
 
-# ponytail: in-process sliding-window dedup; per-worker, resets on restart.
-# Move to Postgres/Redis if you run multiple workers or need durability.
+# Open-incident dedup lives in Postgres (a partial unique index on signature). Only the optional burst
+# threshold is in memory. ponytail: per-worker, resets on restart; it is off at the default threshold of 1.
 _DEDUP_WINDOW_S = float(os.getenv("SENTINEL_DEDUP_WINDOW_S", "300"))
 _DEDUP_THRESHOLD = int(os.getenv("SENTINEL_DEDUP_THRESHOLD", "1"))  # Nth hit in window fires
 _recent_alerts: dict[str, list[float]] = defaultdict(list)  # signature -> hit times
 # Pass matched runbook text into commit ranking. Off: Phase 9 A/B found no measurable benefit (ADR-7).
 _RAG_RANKING = os.getenv("SENTINEL_RAG_RANKING", "0") == "1"
-_open_signatures: dict[str, str] = {}  # signature -> live incident_id
+# Triage takes ~15s; an open incident with no diagnostics this old was orphaned by a dead process.
+STUCK_AFTER_S = float(os.getenv("SENTINEL_STUCK_AFTER_S", "300"))
 
 
-def _dedup_gate(signature: str) -> tuple[str, str | None]:
-    """'fire' (open a new incident), 'duplicate' (fold into an active one),
-    or 'suppress' (below the burst threshold)."""
+def _below_burst_threshold(signature: str) -> bool:
     now = time.monotonic()
     hits = _recent_alerts[signature]
     hits[:] = [t for t in hits if t >= now - _DEDUP_WINDOW_S]
     hits.append(now)
-
-    if signature in _open_signatures:
-        return "duplicate", _open_signatures[signature]
-    if len(hits) < _DEDUP_THRESHOLD:
-        return "suppress", None
-    return "fire", None
-
-
-def _clear_signature(signature: str):
-    _open_signatures.pop(signature, None)
+    return len(hits) < _DEDUP_THRESHOLD
 
 
 def handle_alert(alert_data: dict) -> dict:
     signature = alert_data.get("error_signature", "")
-    decision, existing_id = _dedup_gate(signature)
-    if decision == "duplicate":
-        print(f"[dedup] alert folded into active incident {existing_id}")
-        return {"incident_id": existing_id, "status": "duplicate", "suppressed": True}
-    if decision == "suppress":
+    if _below_burst_threshold(signature):
         seen = len(_recent_alerts[signature])
         print(f"[dedup] alert suppressed ({seen}/{_DEDUP_THRESHOLD} in {int(_DEDUP_WINDOW_S)}s window)")
         return {"incident_id": None, "status": "suppressed", "suppressed": True}
@@ -56,12 +42,23 @@ def handle_alert(alert_data: dict) -> dict:
         "diagnostics": None,
         "postmortem_draft_url": None,
     }
-    db.save_incident(incident)
-    _open_signatures[signature] = incident["incident_id"]
-    print(f"[orchestrator] {incident['incident_id']} created -> triggered")
+    incident_id, created = db.open_incident(incident, signature)
+    if not created:
+        print(f"[dedup] alert folded into active incident {incident_id}")
+        return {"incident_id": incident_id, "status": "duplicate", "suppressed": True}
+    print(f"[orchestrator] {incident_id} created -> triggered")
     return incident
 
 
+def recover_stuck_incidents() -> list[str]:
+    """Re-run triage for incidents orphaned mid-triage (crash, restart). Call once at startup."""
+    stuck = db.stuck_incidents(STUCK_AFTER_S)
+    for inc in stuck:
+        print(f"[orchestrator] {inc['id']} orphaned mid-triage -> re-running diagnostics")
+        run_diagnostics(inc["id"], inc["trigger_data"])
+    return [inc["id"] for inc in stuck]
+
+
 def run_diagnostics(incident_id: str, alert_data: dict):
     """Runs in a background task after the alert response is sent."""
     print(f"[orchestrator] {incident_id} -> triaging")
@@ -95,10 +92,9 @@ def run_diagnostics(incident_id: str, alert_data: dict):
             "suspect_commits": ranked,
             # runbook text is prompt-only; never persisted or sent to Slack
             "matched_runbooks": [{k: v for k, v in r.items() if k != "content"} for r in runbooks],
-            "impact_assessment": {
-                "error_rate_delta_pct": alert_data.get("error_rate_pct"),
-                "estimated_affected_users": None,
-            },
+            # no affected-user estimate: nothing upstream reports user ids, and the dashboard
+            # renders no fabricated figures
+            "impact_assessment": {"error_rate_delta_pct": alert_data.get("error_rate_pct")},
         }
         if degraded_reason:
             diagnostics["degraded"] = True
@@ -110,8 +106,7 @@ def run_diagnostics(incident_id: str, alert_data: dict):
 
         db.update_diagnostics(incident_id, diagnostics)
         if degraded_reason:
-            db.update_status(incident_id, "degraded")
-            _clear_signature(alert_data.get("error_signature", ""))  # terminal: stop folding new alerts
+            db.update_status(incident_id, "degraded")  # terminal: leaves the open set, new alerts open anew
             print(f"[orchestrator] {incident_id} -> degraded (terminal)")
         else:
             alert_dt = datetime.fromisoformat(alert_data["timestamp"].replace("Z", "+00:00"))
@@ -126,8 +121,15 @@ def run_diagnostics(incident_id: str, alert_data: dict):
 
     except Exception as e:
         print(f"[orchestrator] {incident_id} diagnostics failed: {e}")
-        db.update_status(incident_id, "triggered")  # revert so it can be retried
-        _clear_signature(alert_data.get("error_signature", ""))  # allow a retry alert through
+        try:  # terminal and visible, rather than an incident left open with nothing in it
+            reason = f"Diagnostics failed — manual investigation required ({e})"
+            db.update_diagnostics(
+                incident_id,
+                {"suspect_commits": [], "matched_runbooks": [], "degraded": True, "degraded_reason": reason},
+            )
+            db.update_status(incident_id, "degraded")
+        except Exception as e2:  # db down too: the row stays open and empty; startup recovery retries it
+            print(f"[orchestrator] {incident_id} could not record the failure ({e2}); left for recovery")
 
 
 def resolve_incident(incident_id: str) -> dict:
@@ -143,10 +145,7 @@ def resolve_incident(incident_id: str) -> dict:
             "trigger and diagnostics data remain on the incident record."
         )
         print(f"[orchestrator] {incident_id} -- postmortem LLM unavailable, degrading: {e}")
-    db.update_postmortem(incident_id, draft)
-    _clear_signature(
-        (incident.get("trigger_data") or {}).get("error_signature", "")
-    )  # resolved: reopen to new alerts
+    db.update_postmortem(incident_id, draft)  # status -> resolved: leaves the open set
     print(f"[orchestrator] {incident_id} -> resolved, postmortem generated")
     print(f"[metrics] {incident_id} resolve_to_postmortem_s={time.monotonic() - t0:.1f}")
     return db.get_incident(incident_id)