Merge remote-tracking branch 'origin/main' into codex/sre-typed-automation-20260715

apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py

@@ -40,6 +40,7 @@ logger = structlog.get_logger(__name__)
 Recorder = Callable[..., Awaitable[bool]]
 EvidenceCollector = Callable[..., Awaitable[EvidenceSnapshot | None]]
 EvidenceVerifier = Callable[..., Awaitable[bool]]
+IncidentBinder = Callable[..., Awaitable[bool]]
 
 _BACKFILL_DECISION_PATH = "repair_candidate_controlled_queue"
 _BACKFILL_REASON = (
@@ -613,6 +614,163 @@ def _wazuh_ingress_incident(
     }
 
 
+def _public_safe_wazuh_scheduled_incident_record(
+    incident: Mapping[str, Any],
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    """Build the bounded durable Incident row shared by the Wazuh scheduler."""
+
+    incident_id = str(incident.get("incident_id") or "").strip()
+    if not incident_id.startswith("IWZ-") or len(incident_id) > 30:
+        raise ValueError("invalid_wazuh_scheduler_incident_id")
+    if str(incident.get("project_id") or project_id) != project_id:
+        raise ValueError("wazuh_scheduler_project_mismatch")
+
+    alertname = str(incident.get("alertname") or "")[:100]
+    ingress_lane = alertname == "WazuhAlertmanagerIntegrationConvergence"
+    severity = {
+        "critical": "P0",
+        "high": "P1",
+        "medium": "P2",
+        "warning": "P2",
+        "low": "P3",
+    }.get(str(incident.get("severity") or "").lower(), "P2")
+    signal_input = next(
+        (
+            value
+            for value in incident.get("signals") or []
+            if isinstance(value, Mapping)
+        ),
+        {},
+    )
+    raw_labels = signal_input.get("labels")
+    raw_annotations = signal_input.get("annotations")
+    label_keys = {
+        "alertname",
+        "service",
+        "component",
+        "tool",
+        "asset_alias",
+    }
+    labels = {
+        str(key): str(value)[:120]
+        for key, value in (
+            raw_labels.items() if isinstance(raw_labels, Mapping) else []
+        )
+        if str(key) in label_keys and str(value).strip()
+    }
+    annotations = (
+        {"summary": str(raw_annotations.get("summary") or "")[:240]}
+        if isinstance(raw_annotations, Mapping)
+        else {}
+    )
+    observed_at = now_taipei()
+    signal = {
+        "signal_id": uuid5(
+            NAMESPACE_URL,
+            f"awoooi:iwooos:wazuh-scheduled-signal:{incident_id}",
+        ).hex[:8],
+        "alert_name": alertname,
+        "severity": severity,
+        "source": "sensor-agent",
+        "fired_at": observed_at.isoformat(),
+        "resolved_at": None,
+        "labels": labels,
+        "annotations": annotations,
+        "fingerprint": uuid5(
+            NAMESPACE_URL,
+            f"awoooi:iwooos:wazuh-scheduled-fingerprint:{incident_id}",
+        ).hex,
+    }
+    return {
+        "incident_id": incident_id,
+        "project_id": project_id,
+        "status": "INVESTIGATING",
+        "severity": severity,
+        "signals": json.dumps([signal], ensure_ascii=False),
+        "affected_services": json.dumps(
+            [
+                str(value)[:120]
+                for value in incident.get("affected_services") or []
+                if str(value).strip()
+            ][:20],
+            ensure_ascii=False,
+        ),
+        "proposal_ids": "[]",
+        "alertname": alertname,
+        "notification_type": "TYPE-5S" if ingress_lane else "TYPE-2",
+        "alert_category": "security",
+        "created_at": observed_at,
+        "updated_at": observed_at,
+    }
+
+
+async def ensure_iwooos_wazuh_scheduled_incident(
+    *,
+    incident: Mapping[str, Any],
+    project_id: str,
+) -> bool:
+    """Idempotently bind a scheduled Wazuh run to a durable Incident row."""
+
+    record = _public_safe_wazuh_scheduled_incident_record(
+        incident,
+        project_id=project_id,
+    )
+    async with get_db_context(project_id) as db:
+        await db.execute(
+            text("""
+                INSERT INTO incidents (
+                    incident_id, project_id, status, severity,
+                    signals, affected_services, proposal_ids,
+                    alertname, notification_type, alert_category,
+                    created_at, updated_at
+                ) VALUES (
+                    :incident_id, :project_id, :status, :severity,
+                    CAST(:signals AS json),
+                    CAST(:affected_services AS json),
+                    CAST(:proposal_ids AS json),
+                    :alertname, :notification_type, :alert_category,
+                    :created_at, :updated_at
+                )
+                ON CONFLICT (incident_id) DO NOTHING
+            """),
+            record,
+        )
+        readback = await db.execute(
+            text("""
+                SELECT
+                    incident_id,
+                    project_id,
+                    status::text AS status,
+                    alertname,
+                    notification_type,
+                    alert_category,
+                    json_array_length(signals) AS signal_count,
+                    json_array_length(affected_services) AS affected_service_count
+                FROM incidents
+                WHERE incident_id = :incident_id
+                  AND project_id = :project_id
+                LIMIT 1
+            """),
+            {
+                "incident_id": record["incident_id"],
+                "project_id": record["project_id"],
+            },
+        )
+        row = readback.mappings().first()
+    return bool(
+        row
+        and str(row.get("status") or "").upper()
+        in {"INVESTIGATING", "MITIGATING"}
+        and row.get("alertname") == record["alertname"]
+        and row.get("notification_type") == record["notification_type"]
+        and row.get("alert_category") == "security"
+        and int(row.get("signal_count") or 0) > 0
+        and int(row.get("affected_service_count") or 0) > 0
+    )
+
+
 async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
     *,
     project_id: str = "awoooi",
@@ -630,6 +788,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
         "scheduled Wazuh manager posture readback entered the allowlisted "
         "check-mode and independent verifier chain"
     ),
+    _incident_binder: IncidentBinder | None = None,
 ) -> dict[str, Any]:
     """Enqueue one fresh, no-write Wazuh manager posture executor run."""
 
