diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py
index 826a9d398..f9abf4293 100644
--- a/apps/api/src/jobs/asset_scanner_job.py
+++ b/apps/api/src/jobs/asset_scanner_job.py
@@ -102,6 +102,13 @@ _AI_PROGRAM_SCOPE_ASSET_TYPES = {
"data_and_backup": "backup_target",
"observability_and_security": "monitoring_target",
}
+_AI_PROGRAM_RUNTIME_IDENTITY_BINDINGS = {
+ "awoooi:observability_and_security:wazuh": {
+ "canonical_asset_ids": ["service:wazuh-manager"],
+ "identity_kinds": ["manager", "agent", "host", "service", "event"],
+ "source_truth_refs": ["ops/config/service-registry.yaml"],
+ }
+}
# 7 個自動化覆蓋維度 (ADR-090 §3.5)
_COVERAGE_DIMENSIONS = (
@@ -1597,6 +1604,9 @@ def _build_ai_program_catalog_assets(
canonical_ids: set[str] = set()
for declared in declared_assets:
canonical_id = str(declared.get("canonical_id") or "").strip()
+ runtime_identity_binding = _AI_PROGRAM_RUNTIME_IDENTITY_BINDINGS.get(
+ canonical_id
+ )
declared_asset_id = str(declared.get("declared_asset_id") or "").strip()
scope_id = str(declared.get("scope_id") or "").strip()
owner_lane = str(declared.get("owner_lane") or "").strip()
@@ -1605,6 +1615,15 @@ def _build_ai_program_catalog_assets(
for ref in declared.get("source_truth_refs") or []
if str(ref).strip()
]
+ if runtime_identity_binding:
+ source_truth_refs = list(
+ dict.fromkeys(
+ [
+ *source_truth_refs,
+ *runtime_identity_binding["source_truth_refs"],
+ ]
+ )
+ )
asset_type = _AI_PROGRAM_SCOPE_ASSET_TYPES.get(scope_id)
if (
not canonical_id
@@ -1637,6 +1656,21 @@ def _build_ai_program_catalog_assets(
"source_truth_refs": source_truth_refs,
"source_truth_ref_count": len(source_truth_refs),
"projection_authority": "ai_automation_program_ledger",
+ **(
+ {
+ "runtime_canonical_asset_ids": (
+ runtime_identity_binding["canonical_asset_ids"]
+ ),
+ "runtime_identity_kinds": (
+ runtime_identity_binding["identity_kinds"]
+ ),
+ "runtime_identity_binding_authority": (
+ "service_registry_and_controlled_runtime_v1"
+ ),
+ }
+ if runtime_identity_binding
+ else {}
+ ),
"raw_source_content_persisted": False,
"secret_value_read": False,
},
diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
index 3b9bbc302..c17cbe7b9 100644
--- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
+++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
@@ -64,6 +64,13 @@ def _error_backoff_seconds() -> float:
_WAZUH_POSTURE_CATALOG_ID = "ansible:wazuh-manager-posture-readback"
_WAZUH_POSTURE_WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE"
_WAZUH_POSTURE_PUBLIC_ASSET_ALIAS = "security_manager_primary"
+_WAZUH_INCOMPLETE_RUNTIME_RETRY_REASONS = frozenset(
+ {
+ "controlled_apply_failed",
+ "post_apply_learning_incomplete",
+ "recipient_visible_transition_missing",
+ }
+)
_WAZUH_INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration"
_WAZUH_INGRESS_WORK_ITEM_ID = "P0-03-WAZUH-ALERT-INGRESS"
_WAZUH_CONTROLLED_CAPABILITY_PACKET_OPERATION_TYPE = (
@@ -1183,6 +1190,9 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
capability_packet.input
#>> '{gate_lifecycle,expires_at}'
AS latest_controlled_capability_packet_expires_at,
+ telegram.send_status AS latest_telegram_send_status,
+ telegram.provider_message_id
+ AS latest_telegram_provider_message_id,
EXISTS (
SELECT 1
FROM incident_evidence evidence
@@ -1247,6 +1257,26 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
ORDER BY packet.created_at DESC
LIMIT 1
) capability_packet ON TRUE
+ LEFT JOIN LATERAL (
+ SELECT outbound.send_status,
+ outbound.provider_message_id
+ FROM awooop_outbound_message outbound
+ WHERE outbound.project_id = :project_id
+ AND outbound.channel_type = 'telegram'
+ AND coalesce(
+ outbound.source_envelope
+ ->> 'automation_run_id',
+ outbound.source_envelope
+ #>> '{callback_reply,automation_run_id}',
+ outbound.source_envelope
+ #>> '{source_refs,automation_run_ids,0}'
+ ) = candidate.op_id::text
+ AND outbound.source_envelope
+ #>> '{callback_reply,action}'
+ = 'controlled_apply_result'
+ ORDER BY outbound.queued_at DESC
+ LIMIT 1
+ ) telegram ON TRUE
WHERE candidate.operation_type = 'ansible_candidate_matched'
AND candidate.input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb)
AND candidate.input ->> 'project_id' = :project_id
@@ -1296,7 +1326,12 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
existing_row.get("latest_check_status") or ""
) if existing_row else ""
retry_attempt_ref: str | None = None
- retry_reason = _wazuh_posture_retry_reason(existing_row)
+ retry_reason = _wazuh_posture_retry_reason(
+ existing_row,
+ require_recipient_visible_transition=(
+ _catalog_id == _WAZUH_POSTURE_CATALOG_ID
+ ),
+ )
capability_packet_pending = _controlled_capability_packet_is_active(
existing_row,
observed_now=observed_now,
@@ -1375,6 +1410,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
retry_attempt_ref = f"{deploy_ref}-apply"
elif retry_reason == "post_apply_learning_incomplete":
retry_attempt_ref = f"{deploy_ref}-learning"
+ elif retry_reason == "recipient_visible_transition_missing":
+ retry_attempt_ref = f"{deploy_ref}-receipt"
elif latest_check_status == "success":
retry_attempt_ref = f"{deploy_ref}-context"
retry_reason = "predecision_context"
@@ -1423,10 +1460,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
- "retry_of_incomplete_runtime_closure": retry_reason in {
- "controlled_apply_failed",
- "post_apply_learning_incomplete",
- },
+ "retry_of_incomplete_runtime_closure": retry_reason
+ in _WAZUH_INCOMPLETE_RUNTIME_RETRY_REASONS,
"active_blockers": [
"bounded_retry_already_attempted_for_deploy"
if retry_reason != "predecision_context"
@@ -1506,10 +1541,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
- "retry_of_incomplete_runtime_closure": retry_reason in {
- "controlled_apply_failed",
- "post_apply_learning_incomplete",
- },
+ "retry_of_incomplete_runtime_closure": retry_reason
+ in _WAZUH_INCOMPLETE_RUNTIME_RETRY_REASONS,
"active_blockers": [
"predecision_mcp_or_log_evidence_missing"
],
@@ -1583,10 +1616,8 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
- "retry_of_incomplete_runtime_closure": retry_reason in {
- "controlled_apply_failed",
- "post_apply_learning_incomplete",
- },
+ "retry_of_incomplete_runtime_closure": retry_reason
+ in _WAZUH_INCOMPLETE_RUNTIME_RETRY_REASONS,
"active_blockers": (
[]
if queued or existing_row
@@ -1693,6 +1724,8 @@ async def enqueue_scheduled_iwooos_wazuh_candidates_once(
def _wazuh_posture_retry_reason(
existing_row: Mapping[str, Any] | None,
+ *,
+ require_recipient_visible_transition: bool = False,
) -> str | None:
if not existing_row:
return None
@@ -1705,6 +1738,16 @@ def _wazuh_posture_retry_reason(
return "controlled_apply_failed"
if apply_status == "success" and learning_status != "success":
return "post_apply_learning_incomplete"
+ if (
+ require_recipient_visible_transition
+ and apply_status == "success"
+ and learning_status == "success"
+ and not (
+ existing_row.get("latest_telegram_send_status") == "sent"
+ and existing_row.get("latest_telegram_provider_message_id")
+ )
+ ):
+ return "recipient_visible_transition_missing"
return None
diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py
index 0fd936bbd..4c17a46bf 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -122,6 +122,9 @@ _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE = "remediation_executed"
_WAZUH_ALERTMANAGER_INTEGRATION_CATALOG_ID = (
"ansible:wazuh-alertmanager-integration"
)
+_WAZUH_MANAGER_POSTURE_CATALOG_ID = (
+ "ansible:wazuh-manager-posture-readback"
+)
_WAZUH_PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING = (
"wazuh_privileged_convergence_capability_missing"
)
@@ -3400,6 +3403,7 @@ async def _read_verified_apply_closure_prerequisites(
"auto_repair_execution_receipt": row.get("auto_repair_receipt") is True,
"km_playbook_writeback": row.get("km_writeback_receipt") is True,
"rag_writeback": "rag_writeback" in stage_ids,
+ "mcp_context": "mcp_context" in stage_ids,
"playbook_trust": (
row.get("playbook_trust_receipt") is True
and row.get("playbook_readback") is True
@@ -6730,7 +6734,11 @@ async def _send_controlled_apply_telegram_receipt(
or provider_delivery
)
if no_write_observation and verified_success:
- effective_provider_delivery = "shadow_only"
+ effective_provider_delivery = (
+ "state_transition"
+ if claim.catalog_id == _WAZUH_MANAGER_POSTURE_CATALOG_ID
+ else "shadow_only"
+ )
receipt_readback = dict(closure_receipts or {})
response = await get_telegram_gateway().send_controlled_apply_result_receipt(
automation_run_id=str(
diff --git a/apps/api/src/services/iwooos_security_asset_control_plane.py b/apps/api/src/services/iwooos_security_asset_control_plane.py
index aaac11fd1..7ff55f6d8 100644
--- a/apps/api/src/services/iwooos_security_asset_control_plane.py
+++ b/apps/api/src/services/iwooos_security_asset_control_plane.py
@@ -19,6 +19,9 @@ from src.services.ai_automation_runtime_contract import (
AI_AUTOMATION_REQUIRED_LOOP_STAGES,
AI_AUTOMATION_RUNTIME_CONTRACT_SCHEMA_VERSION,
)
+from src.services.iwooos_wazuh_controlled_executor_runtime import (
+ load_iwooos_wazuh_controlled_executor_runtime_lanes,
+)
logger = structlog.get_logger(__name__)
@@ -32,6 +35,10 @@ _DISCOVERY_FRESHNESS_SECONDS = 2 * 60 * 60
_WINDOW_HOURS = 24
_STRICT_RUNTIME_SCHEMA_VERSION = "ai_agent_strict_runtime_completion_v2"
_STRICT_RUNTIME_SOURCE_ID = "autonomous_strict_runtime"
+_WAZUH_RUNTIME_SOURCE_ID = "wazuh_controlled_runtime"
+_WAZUH_PROGRAM_CANONICAL_ID = "awoooi:observability_and_security:wazuh"
+_WAZUH_ASSET_CANONICAL_ID = "service:wazuh-manager"
+_SOURCE_COUNT = 12
_WORK_ITEM_ORDER = {
"source_health": 0,
@@ -533,6 +540,7 @@ def _build_security_program_domains(
certificate_live_unverified_count: int,
certificate_live_healthy_percent: int,
certificate_live_unhealthy_count: int,
+ wazuh_soc_slice: Mapping[str, Any],
) -> list[dict[str, Any]]:
scope_by_id = {row["scope_id"]: row for row in asset_scopes}
function_by_id = {row["function_id"]: row for row in control_functions}
@@ -614,9 +622,12 @@ def _build_security_program_domains(
(
"endpoint_xdr_wazuh",
"端點、XDR 與 Wazuh",
- 0,
- 0,
- "wazuh_runtime_identity_not_joined_to_asset_graph",
+ int(wazuh_soc_slice.get("domain_controlled_percent") or 0),
+ int(wazuh_soc_slice.get("domain_evidence_count") or 0),
+ str(
+ wazuh_soc_slice.get("domain_gap_code")
+ or "wazuh_runtime_identity_not_joined_to_asset_graph"
+ ),
),
(
"network_dns_tls",
@@ -920,6 +931,259 @@ def _strict_runtime_completion_projection(row: Any) -> dict[str, Any]:
}
+def _build_wazuh_soc_slice(
+ *,
+ asset_row: Any,
+ runtime_row: Any,
+) -> dict[str, Any]:
+ """Join public-safe Wazuh runtime facets to one canonical graph asset."""
+
+ runtime = runtime_row if isinstance(runtime_row, Mapping) else {}
+ posture = runtime.get("posture")
+ ingress = runtime.get("ingress")
+ posture = posture if isinstance(posture, Mapping) else {}
+ ingress = ingress if isinstance(ingress, Mapping) else {}
+ posture_summary = posture.get("summary")
+ posture_summary = (
+ posture_summary if isinstance(posture_summary, Mapping) else {}
+ )
+ posture_trace = posture.get("trace")
+ posture_trace = posture_trace if isinstance(posture_trace, Mapping) else {}
+ posture_identity = posture.get("asset_identity")
+ posture_identity = (
+ posture_identity if isinstance(posture_identity, Mapping) else {}
+ )
+ posture_incident = posture.get("incident")
+ posture_incident = (
+ posture_incident if isinstance(posture_incident, Mapping) else {}
+ )
+ posture_verifier = posture.get("verifier")
+ posture_verifier = (
+ posture_verifier if isinstance(posture_verifier, Mapping) else {}
+ )
+ posture_recipient = posture.get("recipient_receipt")
+ posture_recipient = (
+ posture_recipient if isinstance(posture_recipient, Mapping) else {}
+ )
+ ingress_boundaries = ingress.get("boundaries")
+ ingress_boundaries = (
+ ingress_boundaries if isinstance(ingress_boundaries, Mapping) else {}
+ )
+
+ canonical_asset_count = int(
+ _value(asset_row, "canonical_asset_count", 0) or 0
+ )
+ owner_resolved_asset_count = int(
+ _value(asset_row, "owner_resolved_asset_count", 0) or 0
+ )
+ owner_verified_asset_count = int(
+ _value(asset_row, "owner_verified_asset_count", 0) or 0
+ )
+ canonical_asset_ref = str(
+ _value(asset_row, "canonical_asset_ref", "") or ""
+ )
+ runtime_asset_ref = str(
+ posture_identity.get("canonical_asset_ref") or ""
+ )
+ graph_asset_ready = bool(
+ canonical_asset_count == 1
+ and canonical_asset_ref == _WAZUH_ASSET_CANONICAL_ID
+ )
+ asset_graph_joined = bool(
+ graph_asset_ready and runtime_asset_ref == canonical_asset_ref
+ )
+ owner_ready = (
+ owner_resolved_asset_count == 1 and owner_verified_asset_count == 1
+ )
+ identity_kind_count = int(
+ posture_identity.get("linked_identity_kind_count") or 0
+ )
+ required_identity_kind_count = int(
+ posture_identity.get("required_identity_kind_count") or 5
+ )
+ identity_ready = bool(
+ posture_identity.get("all_required_identity_kinds_linked") is True
+ and identity_kind_count == required_identity_kind_count
+ and asset_graph_joined
+ and posture_identity.get("raw_identity_returned") is False
+ )
+ telemetry_ready = bool(
+ graph_asset_ready
+ and posture_trace.get("source_recurrence_verified") is True
+ and posture_summary.get("service_log_evidence_receipt_count") == 1
+ and identity_ready
+ )
+ verifier_ready = bool(
+ posture_summary.get("independent_post_verifier_passed_count") == 1
+ and posture_verifier.get("status") == "passed"
+ and str(posture_verifier.get("receipt_ref") or "")
+ )
+ same_run_closed = bool(
+ posture_summary.get("runtime_closed_count") == 1
+ and posture_incident.get("ledger_closure_verified") is True
+ and posture_incident.get("status") == "resolved"
+ )
+ recipient_visible = bool(
+ posture_summary.get("telegram_provider_send_count") == 1
+ and posture_recipient.get("status") == "recipient_visible"
+ and str(posture_recipient.get("receipt_ref") or "")
+ and str(posture_recipient.get("provider_message_ref") or "")
+ )
+ receipt_suppressed = bool(
+ posture_summary.get("telegram_notification_suppressed_count") == 1
+ )
+ protected_asset_count = int(
+ graph_asset_ready
+ and owner_ready
+ and telemetry_ready
+ and verifier_ready
+ and same_run_closed
+ )
+ critical_break_glass_required = bool(
+ ingress_boundaries.get("critical_break_glass_required") is True
+ )
+
+ remaining_gap_codes: list[str] = []
+ if not graph_asset_ready:
+ remaining_gap_codes.append("canonical_wazuh_asset_missing")
+ elif not asset_graph_joined:
+ remaining_gap_codes.append("wazuh_runtime_asset_graph_join_missing")
+ if not owner_ready:
+ remaining_gap_codes.append("canonical_wazuh_owner_missing")
+ if not identity_ready:
+ remaining_gap_codes.append("wazuh_runtime_identity_chain_incomplete")
+ if not telemetry_ready:
+ remaining_gap_codes.append("wazuh_runtime_telemetry_missing")
+ if not same_run_closed:
+ remaining_gap_codes.append("wazuh_posture_same_run_closure_open")
+ if not recipient_visible:
+ remaining_gap_codes.append("wazuh_recipient_visible_receipt_missing")
+ if critical_break_glass_required:
+ remaining_gap_codes.append("wazuh_critical_ingress_break_glass_pending")
+ remaining_gap_codes.append("global_wazuh_asset_coverage_not_proven")
+
+ evidence_checks = (
+ graph_asset_ready,
+ owner_ready,
+ identity_ready,
+ telemetry_ready,
+ verifier_ready,
+ same_run_closed,
+ recipient_visible,
+ )
+ return {
+ "schema_version": "iwooos_wazuh_ai_soc_slice_v1",
+ "status": (
+ "same_run_closed_recipient_visible"
+ if all(evidence_checks)
+ else "partial"
+ if any(evidence_checks)
+ else "not_evidenced"
+ ),
+ "canonical_asset_count": canonical_asset_count,
+ "canonical_asset_ref": canonical_asset_ref or None,
+ "protected_asset_count": protected_asset_count,
+ "protected_asset_ref": (
+ canonical_asset_ref if protected_asset_count == 1 else None
+ ),
+ "asset_graph_joined": asset_graph_joined,
+ "owner_resolved_asset_count": owner_resolved_asset_count,
+ "owner_verified_asset_count": owner_verified_asset_count,
+ "telemetry_covered_asset_count": 1 if telemetry_ready else 0,
+ "identity_kind_count": identity_kind_count,
+ "required_identity_kind_count": required_identity_kind_count,
+ "identity_kinds": list(
+ posture_identity.get("linked_identity_kinds") or []
+ ),
+ "incident_ref": posture_incident.get("incident_ref"),
+ "incident_status": "resolved" if same_run_closed else "open",
+ "verifier_status": "passed" if verifier_ready else "missing",
+ "verifier_receipt_ref": (
+ str(posture_verifier.get("receipt_ref") or "") or None
+ ),
+ "recipient_receipt_status": (
+ "recipient_visible"
+ if recipient_visible
+ else "suppressed_unchanged"
+ if receipt_suppressed
+ else "missing"
+ ),
+ "recipient_receipt_ref": (
+ str(posture_recipient.get("receipt_ref") or "") or None
+ ),
+ "recipient_provider_message_ref": (
+ str(posture_recipient.get("provider_message_ref") or "") or None
+ ),
+ "same_run_present_stage_count": int(
+ posture_summary.get("same_run_present_stage_count") or 0
+ ),
+ "same_run_required_stage_count": int(
+ posture_summary.get("same_run_required_stage_count")
+ or len(AI_AUTOMATION_REQUIRED_LOOP_STAGES)
+ ),
+ "same_run_closed": same_run_closed,
+ "critical_break_glass_required": critical_break_glass_required,
+ "remaining_gap_codes": remaining_gap_codes,
+ "domain_controlled_percent": min(
+ 70,
+ round(sum(evidence_checks) / len(evidence_checks) * 70),
+ ),
+ "domain_evidence_count": sum(evidence_checks),
+ "domain_gap_code": (
+ "wazuh_bounded_slice_closed_global_coverage_open"
+ if all(evidence_checks)
+ else remaining_gap_codes[0]
+ ),
+ "coverage_overlay_applied_count": 0,
+ "raw_identity_returned": False,
+ "raw_event_payload_returned": False,
+ "secret_value_read": False,
+ }
+
+
+def _overlay_wazuh_runtime_coverage(
+ rows: Iterable[Any],
+ *,
+ asset_row: Any,
+ wazuh_slice: dict[str, Any],
+) -> tuple[list[dict[str, Any]], int]:
+ """Replace only the exact Wazuh asset's unknown telemetry with live evidence."""
+
+ counts: dict[tuple[str, str], int] = {}
+ for row in rows:
+ key = (
+ str(_value(row, "dimension", "")),
+ str(_value(row, "status", "")),
+ )
+ counts[key] = counts.get(key, 0) + int(_value(row, "cnt", 0) or 0)
+ if wazuh_slice["telemetry_covered_asset_count"] != 1:
+ return [
+ {"dimension": dimension, "status": status, "cnt": count}
+ for (dimension, status), count in sorted(counts.items())
+ if count > 0
+ ], 0
+
+ applied = 0
+ for dimension, status_field in (
+ ("auto_monitoring", "monitoring_coverage_status"),
+ ("auto_alerting", "alerting_coverage_status"),
+ ):
+ if str(_value(asset_row, status_field, "")) != "unknown":
+ continue
+ unknown_key = (dimension, "unknown")
+ if counts.get(unknown_key, 0) <= 0:
+ continue
+ counts[unknown_key] -= 1
+ green_key = (dimension, "green")
+ counts[green_key] = counts.get(green_key, 0) + 1
+ applied += 1
+ return [
+ {"dimension": dimension, "status": status, "cnt": count}
+ for (dimension, status), count in sorted(counts.items())
+ if count > 0
+ ], applied
+
+
def build_iwooos_security_asset_control_plane(
*,
discovery_row: Any,
@@ -932,6 +1196,8 @@ def build_iwooos_security_asset_control_plane(
siem_row: Any,
audit_row: Any,
strict_runtime_row: Any = None,
+ wazuh_asset_row: Any = None,
+ wazuh_runtime_row: Any = None,
source_health: Iterable[dict[str, Any]] | None = None,
generated_at: datetime | None = None,
) -> dict[str, Any]:
@@ -942,6 +1208,18 @@ def build_iwooos_security_asset_control_plane(
compliance_source = list(compliance_rows)
automation_source = list(automation_rows)
source_health_rows = list(source_health or [])
+ wazuh_soc_slice = _build_wazuh_soc_slice(
+ asset_row=wazuh_asset_row,
+ runtime_row=wazuh_runtime_row,
+ )
+ coverage_source, coverage_overlay_count = _overlay_wazuh_runtime_coverage(
+ coverage_source,
+ asset_row=wazuh_asset_row,
+ wazuh_slice=wazuh_soc_slice,
+ )
+ wazuh_soc_slice["coverage_overlay_applied_count"] = (
+ coverage_overlay_count
+ )
unavailable_sources = [
row for row in source_health_rows if row.get("status") != "ready"
]
@@ -1231,6 +1509,7 @@ def build_iwooos_security_asset_control_plane(
by_type.get("certificate", 0),
),
certificate_live_unhealthy_count=certificate_live_unhealthy_count,
+ wazuh_soc_slice=wazuh_soc_slice,
)
work_items: list[dict[str, Any]] = []
@@ -1538,6 +1817,7 @@ def build_iwooos_security_asset_control_plane(
"strict_runtime_closure_percent": strict_runtime_closure_percent,
"dimensions": completion_dimensions,
},
+ "wazuh_soc_slice": wazuh_soc_slice,
"siem": {**siem, "pipeline": siem_pipeline},
"ai_automation": {
"window_hours": _WINDOW_HOURS,
@@ -1616,6 +1896,7 @@ def build_iwooos_security_asset_control_plane(
"live_scan_triggered": False,
"runtime_action_triggered": False,
"completion_requires_same_run_receipts": True,
+ "wazuh_slice_is_global_coverage_claim": False,
},
}
@@ -1632,9 +1913,9 @@ def build_unavailable_iwooos_security_asset_control_plane(
"source_status": "live_database_unavailable",
"reason_code": reason_code,
"summary": {
- "source_count": 10,
+ "source_count": _SOURCE_COUNT,
"ready_source_count": 0,
- "unavailable_source_count": 10,
+ "unavailable_source_count": _SOURCE_COUNT,
"managed_asset_count": 0,
"expected_asset_type_count": len(_ASSET_TYPES),
"present_asset_type_count": 0,
@@ -1761,6 +2042,10 @@ def build_unavailable_iwooos_security_asset_control_plane(
)
],
},
+ "wazuh_soc_slice": _build_wazuh_soc_slice(
+ asset_row=None,
+ runtime_row=None,
+ ),
"siem": {
"pipeline": [
{
@@ -1845,7 +2130,9 @@ def build_unavailable_iwooos_security_asset_control_plane(
"ai_decision_trace",
"siem_runtime",
"audit_runtime",
+ "wazuh_asset_graph",
_STRICT_RUNTIME_SOURCE_ID,
+ _WAZUH_RUNTIME_SOURCE_ID,
)
],
"work_items": [
@@ -1883,6 +2170,7 @@ def build_unavailable_iwooos_security_asset_control_plane(
"live_scan_triggered": False,
"runtime_action_triggered": False,
"completion_requires_same_run_receipts": True,
+ "wazuh_slice_is_global_coverage_claim": False,
},
}
@@ -2000,6 +2288,61 @@ async def _read_strict_runtime_source() -> tuple[dict[str, Any], dict[str, Any]]
}
+async def _read_wazuh_runtime_source(
+ *,
+ project_id: str,
+ limiter: asyncio.Semaphore,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+ """Read both durable Wazuh lanes under the aggregate concurrency budget."""
+
+ try:
+ async with limiter:
+ lanes = await asyncio.wait_for(
+ load_iwooos_wazuh_controlled_executor_runtime_lanes(
+ project_id=project_id
+ ),
+ timeout=_STRICT_RUNTIME_SOURCE_TIMEOUT_SECONDS,
+ )
+ if not isinstance(lanes, Mapping) or not {"ingress", "posture"} <= set(
+ lanes
+ ):
+ raise ValueError("wazuh_runtime_lane_contract_mismatch")
+ if any(
+ isinstance(lane, Mapping)
+ and str(lane.get("status") or "").startswith(
+ "degraded_runtime_receipt_read_unavailable"
+ )
+ for lane in lanes.values()
+ ):
+ return {}, {
+ "source_id": _WAZUH_RUNTIME_SOURCE_ID,
+ "status": "unavailable",
+ "reason_code": "wazuh_controlled_runtime_readback_unavailable",
+ }
+ return dict(lanes), {
+ "source_id": _WAZUH_RUNTIME_SOURCE_ID,
+ "status": "ready",
+ "reason_code": None,
+ }
+ except TimeoutError:
+ logger.warning("iwooos_security_asset_control_plane_wazuh_runtime_timeout")
+ return {}, {
+ "source_id": _WAZUH_RUNTIME_SOURCE_ID,
+ "status": "unavailable",
+ "reason_code": "wazuh_controlled_runtime_query_timeout",
+ }
+ except Exception as exc: # noqa: BLE001 - redact public source failures
+ logger.warning(
+ "iwooos_security_asset_control_plane_wazuh_runtime_unavailable",
+ error_type=type(exc).__name__,
+ )
+ return {}, {
+ "source_id": _WAZUH_RUNTIME_SOURCE_ID,
+ "status": "unavailable",
+ "reason_code": "wazuh_controlled_runtime_query_failed",
+ }
+
+
class IwoooSSecurityAssetControlPlaneService:
"""Read aggregate production security state with a bounded DB timeout."""
@@ -2139,6 +2482,77 @@ class IwoooSSecurityAssetControlPlaneService:
result_mode="all",
default=[],
),
+ _read_public_source(
+ project_id="awoooi",
+ limiter=source_limiter,
+ source_id="wazuh_asset_graph",
+ statement=_sql(
+ f"""
+ WITH latest_run AS (
+ SELECT run_id
+ FROM asset_discovery_run
+ WHERE status = 'success'
+ ORDER BY ended_at DESC
+ LIMIT 1
+ ), wazuh_asset AS (
+ SELECT asset_id, owner_team, metadata
+ FROM asset_inventory
+ WHERE lifecycle_state = 'active'
+ AND metadata ->> 'program_canonical_id'
+ = '{_WAZUH_PROGRAM_CANONICAL_ID}'
+ AND metadata -> 'runtime_canonical_asset_ids'
+ @> '["{_WAZUH_ASSET_CANONICAL_ID}"]'::jsonb
+ )
+ SELECT
+ (SELECT count(*) FROM wazuh_asset)
+ AS canonical_asset_count,
+ (SELECT count(*) FROM wazuh_asset
+ WHERE owner_team IS NOT NULL
+ AND btrim(owner_team) <> ''
+ ) AS owner_resolved_asset_count,
+ (SELECT count(*) FROM wazuh_asset
+ WHERE lower(btrim(owner_team)) = 'aisoc'
+ ) AS owner_verified_asset_count,
+ (SELECT metadata #>>
+ '{{runtime_canonical_asset_ids,0}}'
+ FROM wazuh_asset
+ LIMIT 1)
+ AS canonical_asset_ref,
+ coalesce(
+ (
+ SELECT snapshot.coverage_status
+ FROM asset_coverage_snapshot snapshot
+ JOIN wazuh_asset asset
+ ON asset.asset_id = snapshot.asset_id
+ JOIN latest_run
+ ON latest_run.run_id = snapshot.run_id
+ WHERE snapshot.dimension = 'auto_monitoring'
+ ORDER BY snapshot.detected_at DESC,
+ snapshot.snapshot_id DESC
+ LIMIT 1
+ ),
+ 'unknown'
+ ) AS monitoring_coverage_status,
+ coalesce(
+ (
+ SELECT snapshot.coverage_status
+ FROM asset_coverage_snapshot snapshot
+ JOIN wazuh_asset asset
+ ON asset.asset_id = snapshot.asset_id
+ JOIN latest_run
+ ON latest_run.run_id = snapshot.run_id
+ WHERE snapshot.dimension = 'auto_alerting'
+ ORDER BY snapshot.detected_at DESC,
+ snapshot.snapshot_id DESC
+ LIMIT 1
+ ),
+ 'unknown'
+ ) AS alerting_coverage_status
+ """
+ ),
+ result_mode="one",
+ default={},
+ ),
_read_public_source(
project_id="awoooi",
limiter=source_limiter,
@@ -2419,11 +2833,16 @@ class IwoooSSecurityAssetControlPlaneService:
default={},
),
_read_strict_runtime_source(),
+ _read_wazuh_runtime_source(
+ project_id="awoooi",
+ limiter=source_limiter,
+ ),
)
(
(discovery_row, discovery_health),
(inventory_rows, inventory_health),
+ (wazuh_asset_row, wazuh_asset_health),
(coverage_rows, coverage_health),
(relationship_row, relationship_health),
(compliance_rows, compliance_health),
@@ -2432,10 +2851,12 @@ class IwoooSSecurityAssetControlPlaneService:
(siem_row, siem_health),
(audit_row, audit_health),
(strict_runtime_row, strict_runtime_health),
+ (wazuh_runtime_row, wazuh_runtime_health),
) = source_results
source_health = [
discovery_health,
inventory_health,
+ wazuh_asset_health,
coverage_health,
relationship_health,
compliance_health,
@@ -2444,6 +2865,7 @@ class IwoooSSecurityAssetControlPlaneService:
siem_health,
audit_health,
strict_runtime_health,
+ wazuh_runtime_health,
]
return build_iwooos_security_asset_control_plane(
@@ -2457,6 +2879,8 @@ class IwoooSSecurityAssetControlPlaneService:
siem_row=siem_row,
audit_row=audit_row,
strict_runtime_row=strict_runtime_row,
+ wazuh_asset_row=wazuh_asset_row,
+ wazuh_runtime_row=wazuh_runtime_row,
source_health=source_health,
)
diff --git a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
index 0ac1858f1..00aba8e06 100644
--- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
@@ -48,6 +48,23 @@ _PUBLIC_SAFE_CHECK_FAILURE_MARKERS = frozenset(
}
)
+_RUNTIME_LANES = (
+ (
+ "ingress",
+ INGRESS_CATALOG_ID,
+ INGRESS_WORK_ITEM_ID,
+ INGRESS_PROPOSAL_SOURCE,
+ INGRESS_RUN_NAMESPACE,
+ ),
+ (
+ "posture",
+ CATALOG_ID,
+ WORK_ITEM_ID,
+ POSTURE_PROPOSAL_SOURCE,
+ POSTURE_RUN_NAMESPACE,
+ ),
+)
+
_LIVE_RUNTIME_SQL = """
WITH latest_candidate AS (
SELECT
@@ -339,7 +356,43 @@ _LIVE_RUNTIME_SQL = """
km.*,
learning.*,
auto_repair.*,
- telegram.*
+ telegram.*,
+ incident.status::text AS incident_status,
+ incident.resolved_at AS incident_resolved_at,
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,automation_run_id}'
+ AS incident_terminal_automation_run_id,
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,apply_op_id}'
+ AS incident_terminal_apply_op_id,
+ coalesce(
+ (
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,incident_resolved}'
+ )::boolean,
+ false
+ ) AS incident_terminal_resolved,
+ coalesce(
+ (
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,telegram_receipt_acknowledged}'
+ )::boolean,
+ false
+ ) AS incident_terminal_telegram_acknowledged,
+ coalesce(
+ (
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,repository_readback_verified}'
+ )::boolean,
+ false
+ ) AS incident_terminal_repository_readback_verified,
+ coalesce(
+ (
+ cast(incident.outcome AS jsonb) #>>
+ '{automation_terminal,durable_write_acknowledged}'
+ )::boolean,
+ false
+ ) AS incident_terminal_durable_write_acknowledged
FROM latest_candidate candidate
LEFT JOIN latest_check check_mode ON TRUE
LEFT JOIN latest_controlled_capability_packet
@@ -351,6 +404,9 @@ _LIVE_RUNTIME_SQL = """
LEFT JOIN latest_learning learning ON TRUE
LEFT JOIN latest_auto_repair auto_repair ON TRUE
LEFT JOIN latest_telegram telegram ON TRUE
+ LEFT JOIN incidents incident
+ ON incident.incident_id = candidate.incident_id
+ AND incident.project_id = :project_id
"""
@@ -359,11 +415,30 @@ async def load_latest_iwooos_wazuh_controlled_executor_runtime(
) -> dict[str, Any]:
"""Read the latest Wazuh posture executor chain from durable receipts."""
+ lanes = await load_iwooos_wazuh_controlled_executor_runtime_lanes(
+ project_id=project_id
+ )
+ ingress = lanes["ingress"]
+ if ingress["summary"]["dispatch_candidate_count"] > 0:
+ return ingress
+ return lanes["posture"]
+
+
+async def load_iwooos_wazuh_controlled_executor_runtime_lanes(
+ *, project_id: str = "awoooi"
+) -> dict[str, dict[str, Any]]:
+ """Read both Wazuh lanes so a blocked ingress cannot hide posture truth."""
+
try:
async with get_db_context(project_id) as db:
- row = None
- for catalog_id in (INGRESS_CATALOG_ID, CATALOG_ID):
- ingress_mode = catalog_id == INGRESS_CATALOG_ID
+ rows: dict[str, Mapping[str, Any] | None] = {}
+ for (
+ lane_id,
+ catalog_id,
+ work_item_id,
+ proposal_source,
+ run_namespace,
+ ) in _RUNTIME_LANES:
result = await db.execute(
text(_LIVE_RUNTIME_SQL),
{
@@ -371,30 +446,50 @@ async def load_latest_iwooos_wazuh_controlled_executor_runtime(
[{"catalog_id": catalog_id}]
),
"project_id": project_id,
- "work_item_id": (
- INGRESS_WORK_ITEM_ID if ingress_mode else WORK_ITEM_ID
- ),
- "proposal_source": (
- INGRESS_PROPOSAL_SOURCE
- if ingress_mode
- else POSTURE_PROPOSAL_SOURCE
- ),
- "run_namespace": (
- INGRESS_RUN_NAMESPACE
- if ingress_mode
- else POSTURE_RUN_NAMESPACE
- ),
+ "work_item_id": work_item_id,
+ "proposal_source": proposal_source,
+ "run_namespace": run_namespace,
},
)
- row = result.mappings().first()
- if row:
- break
+ rows[lane_id] = result.mappings().first()
except Exception as exc:
- return build_iwooos_wazuh_controlled_executor_runtime_readback(
- None,
- read_error_type=type(exc).__name__,
+ error_type = type(exc).__name__
+ return {
+ lane_id: build_iwooos_wazuh_controlled_executor_runtime_readback(
+ {
+ "catalog_id": catalog_id,
+ "work_item_id": work_item_id,
+ "proposal_source": proposal_source,
+ "run_namespace": run_namespace,
+ },
+ read_error_type=error_type,
+ )
+ for (
+ lane_id,
+ catalog_id,
+ work_item_id,
+ proposal_source,
+ run_namespace,
+ ) in _RUNTIME_LANES
+ }
+ return {
+ lane_id: build_iwooos_wazuh_controlled_executor_runtime_readback(
+ rows.get(lane_id)
+ or {
+ "catalog_id": catalog_id,
+ "work_item_id": work_item_id,
+ "proposal_source": proposal_source,
+ "run_namespace": run_namespace,
+ }
)
- return build_iwooos_wazuh_controlled_executor_runtime_readback(row)
+ for (
+ lane_id,
+ catalog_id,
+ work_item_id,
+ proposal_source,
+ run_namespace,
+ ) in _RUNTIME_LANES
+ }
def build_iwooos_wazuh_controlled_executor_runtime_readback(
@@ -618,6 +713,30 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
str(stage_id) for stage_id in data.get("stage_ids") or [] if stage_id
}
stage_ids = raw_stage_ids if runtime_stage_identity_verified else set()
+ runtime_identity_links = {
+ "manager": bool(
+ source_recurrence_verified and "service_log_evidence" in stage_ids
+ ),
+ "agent": source_recurrence_verified,
+ "host": bool(
+ candidate_identity_verified
+ and int(data.get("asset_scope_alias_count") or 0) > 0
+ ),
+ "service": bool(
+ source_recurrence_verified
+ and candidate_source_recurrence.get("canonical_asset_id")
+ == "service:wazuh-manager"
+ ),
+ "event": bool(
+ source_recurrence_verified
+ and str(candidate_source_recurrence.get("fingerprint") or "")
+ and str(candidate_source_recurrence.get("occurrence_id") or "")
+ and str(candidate_source_recurrence.get("identity_anchor") or "")
+ ),
+ }
+ canonical_asset_ref = str(
+ candidate_source_recurrence.get("canonical_asset_id") or ""
+ )
rag_ready = "rag_writeback" in stage_ids
trust_ready = bool(data.get("playbook_trust_acknowledged")) and (
"playbook_trust" in stage_ids
@@ -636,6 +755,19 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
and data.get("telegram_provider_delivery") == "shadow_only"
)
telegram_ready = telegram_provider_sent or telegram_notification_suppressed
+ incident_ledger_closure_verified = bool(
+ str(data.get("incident_status") or "").upper()
+ in {"RESOLVED", "CLOSED"}
+ and data.get("incident_resolved_at") is not None
+ and str(data.get("incident_terminal_automation_run_id") or "")
+ == candidate_op_id
+ and str(data.get("incident_terminal_apply_op_id") or "")
+ == str(data.get("apply_op_id") or "")
+ and data.get("incident_terminal_resolved") is True
+ and data.get("incident_terminal_telegram_acknowledged") is True
+ and data.get("incident_terminal_repository_readback_verified") is True
+ and data.get("incident_terminal_durable_write_acknowledged") is True
+ )
incident_closed = all(
(
apply_passed,
@@ -644,10 +776,13 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
km_ready,
telegram_ready,
identity_chain_verified,
+ incident_ledger_closure_verified,
)
)
present_stages = set(stage_ids)
+ present_stages.discard("telegram_receipt")
+ present_stages.discard("incident_closure")
for stage_id, present in (
("candidate", candidate_identity_verified),
("check_mode", check_passed and check_identity_verified),
@@ -721,6 +856,8 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
active_blockers.append("same_run_mcp_context_missing")
if verifier_passed and not telegram_ready:
active_blockers.append("same_run_telegram_receipt_missing")
+ if verifier_passed and telegram_ready and not incident_ledger_closure_verified:
+ active_blockers.append("same_run_incident_closure_missing")
return {
"schema_version": SCHEMA_VERSION,
@@ -736,6 +873,44 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
),
"mode": "live_durable_receipt_readback_no_raw_payload_no_secret",
"work_item_id": str(data.get("work_item_id") or WORK_ITEM_ID),
+ "incident": {
+ "incident_ref": str(data.get("incident_id") or "") or None,
+ "status": "resolved" if incident_ledger_closure_verified else "open",
+ "ledger_closure_verified": incident_ledger_closure_verified,
+ "raw_event_payload_returned": False,
+ },
+ "asset_identity": {
+ "canonical_asset_ref": canonical_asset_ref or None,
+ "required_identity_kinds": list(runtime_identity_links),
+ "linked_identity_kinds": [
+ identity_kind
+ for identity_kind, linked in runtime_identity_links.items()
+ if linked
+ ],
+ "required_identity_kind_count": len(runtime_identity_links),
+ "linked_identity_kind_count": sum(runtime_identity_links.values()),
+ "all_required_identity_kinds_linked": all(
+ runtime_identity_links.values()
+ ),
+ "raw_identity_returned": False,
+ },
+ "verifier": {
+ "status": "passed" if verifier_passed else "missing",
+ "receipt_ref": str(data.get("verifier_id") or "") or None,
+ },
+ "recipient_receipt": {
+ "status": (
+ "recipient_visible"
+ if telegram_provider_sent
+ else "suppressed_unchanged"
+ if telegram_notification_suppressed
+ else "missing"
+ ),
+ "receipt_ref": str(data.get("telegram_receipt_id") or "") or None,
+ "provider_message_ref": (
+ str(data.get("telegram_provider_message_id") or "") or None
+ ),
+ },
"trace": {
"trace_id": str(data.get("trace_id") or ""),
"run_id": str(data.get("run_id") or ""),
@@ -777,6 +952,9 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
"telegram_notification_suppressed_count": (
1 if telegram_notification_suppressed else 0
),
+ "incident_ledger_closure_verified_count": (
+ 1 if incident_ledger_closure_verified else 0
+ ),
"same_run_identity_chain_verified_count": (
1 if identity_chain_verified else 0
),
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index efc223058..72f24f582 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -13812,36 +13812,62 @@ class TelegramGateway:
if isinstance(source_refs, dict):
source_refs["automation_run_ids"] = [automation_run_id]
if no_write_posture:
- effective_provider_delivery = (
- "shadow_only"
- if success or provider_delivery == "shadow_only"
- else "digest"
- )
- source_extra["notification_policy"] = {
- "policy_version": "telegram_posture_state_digest_v1",
- "failure_fingerprint": failure_fingerprint,
- "group_scope": (
- "scheduled_posture_healthy"
- if success
- else "catalog_posture_failure"
- ),
- "provider_delivery": effective_provider_delivery,
- "digest_window_minutes": (
- _NO_WRITE_REPLAY_DIGEST_WINDOW_MINUTES
- ),
- "disposition": (
- "suppressed"
- if effective_provider_delivery == "shadow_only"
- else "send_or_suppress"
- ),
- "provider_send_performed": False if success else None,
- "runtime_write_performed": False,
- "details_retained_in": [
- "automation_operation_log",
- "awooop_runs",
- "km_rag_playbook",
- ],
- }
+ if provider_delivery == "state_transition":
+ source_extra["notification_policy"] = {
+ "policy_version": (
+ "telegram_posture_lifecycle_transition_v1"
+ ),
+ "canonical_category": normalized_category,
+ "lifecycle_scope": lifecycle_scope,
+ "lifecycle_state": lifecycle_state,
+ "group_scope": "controlled_asset_lifecycle",
+ "provider_delivery": "state_transition",
+ "disposition": "send_on_transition",
+ "pending_lease_minutes": (
+ _CONTROLLED_APPLY_LIFECYCLE_PENDING_LEASE_MINUTES
+ ),
+ "delivery_semantics": (
+ "at_least_once_after_unknown_ack_lease"
+ ),
+ "provider_send_performed": None,
+ "runtime_write_performed": False,
+ "details_retained_in": [
+ "automation_operation_log",
+ "awooop_runs",
+ "km_rag_playbook",
+ ],
+ }
+ else:
+ effective_provider_delivery = (
+ "shadow_only"
+ if success or provider_delivery == "shadow_only"
+ else "digest"
+ )
+ source_extra["notification_policy"] = {
+ "policy_version": "telegram_posture_state_digest_v1",
+ "failure_fingerprint": failure_fingerprint,
+ "group_scope": (
+ "scheduled_posture_healthy"
+ if success
+ else "catalog_posture_failure"
+ ),
+ "provider_delivery": effective_provider_delivery,
+ "digest_window_minutes": (
+ _NO_WRITE_REPLAY_DIGEST_WINDOW_MINUTES
+ ),
+ "disposition": (
+ "suppressed"
+ if effective_provider_delivery == "shadow_only"
+ else "send_or_suppress"
+ ),
+ "provider_send_performed": False if success else None,
+ "runtime_write_performed": False,
+ "details_retained_in": [
+ "automation_operation_log",
+ "awooop_runs",
+ "km_rag_playbook",
+ ],
+ }
elif no_write_replay:
effective_provider_delivery = (
"shadow_only"
@@ -13933,7 +13959,9 @@ class TelegramGateway:
f"{html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}"
),
(
- "通知: 健康週期讀回不外送；異常相同原因 30 分鐘只通知一次。"
+ "通知: 狀態轉換僅外送一次；相同狀態由 durable receipt 去重。"
+ if provider_delivery == "state_transition"
+ else "通知: 健康週期讀回不外送；異常相同原因 30 分鐘只通知一次。"
),
]
elif no_write_replay:
diff --git a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
index a1be55f1b..d2cbc1200 100644
--- a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
+++ b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
@@ -446,14 +446,14 @@ async def test_backfill_repairs_orphan_incident_and_reuses_verifier(
@pytest.mark.asyncio
-async def test_wazuh_posture_success_receipt_is_shadow_only_no_write(
+async def test_wazuh_posture_success_receipt_is_recipient_visible_transition(
monkeypatch: pytest.MonkeyPatch,
) -> None:
gateway = AsyncMock()
gateway.send_controlled_apply_result_receipt.return_value = {
"ok": True,
"_awooop_outbound_mirror_acknowledged": True,
- "_awooop_delivery_status": "suppressed",
+ "_awooop_delivery_status": "sent",
}
monkeypatch.setattr(
"src.services.telegram_gateway.get_telegram_gateway",
@@ -481,7 +481,7 @@ async def test_wazuh_posture_success_receipt_is_shadow_only_no_write(
assert acknowledged is True
call = gateway.send_controlled_apply_result_receipt.await_args.kwargs
assert call["execution_kind"] == "no_write_posture_readback"
- assert call["provider_delivery"] == "shadow_only"
+ assert call["provider_delivery"] == "state_transition"
def test_wazuh_posture_shadow_receipt_is_a_terminal_notification_receipt() -> None:
diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py
index 1988d1403..cb00407dd 100644
--- a/apps/api/tests/test_ansible_verified_closure.py
+++ b/apps/api/tests/test_ansible_verified_closure.py
@@ -488,6 +488,55 @@ async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
atomic_writer.assert_not_awaited()
+@pytest.mark.asyncio
+async def test_closure_readback_projects_mcp_context_into_telegram_gate(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ apply_input = {
+ "runtime_stage_receipts": [
+ {"stage_id": stage_id, "durable_receipt": True}
+ for stage_id in service._VERIFIED_CLOSURE_REQUIRED_STAGE_IDS
+ ]
+ }
+ db = _SequenceDB(
+ _MappingResult(
+ row={
+ "apply_success": True,
+ "apply_input": apply_input,
+ "apply_verifier_terminal": True,
+ "check_mode_success": True,
+ "candidate_present": True,
+ "auto_repair_receipt": True,
+ "post_verifier_receipt": True,
+ "km_writeback_receipt": True,
+ "playbook_trust_receipt": True,
+ "playbook_readback": True,
+ "timeline_readback": True,
+ "telegram_receipt": False,
+ "approval_projection": True,
+ "execution_lifecycle": True,
+ "telegram_lifecycle": False,
+ }
+ )
+ )
+
+ @asynccontextmanager
+ async def fake_get_db_context(_project_id):
+ yield db
+
+ monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
+
+ readback = await service._read_verified_apply_closure_prerequisites(
+ _claim(),
+ apply_op_id="00000000-0000-0000-0000-000000000104",
+ project_id="awoooi",
+ )
+
+ assert readback["receipts"]["mcp_context"] is True
+ assert readback["missing"] == ["telegram_lifecycle", "telegram_receipt"]
+ assert service._telegram_final_evidence_ready(readback["receipts"]) is True
+
+
@pytest.mark.asyncio
async def test_closure_resolves_only_after_durable_readback(
monkeypatch: pytest.MonkeyPatch,
diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py
index c430673d6..200f143e4 100644
--- a/apps/api/tests/test_asset_scanner_job.py
+++ b/apps/api/tests/test_asset_scanner_job.py
@@ -225,6 +225,25 @@ def test_ai_program_catalog_projects_every_declared_asset_once() -> None:
for asset in assets
)
assert all(asset["metadata"]["secret_value_read"] is False for asset in assets)
+ wazuh = next(
+ asset
+ for asset in assets
+ if asset["metadata"]["program_canonical_id"]
+ == "awoooi:observability_and_security:wazuh"
+ )
+ assert wazuh["metadata"]["runtime_canonical_asset_ids"] == [
+ "service:wazuh-manager"
+ ]
+ assert wazuh["metadata"]["runtime_identity_kinds"] == [
+ "manager",
+ "agent",
+ "host",
+ "service",
+ "event",
+ ]
+ assert "ops/config/service-registry.yaml" in (
+ wazuh["metadata"]["source_truth_refs"]
+ )
assert {
asset["metadata"]["program_scope_id"]: asset["asset_type"]
for asset in assets
diff --git a/apps/api/tests/test_iwooos_security_asset_control_plane.py b/apps/api/tests/test_iwooos_security_asset_control_plane.py
index 74ad6541b..7ca255227 100644
--- a/apps/api/tests/test_iwooos_security_asset_control_plane.py
+++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py
@@ -66,6 +66,60 @@ def _closed_strict_runtime() -> dict:
}
+def _closed_wazuh_runtime_lanes() -> dict:
+ return {
+ "posture": {
+ "summary": {
+ "service_log_evidence_receipt_count": 1,
+ "independent_post_verifier_passed_count": 1,
+ "telegram_provider_send_count": 1,
+ "telegram_notification_suppressed_count": 0,
+ "same_run_present_stage_count": 18,
+ "same_run_required_stage_count": 18,
+ "runtime_closed_count": 1,
+ },
+ "trace": {
+ "source_recurrence_verified": True,
+ "identity_chain_verified": True,
+ },
+ "asset_identity": {
+ "canonical_asset_ref": "service:wazuh-manager",
+ "linked_identity_kinds": [
+ "manager",
+ "agent",
+ "host",
+ "service",
+ "event",
+ ],
+ "linked_identity_kind_count": 5,
+ "required_identity_kind_count": 5,
+ "all_required_identity_kinds_linked": True,
+ "raw_identity_returned": False,
+ },
+ "incident": {
+ "incident_ref": "IWZ-POSTURE-2026072205",
+ "status": "resolved",
+ "ledger_closure_verified": True,
+ "raw_event_payload_returned": False,
+ },
+ "verifier": {
+ "status": "passed",
+ "receipt_ref": "verifier-wazuh-2205",
+ },
+ "recipient_receipt": {
+ "status": "recipient_visible",
+ "receipt_ref": "telegram-wazuh-2205",
+ "provider_message_ref": "2205",
+ },
+ },
+ "ingress": {
+ "boundaries": {
+ "critical_break_glass_required": True,
+ },
+ },
+ }
+
+
def _ready_payload(
source_health: list[dict] | None = None,
*,
@@ -76,6 +130,8 @@ def _ready_payload(
automation_rows: list[SimpleNamespace] | None = None,
runtime_receipt_counts: dict[str, int] | None = None,
strict_runtime: dict | None = None,
+ wazuh_asset_row: SimpleNamespace | None = None,
+ wazuh_runtime: dict | None = None,
) -> dict:
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
discovery = _row(
@@ -176,6 +232,8 @@ def _ready_payload(
mttr_minutes_24h=14.25,
),
strict_runtime_row=strict_runtime,
+ wazuh_asset_row=wazuh_asset_row,
+ wazuh_runtime_row=wazuh_runtime,
audit_row=_row(
k8s_audit_count_24h=2,
k8s_audit_failure_count_24h=0,
@@ -271,6 +329,113 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
assert 'secret_value_collection_allowed": true' not in public_text.lower()
+def test_projects_one_closed_wazuh_soc_slice_without_claiming_global_coverage() -> (
+ None
+):
+ coverage_rows = [
+ _row(dimension=dimension, status="green", cnt=20)
+ for dimension in (
+ "auto_rule_creation",
+ "auto_rule_matching",
+ "auto_playbook",
+ "auto_remediation",
+ "auto_km_creation",
+ )
+ ]
+ coverage_rows.extend(
+ [
+ _row(dimension="auto_monitoring", status="green", cnt=19),
+ _row(dimension="auto_monitoring", status="unknown", cnt=1),
+ _row(dimension="auto_alerting", status="green", cnt=19),
+ _row(dimension="auto_alerting", status="unknown", cnt=1),
+ ]
+ )
+ payload = _ready_payload(
+ coverage_rows=coverage_rows,
+ wazuh_asset_row=_row(
+ canonical_asset_count=1,
+ canonical_asset_ref="service:wazuh-manager",
+ owner_resolved_asset_count=1,
+ owner_verified_asset_count=1,
+ monitoring_coverage_status="unknown",
+ alerting_coverage_status="unknown",
+ ),
+ wazuh_runtime=_closed_wazuh_runtime_lanes(),
+ )
+
+ wazuh = payload["wazuh_soc_slice"]
+ assert wazuh["status"] == "same_run_closed_recipient_visible"
+ assert wazuh["protected_asset_count"] == 1
+ assert wazuh["protected_asset_ref"] == "service:wazuh-manager"
+ assert wazuh["asset_graph_joined"] is True
+ assert wazuh["owner_verified_asset_count"] == 1
+ assert wazuh["telemetry_covered_asset_count"] == 1
+ assert wazuh["identity_kind_count"] == 5
+ assert wazuh["incident_ref"] == "IWZ-POSTURE-2026072205"
+ assert wazuh["verifier_status"] == "passed"
+ assert wazuh["verifier_receipt_ref"] == "verifier-wazuh-2205"
+ assert wazuh["recipient_receipt_status"] == "recipient_visible"
+ assert wazuh["recipient_receipt_ref"] == "telegram-wazuh-2205"
+ assert wazuh["recipient_provider_message_ref"] == "2205"
+ assert wazuh["same_run_present_stage_count"] == 18
+ assert wazuh["same_run_closed"] is True
+ assert wazuh["coverage_overlay_applied_count"] == 2
+ assert wazuh["critical_break_glass_required"] is True
+ assert "wazuh_critical_ingress_break_glass_pending" in (
+ wazuh["remaining_gap_codes"]
+ )
+ assert "global_wazuh_asset_coverage_not_proven" in (
+ wazuh["remaining_gap_codes"]
+ )
+ assert payload["summary"]["automation_coverage_unknown_count"] == 0
+ domains = {
+ domain["domain_id"]: domain
+ for domain in payload["security_program_domains"]
+ }
+ assert domains["endpoint_xdr_wazuh"]["controlled_percent"] == 70
+ assert domains["endpoint_xdr_wazuh"]["status"] == "partial"
+ assert domains["endpoint_xdr_wazuh"]["gap_code"] == (
+ "wazuh_bounded_slice_closed_global_coverage_open"
+ )
+ assert payload["boundaries"]["wazuh_slice_is_global_coverage_claim"] is False
+ assert wazuh["raw_identity_returned"] is False
+ assert wazuh["raw_event_payload_returned"] is False
+ assert wazuh["secret_value_read"] is False
+
+
+def test_wazuh_slice_refuses_coverage_overlay_when_runtime_asset_does_not_join() -> (
+ None
+):
+ runtime = _closed_wazuh_runtime_lanes()
+ runtime["posture"]["asset_identity"]["canonical_asset_ref"] = (
+ "service:wazuh-shadow"
+ )
+ payload = _ready_payload(
+ coverage_rows=[
+ _row(dimension="auto_monitoring", status="unknown", cnt=1),
+ _row(dimension="auto_alerting", status="unknown", cnt=1),
+ ],
+ wazuh_asset_row=_row(
+ canonical_asset_count=1,
+ canonical_asset_ref="service:wazuh-manager",
+ owner_resolved_asset_count=1,
+ owner_verified_asset_count=1,
+ monitoring_coverage_status="unknown",
+ alerting_coverage_status="unknown",
+ ),
+ wazuh_runtime=runtime,
+ )
+
+ wazuh = payload["wazuh_soc_slice"]
+ assert wazuh["asset_graph_joined"] is False
+ assert wazuh["protected_asset_count"] == 0
+ assert wazuh["telemetry_covered_asset_count"] == 0
+ assert wazuh["coverage_overlay_applied_count"] == 0
+ assert "wazuh_runtime_asset_graph_join_missing" in (
+ wazuh["remaining_gap_codes"]
+ )
+
+
def test_uncovered_security_domains_are_explicit_ordered_p0_work_items() -> None:
payload = _ready_payload(coverage_rows=[], compliance_rows=[])
@@ -783,8 +948,8 @@ def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> N
assert payload["summary"]["overall_security_completion_percent"] == 0
assert payload["completion"]["overall_percent"] == 0
assert payload["completion"]["strict_runtime_closure_percent"] == 0
- assert payload["summary"]["source_count"] == 10
- assert payload["summary"]["unavailable_source_count"] == 10
+ assert payload["summary"]["source_count"] == 12
+ assert payload["summary"]["unavailable_source_count"] == 12
assert payload["ai_automation"]["strict_runtime_completion"]["closed"] is False
assert payload["discovery"]["fresh"] is False
assert payload["boundaries"]["raw_asset_identity_returned"] is False
@@ -862,13 +1027,25 @@ def test_service_uses_canonical_awoooi_project_scope(monkeypatch) -> None:
"src.services.iwooos_security_asset_control_plane._read_strict_runtime_source",
strict_runtime_unavailable,
)
+
+ async def wazuh_runtime_unavailable(**_kwargs):
+ return {}, {
+ "source_id": "wazuh_controlled_runtime",
+ "status": "unavailable",
+ "reason_code": "wazuh_controlled_runtime_query_failed",
+ }
+
+ monkeypatch.setattr(
+ "src.services.iwooos_security_asset_control_plane._read_wazuh_runtime_source",
+ wazuh_runtime_unavailable,
+ )
payload = asyncio.run(service.get_snapshot())
- assert requested_projects == ["awoooi"] * 9
+ assert requested_projects == ["awoooi"] * 10
assert payload["status"] == "degraded"
assert payload["source_status"] == "live_database_unavailable"
- assert payload["summary"]["source_count"] == 10
- assert payload["summary"]["unavailable_source_count"] == 10
+ assert payload["summary"]["source_count"] == 12
+ assert payload["summary"]["unavailable_source_count"] == 12
def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> None:
@@ -918,11 +1095,23 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
"src.services.iwooos_security_asset_control_plane._read_strict_runtime_source",
strict_runtime_ready,
)
+
+ async def wazuh_runtime_ready(**_kwargs):
+ return _closed_wazuh_runtime_lanes(), {
+ "source_id": "wazuh_controlled_runtime",
+ "status": "ready",
+ "reason_code": None,
+ }
+
+ monkeypatch.setattr(
+ "src.services.iwooos_security_asset_control_plane._read_wazuh_runtime_source",
+ wazuh_runtime_ready,
+ )
payload = asyncio.run(service._load_snapshot())
assert concurrency["maximum"] == 3
- assert payload["summary"]["source_count"] == 10
- assert payload["summary"]["ready_source_count"] == 10
+ assert payload["summary"]["source_count"] == 12
+ assert payload["summary"]["ready_source_count"] == 12
assert payload["summary"]["unavailable_source_count"] == 0
compliance_query = next(
statement
@@ -944,6 +1133,18 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
assert "digest_pinned" in inventory_query
assert "live_tls_probe_evidenced" in inventory_query
assert "live_tls_healthy" in inventory_query
+ wazuh_asset_query = next(
+ statement
+ for statement in statements
+ if "program_canonical_id" in statement
+ )
+ assert "awoooi:observability_and_security:wazuh" in wazuh_asset_query
+ assert "service:wazuh-manager" in wazuh_asset_query
+ assert "runtime_canonical_asset_ids" in wazuh_asset_query
+ assert "lower(btrim(owner_team)) = 'aisoc'" in wazuh_asset_query
+ assert "auto_monitoring" in wazuh_asset_query
+ assert "auto_alerting" in wazuh_asset_query
+ assert "snapshot.detected_at DESC" in wazuh_asset_query
siem_query = next(
statement for statement in statements if "FROM alert_rule_catalog" in statement
)
diff --git a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
index 58e021b27..7ea3be560 100644
--- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
@@ -31,6 +31,7 @@ from src.services.awooop_ansible_post_verifier import postconditions_for_catalog
from src.services.iwooos_wazuh_controlled_executor_runtime import (
_LIVE_RUNTIME_SQL,
build_iwooos_wazuh_controlled_executor_runtime_readback,
+ load_iwooos_wazuh_controlled_executor_runtime_lanes,
)
@@ -51,6 +52,9 @@ def test_wazuh_runtime_sql_reads_schema_safe_packet_fallback() -> None:
assert "ansible_controlled_capability_repair_queued" in _LIVE_RUNTIME_SQL
assert "remediation_executed" in _LIVE_RUNTIME_SQL
assert "semantic_operation_type" in _LIVE_RUNTIME_SQL
+ assert "LEFT JOIN incidents incident" in _LIVE_RUNTIME_SQL
+ assert "incident_terminal_automation_run_id" in _LIVE_RUNTIME_SQL
+ assert "incident_terminal_durable_write_acknowledged" in _LIVE_RUNTIME_SQL
def _durable_scheduled_wazuh_source_recurrence(
@@ -80,6 +84,7 @@ def _complete_row() -> dict[str, object]:
)
return {
"candidate_op_id": run_id,
+ "incident_id": "IWZ-POSTURE-2026071500",
"candidate_status": "dry_run",
"automation_run_id": run_id,
"trace_id": run_id,
@@ -123,6 +128,16 @@ def _complete_row() -> dict[str, object]:
"telegram_receipt_id": "telegram-303",
"telegram_send_status": "sent",
"telegram_provider_message_id": "303",
+ "incident_status": "resolved",
+ "incident_resolved_at": "2026-07-11T03:36:30+00:00",
+ "incident_terminal_automation_run_id": run_id,
+ "incident_terminal_apply_op_id": (
+ "00000000-0000-0000-0000-000000000303"
+ ),
+ "incident_terminal_resolved": True,
+ "incident_terminal_telegram_acknowledged": True,
+ "incident_terminal_repository_readback_verified": True,
+ "incident_terminal_durable_write_acknowledged": True,
"stage_ids": [
"mcp_context",
"service_log_evidence",
@@ -160,12 +175,103 @@ def test_wazuh_runtime_readback_closes_only_with_complete_same_run_receipts() ->
assert payload["summary"]["same_run_present_stage_count"] == 18
assert payload["summary"]["same_run_completion_percent"] == 100
assert payload["summary"]["runtime_closed_count"] == 1
+ assert payload["incident"] == {
+ "incident_ref": "IWZ-POSTURE-2026071500",
+ "status": "resolved",
+ "ledger_closure_verified": True,
+ "raw_event_payload_returned": False,
+ }
+ assert payload["asset_identity"]["canonical_asset_ref"] == (
+ "service:wazuh-manager"
+ )
+ assert payload["asset_identity"]["linked_identity_kind_count"] == 5
+ assert payload["asset_identity"]["all_required_identity_kinds_linked"] is True
+ assert payload["asset_identity"]["raw_identity_returned"] is False
+ assert payload["verifier"] == {
+ "status": "passed",
+ "receipt_ref": "verifier-303",
+ }
+ assert payload["recipient_receipt"] == {
+ "status": "recipient_visible",
+ "receipt_ref": "telegram-303",
+ "provider_message_ref": "303",
+ }
assert payload["missing_stage_ids"] == []
assert payload["boundaries"]["host_write_performed"] is False
assert payload["boundaries"]["wazuh_active_response_performed"] is False
assert payload["boundaries"]["command_output_returned"] is False
+def test_wazuh_runtime_refuses_derived_closure_without_incident_ledger_readback() -> (
+ None
+):
+ row = _complete_row()
+ row["incident_status"] = "mitigating"
+ row["incident_resolved_at"] = None
+
+ payload = build_iwooos_wazuh_controlled_executor_runtime_readback(
+ row,
+ executor_enabled=True,
+ )
+
+ assert payload["incident"]["ledger_closure_verified"] is False
+ assert payload["incident"]["status"] == "open"
+ assert payload["summary"]["runtime_closed_count"] == 0
+ assert "incident_closure" in payload["missing_stage_ids"]
+ assert "same_run_incident_closure_missing" in payload["active_blockers"]
+
+
+@pytest.mark.asyncio
+async def test_wazuh_runtime_loader_keeps_ingress_and_posture_lanes_visible(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ ingress = _complete_row()
+ ingress.update(
+ {
+ "catalog_id": "ansible:wazuh-alertmanager-integration",
+ "work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
+ "proposal_source": "iwooos_wazuh_alert_ingress_scheduler",
+ "run_namespace": "awoooi:iwooos:wazuh-alert-ingress",
+ "check_work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
+ "apply_work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
+ "verifier_work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
+ }
+ )
+ posture = _complete_row()
+ rows = [ingress, posture]
+
+ class Result:
+ def __init__(self, row):
+ self.row = row
+
+ def mappings(self):
+ return self
+
+ def first(self):
+ return self.row
+
+ class Db:
+ async def execute(self, _statement, _params):
+ return Result(rows.pop(0))
+
+ @asynccontextmanager
+ async def fake_get_db_context(project_id: str):
+ assert project_id == "awoooi"
+ yield Db()
+
+ monkeypatch.setattr(
+ "src.services.iwooos_wazuh_controlled_executor_runtime.get_db_context",
+ fake_get_db_context,
+ )
+
+ lanes = await load_iwooos_wazuh_controlled_executor_runtime_lanes()
+
+ assert set(lanes) == {"ingress", "posture"}
+ assert lanes["ingress"]["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS"
+ assert lanes["posture"]["work_item_id"] == "P0-03-WAZUH-MANAGER-POSTURE"
+ assert lanes["posture"]["summary"]["runtime_closed_count"] == 1
+
+
def test_wazuh_posture_shadow_receipt_closes_without_provider_send() -> None:
row = _complete_row()
row.update(
@@ -1462,6 +1568,101 @@ async def test_wazuh_posture_scheduler_retries_incomplete_learning_once_per_depl
duplicate_collector.assert_not_awaited()
+@pytest.mark.asyncio
+async def test_wazuh_posture_scheduler_queues_one_recipient_transition_per_deploy(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ from src.jobs import awooop_ansible_candidate_backfill_job as job
+
+ existing_row = {
+ "op_id": "00000000-0000-0000-0000-000000000326",
+ "candidate_status": "success",
+ "automation_run_id": "00000000-0000-0000-0000-000000000326",
+ "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071100",
+ "latest_check_status": "success",
+ "latest_apply_status": "success",
+ "latest_learning_status": "success",
+ "latest_telegram_send_status": "shadow",
+ "latest_telegram_provider_message_id": None,
+ "predecision_evidence_ready": True,
+ }
+
+ class _Mappings:
+ def first(self):
+ return existing_row
+
+ class _Result:
+ def mappings(self):
+ return _Mappings()
+
+ class _Db:
+ async def execute(self, _statement, _params):
+ return _Result()
+
+ @asynccontextmanager
+ async def fake_db_context(project_id: str):
+ assert project_id == "awoooi"
+ yield _Db()
+
+ recorder = AsyncMock(return_value=True)
+ monkeypatch.setattr(job, "get_db_context", fake_db_context)
+ monkeypatch.setattr(
+ job.settings,
+ "ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
+ True,
+ )
+ monkeypatch.setattr(
+ job.settings,
+ "IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
+ 6,
+ )
+ monkeypatch.setattr(
+ job,
+ "now_taipei",
+ MagicMock(return_value=datetime.fromisoformat("2026-07-11T05:59:59+08:00")),
+ )
+ monkeypatch.setenv(
+ "AWOOOI_BUILD_COMMIT_SHA",
+ "abcdef1234567890abcdef1234567890abcdef12",
+ )
+
+ result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once(
+ recorder=recorder,
+ evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-5")),
+ evidence_verifier=AsyncMock(return_value=True),
+ source_recurrence=_current_wazuh_source_recurrence(
+ "2026-07-11T05:59:58+08:00"
+ ),
+ )
+
+ assert result["status"] == "queued_for_controlled_executor_retry"
+ assert result["retry_reason"] == "recipient_visible_transition_missing"
+ assert result["retry_of_incomplete_runtime_closure"] is True
+ recorded = recorder.await_args.kwargs
+ assert recorded["incident"]["source_receipt_ref"] == (
+ "scheduled-wazuh-manager-posture:2026071100-ABCDEF123456-RECEIPT"
+ )
+
+ existing_row["automation_run_id"] = result["automation_run_id"]
+ existing_row["source_receipt_ref"] = recorded["incident"]["source_receipt_ref"]
+ duplicate_collector = AsyncMock()
+ duplicate = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once(
+ recorder=recorder,
+ evidence_collector=duplicate_collector,
+ source_recurrence=_current_wazuh_source_recurrence(
+ "2026-07-11T05:59:58+08:00"
+ ),
+ )
+
+ assert duplicate["status"] == (
+ "incomplete_runtime_retry_already_attempted_for_deploy"
+ )
+ assert duplicate["retry_reason"] == "recipient_visible_transition_missing"
+ assert duplicate["queued"] == 0
+ recorder.assert_awaited_once()
+ duplicate_collector.assert_not_awaited()
+
+
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate(
monkeypatch: pytest.MonkeyPatch,
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 44697cf0d..3658406af 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/apps/api/tests/test_telegram_message_templates.py
@@ -1636,6 +1636,74 @@ async def test_healthy_posture_readback_is_suppressed_and_never_claims_apply(
assert policy["runtime_write_performed"] is False
+@pytest.mark.asyncio
+async def test_healthy_wazuh_posture_transition_is_recipient_visible_once(
+ monkeypatch,
+) -> None:
+ gateway = TelegramGateway()
+ sent_requests = []
+
+ async def fake_send_request(method, payload):
+ sent_requests.append((method, payload))
+ return {
+ "ok": True,
+ "_awooop_outbound_mirror_acknowledged": True,
+ "_awooop_delivery_status": "sent",
+ }
+
+ async def fake_fetch_truth_chain(**_kwargs):
+ return {"truth_status": {}, "automation_quality": {}, "execution": {}}
+
+ async def fake_fetch_km_completion_summary(**_kwargs):
+ return {}
+
+ monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat"))
+ monkeypatch.setattr(gateway, "_send_request", fake_send_request)
+ monkeypatch.setattr(
+ "src.services.awooop_truth_chain_service.fetch_truth_chain",
+ fake_fetch_truth_chain,
+ )
+ monkeypatch.setattr(
+ "src.services.telegram_gateway._fetch_km_stale_completion_summary_for_incident",
+ fake_fetch_km_completion_summary,
+ )
+
+ result = await gateway.send_controlled_apply_result_receipt(
+ automation_run_id="00000000-0000-0000-0000-000000000013",
+ incident_id="IWZ-POSTURE-2026072206",
+ catalog_id="ansible:wazuh-manager-posture-readback",
+ apply_op_id="00000000-0000-0000-0000-000000000014",
+ playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+ verification_result="success",
+ returncode=0,
+ duration_ms=120,
+ verifier_written=True,
+ learning_written=True,
+ rag_written=True,
+ mcp_context_ready=True,
+ playbook_trust_written=True,
+ alert_category="security",
+ canonical_asset_id="service:wazuh-manager",
+ execution_kind="no_write_posture_readback",
+ provider_delivery="state_transition",
+ project_id="awoooi",
+ )
+
+ assert result["ok"] is True
+ method, payload = sent_requests[0]
+ assert method == "sendMessage"
+ assert "Runtime write: 0" in payload["text"]
+ assert "狀態轉換僅外送一次" in payload["text"]
+ policy = payload["_awooop_source_envelope_extra"]["notification_policy"]
+ assert policy["policy_version"] == (
+ "telegram_posture_lifecycle_transition_v1"
+ )
+ assert policy["provider_delivery"] == "state_transition"
+ assert policy["disposition"] == "send_on_transition"
+ assert policy["lifecycle_state"] == "recovered"
+ assert policy["runtime_write_performed"] is False
+
+
@pytest.mark.asyncio
async def test_failed_posture_readback_is_digest_deduplicated_and_queues_repair(
monkeypatch,
diff --git a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx
index 0d1db45bf..fe87d96a0 100644
--- a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx
+++ b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx
@@ -57,6 +57,17 @@ type Copy = {
forbiddenSources: string;
imagePolicy: string;
publicTls: string;
+ wazuhSlice: string;
+ protectedAssets: string;
+ ownerTelemetry: string;
+ identityLink: string;
+ incident: string;
+ verifier: string;
+ recipientReceipt: string;
+ remainingGaps: string;
+ delivered: string;
+ closed: string;
+ open: string;
certificates: string;
tlsEvidence: string;
tlsHealthy: string;
@@ -113,6 +124,17 @@ const COPY: Record<"zh" | "en", Copy> = {
forbiddenSources: "禁用來源",
imagePolicy: "映像政策",
publicTls: "Public TLS",
+ wazuhSlice: "Wazuh AI SOC 同 run",
+ protectedAssets: "受保護資產",
+ ownerTelemetry: "Owner / 遙測",
+ identityLink: "Identity 串接",
+ incident: "事件處置",
+ verifier: "獨立驗證",
+ recipientReceipt: "Telegram 收據",
+ remainingGaps: "剩餘缺口",
+ delivered: "已送達",
+ closed: "已閉環",
+ open: "未閉環",
certificates: "憑證資產",
tlsEvidence: "Live 驗證",
tlsHealthy: "健康憑證",
@@ -167,6 +189,17 @@ const COPY: Record<"zh" | "en", Copy> = {
forbiddenSources: "Forbidden sources",
imagePolicy: "Image policy",
publicTls: "Public TLS",
+ wazuhSlice: "Wazuh AI SOC same run",
+ protectedAssets: "Protected assets",
+ ownerTelemetry: "Owner / telemetry",
+ identityLink: "Identity link",
+ incident: "Incident",
+ verifier: "Independent verify",
+ recipientReceipt: "Telegram receipt",
+ remainingGaps: "Remaining gaps",
+ delivered: "DELIVERED",
+ closed: "CLOSED",
+ open: "OPEN",
certificates: "Certificates",
tlsEvidence: "Live evidence",
tlsHealthy: "Healthy certs",
@@ -290,12 +323,14 @@ function TruthCell({
icon: Icon,
tone = "neutral",
wideOnMobile = false,
+ title,
}: {
label: string;
value: string | number;
icon: typeof ShieldCheck;
tone?: "neutral" | "healthy" | "critical";
wideOnMobile?: boolean;
+ title?: string;
}) {
return (
0,
+ );
+ const identityReady = Boolean(
+ wazuh && wazuh.identity_kind_count === wazuh.required_identity_kind_count,
+ );
+ const receiptVisible =
+ wazuh?.recipient_receipt_status === "recipient_visible";
+ const verifierRef = wazuh?.verifier_receipt_ref ?? "";
+ const recipientMessageRef = wazuh?.recipient_provider_message_ref ?? "";
+
+ return (
+
+
+
+
+ {copy.wazuhSlice}
+
+ {wazuh?.protected_asset_ref ?? wazuh?.canonical_asset_ref ?? "--"}
+
+ →
+
+ {wazuh?.incident_ref ?? "--"}
+
+
+
+ {copy.remainingGaps} {wazuh?.remaining_gap_codes.length ?? "--"}
+
+
+
+ 0 ? "healthy" : "critical"
+ }
+ />
+
+
+
+
+
+
+
) : null}
+ {view === "command" ? (
+
+ ) : null}
+
{view === "command" ? (
{
false,
);
+ const unsafeWazuhIdentity = {
+ ...safe,
+ wazuh_soc_slice: {
+ ...safe.wazuh_soc_slice,
+ raw_identity_returned: true,
+ },
+ };
+ expect(isSecurityAssetControlPlanePayload(unsafeWazuhIdentity)).toBe(false);
+
const unsafeDiscovery = {
...safe,
discovery: {
diff --git a/apps/web/src/lib/security-asset-control-plane.ts b/apps/web/src/lib/security-asset-control-plane.ts
index e64e1c870..63423f644 100644
--- a/apps/web/src/lib/security-asset-control-plane.ts
+++ b/apps/web/src/lib/security-asset-control-plane.ts
@@ -121,6 +121,44 @@ export type SecurityControlPlaneCertificateInventory = {
private_key_material_read: false;
};
+export type SecurityWazuhSocSlice = {
+ schema_version: "iwooos_wazuh_ai_soc_slice_v1";
+ status: "same_run_closed_recipient_visible" | "partial" | "not_evidenced";
+ canonical_asset_count: number;
+ canonical_asset_ref: "service:wazuh-manager" | null;
+ protected_asset_count: number;
+ protected_asset_ref: "service:wazuh-manager" | null;
+ asset_graph_joined: boolean;
+ owner_resolved_asset_count: number;
+ owner_verified_asset_count: number;
+ telemetry_covered_asset_count: number;
+ identity_kind_count: number;
+ required_identity_kind_count: number;
+ identity_kinds: Array<"manager" | "agent" | "host" | "service" | "event">;
+ incident_ref: string | null;
+ incident_status: "resolved" | "open";
+ verifier_status: "passed" | "missing";
+ verifier_receipt_ref: string | null;
+ recipient_receipt_status:
+ | "recipient_visible"
+ | "suppressed_unchanged"
+ | "missing";
+ recipient_receipt_ref: string | null;
+ recipient_provider_message_ref: string | null;
+ same_run_present_stage_count: number;
+ same_run_required_stage_count: number;
+ same_run_closed: boolean;
+ critical_break_glass_required: boolean;
+ remaining_gap_codes: string[];
+ domain_controlled_percent: number;
+ domain_evidence_count: number;
+ domain_gap_code: string;
+ coverage_overlay_applied_count: number;
+ raw_identity_returned: false;
+ raw_event_payload_returned: false;
+ secret_value_read: false;
+};
+
export type SecurityStrictRuntimeCompletion = {
schema_version: "ai_agent_strict_runtime_completion_v2";
runtime_contract_schema_version: "ai_automation_runtime_completion_contract_v2";
@@ -184,6 +222,7 @@ export type SecurityAssetControlPlanePayload = {
completion_percent: number;
}>;
};
+ wazuh_soc_slice: SecurityWazuhSocSlice;
siem: {
pipeline: SecurityControlPlaneStage[];
incident_total_24h?: number;
@@ -232,6 +271,7 @@ export type SecurityAssetControlPlanePayload = {
live_scan_triggered: boolean;
runtime_action_triggered: boolean;
completion_requires_same_run_receipts: boolean;
+ wazuh_slice_is_global_coverage_claim: false;
};
};
@@ -270,6 +310,98 @@ export function isSecurityAssetControlPlanePayload(
}
if (!isRecord(value.siem) || !Array.isArray(value.siem.pipeline))
return false;
+ const wazuh = value.wazuh_soc_slice;
+ const wazuhIdentityKinds = new Set([
+ "manager",
+ "agent",
+ "host",
+ "service",
+ "event",
+ ]);
+ if (
+ !isRecord(wazuh) ||
+ wazuh.schema_version !== "iwooos_wazuh_ai_soc_slice_v1" ||
+ !["same_run_closed_recipient_visible", "partial", "not_evidenced"].includes(
+ String(wazuh.status),
+ ) ||
+ typeof wazuh.canonical_asset_count !== "number" ||
+ !(
+ wazuh.canonical_asset_ref === null ||
+ wazuh.canonical_asset_ref === "service:wazuh-manager"
+ ) ||
+ typeof wazuh.protected_asset_count !== "number" ||
+ !(
+ wazuh.protected_asset_ref === null ||
+ wazuh.protected_asset_ref === "service:wazuh-manager"
+ ) ||
+ typeof wazuh.asset_graph_joined !== "boolean" ||
+ typeof wazuh.owner_resolved_asset_count !== "number" ||
+ typeof wazuh.owner_verified_asset_count !== "number" ||
+ typeof wazuh.telemetry_covered_asset_count !== "number" ||
+ typeof wazuh.identity_kind_count !== "number" ||
+ typeof wazuh.required_identity_kind_count !== "number" ||
+ !Array.isArray(wazuh.identity_kinds) ||
+ !wazuh.identity_kinds.every(
+ (identityKind) =>
+ typeof identityKind === "string" &&
+ wazuhIdentityKinds.has(identityKind),
+ ) ||
+ new Set(wazuh.identity_kinds).size !== wazuh.identity_kinds.length ||
+ wazuh.identity_kind_count !== wazuh.identity_kinds.length ||
+ wazuh.required_identity_kind_count !== 5 ||
+ !(wazuh.incident_ref === null || typeof wazuh.incident_ref === "string") ||
+ !["resolved", "open"].includes(String(wazuh.incident_status)) ||
+ !["passed", "missing"].includes(String(wazuh.verifier_status)) ||
+ !(
+ wazuh.verifier_receipt_ref === null ||
+ typeof wazuh.verifier_receipt_ref === "string"
+ ) ||
+ !["recipient_visible", "suppressed_unchanged", "missing"].includes(
+ String(wazuh.recipient_receipt_status),
+ ) ||
+ !(
+ wazuh.recipient_receipt_ref === null ||
+ typeof wazuh.recipient_receipt_ref === "string"
+ ) ||
+ !(
+ wazuh.recipient_provider_message_ref === null ||
+ typeof wazuh.recipient_provider_message_ref === "string"
+ ) ||
+ typeof wazuh.same_run_present_stage_count !== "number" ||
+ wazuh.same_run_required_stage_count !== 18 ||
+ typeof wazuh.same_run_closed !== "boolean" ||
+ typeof wazuh.critical_break_glass_required !== "boolean" ||
+ !Array.isArray(wazuh.remaining_gap_codes) ||
+ !wazuh.remaining_gap_codes.every(
+ (gapCode) => typeof gapCode === "string",
+ ) ||
+ typeof wazuh.domain_controlled_percent !== "number" ||
+ wazuh.domain_controlled_percent < 0 ||
+ wazuh.domain_controlled_percent > 70 ||
+ typeof wazuh.domain_evidence_count !== "number" ||
+ typeof wazuh.domain_gap_code !== "string" ||
+ typeof wazuh.coverage_overlay_applied_count !== "number" ||
+ wazuh.raw_identity_returned !== false ||
+ wazuh.raw_event_payload_returned !== false ||
+ wazuh.secret_value_read !== false ||
+ wazuh.protected_asset_count > wazuh.canonical_asset_count ||
+ (wazuh.asset_graph_joined &&
+ wazuh.canonical_asset_ref !== "service:wazuh-manager") ||
+ (wazuh.protected_asset_count === 1 &&
+ wazuh.protected_asset_ref !== wazuh.canonical_asset_ref) ||
+ (wazuh.status === "same_run_closed_recipient_visible" &&
+ (!wazuh.same_run_closed ||
+ !wazuh.asset_graph_joined ||
+ wazuh.same_run_present_stage_count !== 18 ||
+ wazuh.incident_status !== "resolved" ||
+ wazuh.verifier_status !== "passed" ||
+ !wazuh.verifier_receipt_ref ||
+ wazuh.recipient_receipt_status !== "recipient_visible" ||
+ !wazuh.recipient_receipt_ref ||
+ !wazuh.recipient_provider_message_ref))
+ ) {
+ return false;
+ }
if (
!isRecord(value.ai_automation) ||
!Array.isArray(value.ai_automation.stages)
@@ -371,7 +503,8 @@ export function isSecurityAssetControlPlanePayload(
value.boundaries.raw_discovery_error_returned === false &&
value.boundaries.secret_value_collection_allowed === false &&
value.boundaries.live_scan_triggered === false &&
- value.boundaries.runtime_action_triggered === false
+ value.boundaries.runtime_action_triggered === false &&
+ value.boundaries.wazuh_slice_is_global_coverage_claim === false
);
}