multi-entity stream profiles.

src/fgai/incidents.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
+import hashlib
+import json
+import time
 from collections import defaultdict
+from pathlib import Path
 
 from .entities import entity_type
 from .models import AnomalyFinding
@@ -39,7 +43,9 @@ def build_incidents(anomalies: list[AnomalyFinding], field_deviations: dict[str,
         severity = "critical" if score >= 85 else "high" if score >= 60 else "medium" if score >= 35 else "low"
         streams = sorted({stream for item in group["correlations"] for stream in item.get("streams", [])} | {str(item.get("stream_name") or item.get("stream_title") or item.get("stream_id", "")) for item in group["fields"] if item.get("stream_id") or item.get("stream_name") or item.get("stream_title")})
         timeline = sorted(group["timeline"], key=lambda item: str(item.get("timestamp", "")))[:20]
+        incident_id = hashlib.sha256(json.dumps({"entity": entity, "streams": streams, "evidence": list(dict.fromkeys(str(item) for item in group["evidence"] if item))[:4]}, sort_keys=True).encode("utf-8")).hexdigest()[:16]
         incidents.append({
+            "id": incident_id,
             "entity": entity,
             "entity_type": entity_type(entity),
             "score": score,
@@ -52,3 +58,50 @@ def build_incidents(anomalies: list[AnomalyFinding], field_deviations: dict[str,
             "last_seen": timeline[-1].get("timestamp", "") if timeline else "",
         })
     return sorted(incidents, key=lambda item: int(item["score"]), reverse=True)
+
+
+class IncidentStore:
+    def __init__(self, path: str = "state/signalscope-incidents.json") -> None:
+        self.path = Path(path)
+
+    def entries(self) -> dict[str, dict[str, object]]:
+        try:
+            items = json.loads(self.path.read_text(encoding="utf-8"))
+        except (FileNotFoundError, json.JSONDecodeError):
+            return {}
+        return {str(key): value for key, value in items.items() if isinstance(value, dict)} if isinstance(items, dict) else {}
+
+    def apply(self, incidents: list[dict[str, object]]) -> list[dict[str, object]]:
+        states = self.entries()
+        now = int(time.time())
+        changed = False
+        for incident in incidents:
+            incident_id = str(incident.get("id", ""))
+            if not incident_id:
+                continue
+            state = states.get(incident_id)
+            if not state:
+                state = {"status": "open", "note": "", "created_at": now, "updated_at": now}
+                states[incident_id] = state
+                changed = True
+            incident["lifecycle_status"] = str(state.get("status", "open"))
+            incident["note"] = str(state.get("note", ""))
+            incident["updated_at"] = int(state.get("updated_at", 0) or 0)
+        if changed:
+            self._write(states)
+        return incidents
+
+    def update(self, incident_id: str, status: str, note: str = "") -> dict[str, object]:
+        status = status.lower()
+        if status not in {"open", "acknowledged", "resolved"}:
+            raise ValueError("invalid incident status")
+        states = self.entries()
+        current = states.get(incident_id, {"created_at": int(time.time())})
+        entry = {**current, "status": status, "note": note, "updated_at": int(time.time())}
+        states[incident_id] = entry
+        self._write(states)
+        return {"id": incident_id, **entry}
+
+    def _write(self, states: dict[str, dict[str, object]]) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.path.write_text(json.dumps(states, indent=2, sort_keys=True), encoding="utf-8")