@@ -835,6 +994,32 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
             bucket_ref=bucket.strftime("%Y%m%d%H"),
             attempt_ref=retry_attempt_ref,
         )
+        if _incident_binder is not None:
+            try:
+                incident_bound = await _incident_binder(
+                    incident=incident,
+                    project_id=project_id,
+                )
+            except Exception as exc:
+                incident_bound = False
+                logger.warning(
+                    "iwooos_wazuh_scheduled_incident_bind_failed",
+                    incident_id=incident["incident_id"],
+                    error_type=type(exc).__name__,
+                )
+            if not incident_bound:
+                return {
+                    "status": "blocked_scheduled_incident_write_not_acknowledged",
+                    "queued": 0,
+                    "existing": 1 if existing_row else 0,
+                    "evidence_ready": 0,
+                    "automation_run_id": automation_run_id,
+                    "work_item_id": _work_item_id,
+                    "retry_reason": retry_reason,
+                    "active_blockers": [
+                        "scheduled_incident_write_or_readback_not_acknowledged"
+                    ],
+                }
         try:
             snapshot = await evidence_collector(
                 incident=incident,
@@ -959,6 +1144,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
     recorder: Recorder = record_ansible_decision_audit,
     evidence_collector: EvidenceCollector = collect_ansible_candidate_pre_decision_evidence,
     evidence_verifier: EvidenceVerifier = verify_ansible_candidate_pre_decision_evidence,
+    incident_binder: IncidentBinder = ensure_iwooos_wazuh_scheduled_incident,
 ) -> dict[str, Any]:
     """Enqueue idempotent Wazuh event-ingress convergence before posture work."""
 
@@ -978,6 +1164,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
             "scheduled Wazuh alert ingress convergence entered the allowlisted "
             "check-mode, bounded apply, rollback, and independent verifier chain"
         ),
+        _incident_binder=incident_binder,
     )
 
 
@@ -1237,7 +1424,9 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None:
             ingress_result = (
                 await enqueue_iwooos_wazuh_alertmanager_integration_candidate_once()
             )
-            posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once()
+            posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once(
+                _incident_binder=ensure_iwooos_wazuh_scheduled_incident,
+            )
             result = await enqueue_missing_ansible_candidates_once(
                 limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT,
                 window_hours=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS,

apps/api/src/services/awooop_ansible_audit_service.py

@@ -542,7 +542,7 @@ _CATALOG: tuple[dict[str, Any], ...] = (
         "auto_apply_enabled": True,
         "approval_required": False,
         "risk_level": "high",
-        "catalog_revision": "2026-07-15-wazuh-alert-ingress-v1",
+        "catalog_revision": "2026-07-15-wazuh-alert-ingress-v2",
     },
     {
         "catalog_id": "ansible:110-host-pressure-readonly",