Build Week: Preserve SOG correlation IDs in Argus

agent/src/webhook.py

@@ -219,6 +219,9 @@ async def process_alert(payload: dict) -> None:
         tags=payload.get("tags", []),
         hostname=payload.get("hostname"),
         assessment=assessment_dict,
+        correlation_id=payload.get("correlation_id") or raw_fields.get("sentinel.correlation_id"),
+        provenance=payload.get("provenance", "replayed" if payload.get("replayed") else "observed"),
+        replayed=bool(payload.get("replayed")),
     )
 
     # Upgrade entity security posture if reasoning found HIGH/CRITICAL and not FP

agent/src/world_model.py

@@ -118,6 +118,9 @@ async def post_finding(
     tags: list,
     hostname: str | None = None,
     assessment: dict | None = None,
+    correlation_id: str | None = None,
+    provenance: str = "observed",
+    replayed: bool = False,
 ) -> dict[str, Any]:
     finding_event_id = event_id(dedup_key, alert_time)
     affected_entity_id = entity_id(raw_fields, hostname)
@@ -130,6 +133,7 @@ async def post_finding(
         "priority": priority,
         "tags": tags,
         "raw_fields": raw_fields,
+        "provenance": provenance,
     }
     if assessment:
         payload["assessment"] = assessment
@@ -140,10 +144,13 @@ async def post_finding(
         "source": "argus",
         "timestamp": alert_time,
         "severity": severity,
+        "replayed": replayed,
         "payload": payload,
     }
     if affected_entity_id:
         body["entity_id"] = affected_entity_id
+    if correlation_id:
+        body["correlation_id"] = correlation_id
 
     result = await _request("POST", "/findings", body, timeout=8.0)
     log_method = log.warning if result["status"] == "failed" else log.debug

agent/tests/test_world_model.py

@@ -102,6 +102,24 @@ async def test_post_finding_uses_world_model_contract():
     assert body["severity"] == "critical"
     assert body["payload"]["finding_type"] == "falco_alert"
     assert body["payload"]["description"] == "Unexpected shell spawned"
+    assert body["payload"]["provenance"] == "observed"
+
+
+@pytest.mark.asyncio
+async def test_post_finding_preserves_explicit_correlation_id():
+    FakeClient.responses = [response(200, {"status": "accepted"})]
+    await world_model.post_finding(**finding_args(correlation_id="case-argus-phoenix-1"))
+    body = FakeClient.requests[0][2]
+    assert body["correlation_id"] == "case-argus-phoenix-1"
+
+
+@pytest.mark.asyncio
+async def test_replay_provenance_cannot_masquerade_as_observed():
+    FakeClient.responses = [response(200, {"status": "accepted"})]
+    await world_model.post_finding(**finding_args(provenance="replayed", replayed=True))
+    body = FakeClient.requests[0][2]
+    assert body["replayed"] is True
+    assert body["payload"]["provenance"] == "replayed"
 
 
 @pytest.mark.asyncio