feat(iwooos): run wazuh posture receipt chain

apps/api/src/api/v1/iwooos.py

@@ -49,6 +49,9 @@ from src.services.iwooos_wazuh_controlled_executor_handoff import (
 from src.services.iwooos_wazuh_controlled_executor_handoff import (
     preview_iwooos_wazuh_controlled_executor_handoff as preview_wazuh_controlled_executor_handoff_payload,
 )
+from src.services.iwooos_wazuh_controlled_executor_runtime import (
+    load_latest_iwooos_wazuh_controlled_executor_runtime,
+)
 from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
     load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
 )
@@ -828,6 +831,23 @@ async def preview_iwooos_wazuh_controlled_executor_handoff(
         ) from exc
 
 
+@router.get(
+    "/api/v1/iwooos/wazuh-controlled-executor-runtime-readback",
+    response_model=dict[str, Any],
+    summary="取得 Wazuh controlled executor live runtime receipt 讀回",
+    description=(
+        "讀取內部 worker 排程的 Wazuh manager bounded no-write posture executor "
+        "candidate、check-mode、execution、independent verifier、KM / RAG / PlayBook、"
+        "MCP context 與 Telegram durable receipts。此端點不執行命令、不回傳 command output、"
+        "inventory identity、主機位址、raw Wazuh payload 或機密值。"
+    ),
+)
+async def get_iwooos_wazuh_controlled_executor_runtime_readback() -> dict[str, Any]:
+    """回傳 Wazuh controlled executor 的公開安全 production receipt 狀態。"""
+    payload = await load_latest_iwooos_wazuh_controlled_executor_runtime()
+    return redact_public_lan_topology(payload)
+
+
 @router.get(
     "/api/v1/iwooos/wazuh-runtime-gate-owner-review-readback",
     response_model=dict[str, Any],

apps/api/src/core/config.py

@@ -650,6 +650,21 @@ class Settings(BaseSettings):
             "routes remain blocked by catalog and guardrails."
         ),
     )
+    ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR: bool = Field(
+        default=False,
+        description=(
+            "True=periodically enqueue the allowlisted no-write Wazuh manager "
+            "posture playbook into the existing Ansible controlled executor."
+        ),
+    )
+    IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS: int = Field(
+        default=6,
+        ge=1,
+        le=24,
+        description=(
+            "Minimum freshness window between Wazuh manager posture executor runs."
+        ),
+    )
     AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS: str = Field(
         default="low,medium,high",
         description=(

apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py

@@ -14,7 +14,7 @@ import random
 from collections.abc import Awaitable, Callable
 from types import SimpleNamespace
 from typing import Any
-from uuid import uuid4
+from uuid import NAMESPACE_URL, uuid4, uuid5
 
 import structlog
 from sqlalchemy import text
@@ -31,6 +31,7 @@ from src.services.awooop_ansible_check_mode_service import (
 )
 from src.services.evidence_snapshot import EvidenceSnapshot
 from src.services.pre_decision_investigator import get_pre_decision_investigator
+from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
 
@@ -51,6 +52,11 @@ def _error_backoff_seconds() -> float:
     return random.uniform(_ERROR_BACKOFF_MIN_SECONDS, _ERROR_BACKOFF_MAX_SECONDS)
 
 
+_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"
+
+
 async def _fetch_missing_candidate_incidents(
     *,
     project_id: str,
@@ -343,6 +349,179 @@ async def enqueue_ai_decision_ansible_candidate(
     }
 
 
+def _wazuh_posture_incident(
+    *, automation_run_id: str, bucket_ref: str
+) -> dict[str, Any]:
+    return {
+        "incident_id": f"IWZ-POSTURE-{bucket_ref}",
+        "project_id": "awoooi",
+        "status": "INVESTIGATING",
+        "severity": "low",
+        "alertname": "WazuhManagerPostureScheduled",
+        "alert_category": "security",
+        "notification_type": "scheduled_posture",
+        "trace_id": automation_run_id,
+        "run_id": automation_run_id,
+        "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
+        "source_receipt_ref": f"scheduled-wazuh-manager-posture:{bucket_ref}",
+        "asset_scope_aliases": [_WAZUH_POSTURE_PUBLIC_ASSET_ALIAS],
+        "affected_services": ["wazuh-manager", "siem"],
+        "signals": [
+            {
+                "alert_name": "WazuhManagerPostureScheduled",
+                "source": "iwooos_scheduler",
+                "labels": {
+                    "alertname": "WazuhManagerPostureScheduled",
+                    "service": "wazuh-manager",
+                    "tool": "wazuh",
+                    "asset_alias": _WAZUH_POSTURE_PUBLIC_ASSET_ALIAS,
+                },
+                "annotations": {
+                    "summary": "bounded no-write Wazuh manager posture readback"
+                },
+            }
+        ],
+    }
+
+
+async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
+    *,
+    project_id: str = "awoooi",
+    recorder: Recorder = record_ansible_decision_audit,
+    evidence_collector: EvidenceCollector = collect_ansible_candidate_pre_decision_evidence,
+    evidence_verifier: EvidenceVerifier = verify_ansible_candidate_pre_decision_evidence,
+) -> dict[str, Any]:
+    """Enqueue one fresh, no-write Wazuh manager posture executor run."""
+
+    if not settings.ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR:
+        return {
+            "status": "disabled",
+            "queued": 0,
+            "existing": 0,
+            "evidence_ready": 0,
+            "active_blockers": ["wazuh_manager_posture_executor_disabled"],
+        }
+
+    freshness_hours = max(
+        1,
+        int(settings.IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS),
+    )
+    try:
+        async with get_db_context(project_id) as db:
+            existing = await db.execute(
+                text("""
+                    SELECT
+                        op_id::text AS op_id,
+                        status,
+                        input ->> 'automation_run_id' AS automation_run_id
+                    FROM automation_operation_log
+                    WHERE operation_type = 'ansible_candidate_matched'
+                      AND input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb)
+                      AND created_at >= NOW() - (:freshness_hours * INTERVAL '1 hour')
+                    ORDER BY created_at DESC
+                    LIMIT 1
+                """),
+                {
+                    "catalog_match": json.dumps(
+                        [{"catalog_id": _WAZUH_POSTURE_CATALOG_ID}]
+                    ),
+                    "freshness_hours": freshness_hours,
+                },
+            )
+            existing_row = existing.mappings().first()
+        if existing_row:
+            return {
+                "status": "fresh_candidate_already_present",
+                "queued": 0,
+                "existing": 1,
+                "evidence_ready": 1,
+                "automation_run_id": str(
+                    existing_row.get("automation_run_id")
+                    or existing_row.get("op_id")
+                    or ""
+                ),
+                "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
+                "active_blockers": [],
+            }
+
+        now = now_taipei()
+        bucket_hour = (now.hour // freshness_hours) * freshness_hours
+        bucket = now.replace(hour=bucket_hour, minute=0, second=0, microsecond=0)
+        automation_run_id = str(
+            uuid5(
+                NAMESPACE_URL,
+                f"awoooi:iwooos:wazuh-manager-posture:{bucket.isoformat()}",
+            )
+        )
+        incident = _wazuh_posture_incident(
+            automation_run_id=automation_run_id,
+            bucket_ref=bucket.strftime("%Y%m%d%H"),
+        )
+        try:
+            snapshot = await evidence_collector(
+                incident=incident,
+                project_id=project_id,
+                automation_run_id=automation_run_id,
+            )
+            evidence_ready = await evidence_verifier(
+                snapshot=snapshot,
+                project_id=project_id,
+            )
+        except Exception as exc:
+            evidence_ready = False
+            logger.warning(
+                "iwooos_wazuh_manager_posture_predecision_evidence_degraded",
+                error_type=type(exc).__name__,
+            )
+
+        queued = await recorder(
+            incident=incident,
+            proposal_data={
+                "source": "iwooos_wazuh_manager_posture_scheduler",
+                "risk_level": "low",
+                "action": "run_bounded_no_write_wazuh_manager_posture_readback",
+            },
+            decision_path=_BACKFILL_DECISION_PATH,
+            not_used_reason=(
+                "scheduled Wazuh manager posture readback entered the allowlisted "
+                "check-mode and independent verifier chain"
+            ),
+            automation_run_id=automation_run_id,
+        )
+        return {
+            "status": (
+                "queued_for_controlled_executor"
+                if queued and evidence_ready
+                else (
+                    "queued_for_controlled_executor_with_degraded_context"
+                    if queued
+                    else "candidate_race_deduplicated"
+                )
+            ),
+            "queued": 1 if queued else 0,
+            "existing": 0 if queued else 1,
+            "evidence_ready": 1 if evidence_ready else 0,
+            "automation_run_id": automation_run_id,
+            "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
+            "active_blockers": (
+                [] if evidence_ready else ["predecision_mcp_or_log_evidence_missing"]
+            ),
+        }
+    except Exception as exc:
+        logger.warning(
+            "iwooos_wazuh_manager_posture_candidate_failed",
+            error_type=type(exc).__name__,
+        )
+        return {
+            "status": "blocked_wazuh_manager_posture_candidate_error",
+            "queued": 0,
+            "existing": 0,
+            "evidence_ready": 0,
+            "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
+            "active_blockers": [f"candidate_error:{type(exc).__name__}"],
+        }
+
+
 async def enqueue_missing_ansible_candidates_once(
     *,
     project_id: str = "awoooi",
@@ -384,7 +563,9 @@ async def enqueue_missing_ansible_candidates_once(
             "error": None,
         }
 
-    bounded_limit = max(1, limit or settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT)
+    bounded_limit = max(
+        1, limit or settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT
+    )
     bounded_window_hours = max(
         1,
         window_hours or settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS,
@@ -525,13 +706,16 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None:
         batch_limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT,
         window_hours=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS,
     )
-    await asyncio.sleep(settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS)
+    await asyncio.sleep(
+        settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS
+    )
 
     while True:
         sleep_seconds = float(
             settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS
         )
         try:
+            posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once()
             result = await enqueue_missing_ansible_candidates_once(
                 limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT,
                 window_hours=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS,
@@ -544,8 +728,15 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None:
                 or result.get("failed_apply_retry_priority_tick")
             ):
                 logger.info("awooop_ansible_candidate_backfill_worker_tick", **result)
+            if posture_result.get("queued") or posture_result.get("active_blockers"):
+                logger.info(
+                    "iwooos_wazuh_manager_posture_candidate_tick",
+                    **posture_result,
+                )
         except Exception as exc:
-            logger.warning("awooop_ansible_candidate_backfill_worker_failed", error=str(exc))
+            logger.warning(
+                "awooop_ansible_candidate_backfill_worker_failed", error=str(exc)
+            )
             sleep_seconds = _error_backoff_seconds()
 
         await asyncio.sleep(sleep_seconds)

apps/api/src/services/awooop_ansible_audit_service.py

@@ -36,6 +36,26 @@ ANSIBLE_OPERATION_TYPES = frozenset({
 })
 
 _CATALOG: tuple[dict[str, Any], ...] = (
+    {
+        "catalog_id": "ansible:wazuh-manager-posture-readback",
+        "playbook_path": "infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+        "check_mode_playbook_path": "infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+        "inventory_hosts": ["host_112"],
+        "domains": ["wazuh", "siem", "manager_posture", "security"],
+        "keywords": [
+            "wazuh",
+            "wazuhmanager",
+            "wazuh-manager",
+            "siem",
+            "fim",
+            "security manager",
+        ],
+        "supports_check_mode": True,
+        "auto_apply_enabled": True,
+        "approval_required": False,
+        "risk_level": "low",
+        "no_write_only": True,
+    },
     {
         "catalog_id": "ansible:110-devops",
         "playbook_path": "infra/ansible/playbooks/110-devops.yml",
@@ -549,6 +569,15 @@ def build_ansible_decision_audit_payload(
     input_payload = {
         "incident_id": incident_id,
         "project_id": project_id,
+        "trace_id": str(incident_payload.get("trace_id") or ""),
+        "run_id": str(incident_payload.get("run_id") or ""),
+        "work_item_id": str(incident_payload.get("work_item_id") or ""),
+        "source_receipt_ref": str(incident_payload.get("source_receipt_ref") or ""),
+        "asset_scope_aliases": [
+            str(alias)
+            for alias in incident_payload.get("asset_scope_aliases") or []
+            if isinstance(alias, str) and alias
+        ][:20],
         "executor": "ansible",
         "execution_backend": "ansible",
         "single_writer_executor": "awoooi-ansible-executor-broker",
@@ -681,7 +710,7 @@ async def record_ansible_decision_audit(
             )
             if existing.scalar() is not None:
                 return False
-            await db.execute(
+            inserted = await db.execute(
                 text("""
                     INSERT INTO automation_operation_log (
                         op_id, operation_type, actor, status, incident_id,
@@ -697,6 +726,8 @@ async def record_ansible_decision_audit(
                         CAST(:dry_run_result AS jsonb),
                         :tags
                     )
+                    ON CONFLICT (op_id) DO NOTHING
+                    RETURNING op_id::text
                 """),
                 {
                     "operation_type": payload["operation_type"],
@@ -710,7 +741,7 @@ async def record_ansible_decision_audit(
                     "tags": payload["tags"],
                 },
             )
-            return True
+            return inserted.scalar() is not None
     except Exception as exc:
         logger.warning(
             "ansible_decision_audit_write_failed",

apps/api/src/services/iwooos_wazuh_controlled_executor_handoff.py

@@ -308,6 +308,7 @@ def _candidate(summary: dict[str, Any]) -> dict[str, Any]:
         "runtime_write_performed": False,
         "side_effect_count": 0,
         "required_inputs": [
+            "work_item_id",
             "redacted_wazuh_event_receipt_ref",
             "redacted_forensics_evidence_ref",
             "incident_case_ref",
@@ -334,6 +335,7 @@ def _packet_field_bindings(packet: dict[str, Any]) -> list[dict[str, str]]:
     allowed_fields = (
         "trace_id",
         "run_id",
+        "work_item_id",
         "wazuh_event_receipt_ref",
         "forensics_evidence_ref",
         "incident_case_ref",

apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py

@@ -0,0 +1,452 @@
+"""Live Wazuh controlled-executor runtime receipt readback.
+
+The dedicated execution broker owns the SSH/Ansible boundary. This service only reads
+public-safe receipt metadata from the durable production tables; it never
+returns command output, inventory identities, hostnames, addresses, secrets,
+or raw Wazuh payloads.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import text
+
+from src.core.config import settings
+from src.db.base import get_db_context
+from src.services.ai_automation_runtime_contract import (
+    AI_AUTOMATION_REQUIRED_LOOP_STAGES,
+)
+from src.services.awooop_ansible_audit_service import get_ansible_catalog_item
+
+SCHEMA_VERSION = "iwooos_wazuh_controlled_executor_runtime_readback_v1"
+CATALOG_ID = "ansible:wazuh-manager-posture-readback"
+WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE"
+
+_LIVE_RUNTIME_SQL = """
+    WITH latest_candidate AS (
+        SELECT
+            op_id::text AS candidate_op_id,
+            status AS candidate_status,
+            input ->> 'automation_run_id' AS automation_run_id,
+            input ->> 'trace_id' AS trace_id,
+            input ->> 'run_id' AS run_id,
+            input ->> 'work_item_id' AS work_item_id,
+            input ->> 'source_receipt_ref' AS source_receipt_ref,
+            jsonb_array_length(
+                coalesce(input -> 'asset_scope_aliases', '[]'::jsonb)
+            ) AS asset_scope_alias_count,
+            coalesce(incident_id::text, input ->> 'incident_id') AS incident_id,
+            created_at AS candidate_created_at
+        FROM automation_operation_log
+        WHERE operation_type = 'ansible_candidate_matched'
+          AND input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb)
+        ORDER BY created_at DESC
+        LIMIT 1
+    ), latest_check AS (
+        SELECT
+            check_mode.op_id::text AS check_op_id,
+            check_mode.status AS check_status,
+            coalesce(
+                check_mode.output ->> 'returncode',
+                check_mode.dry_run_result ->> 'returncode'
+            ) AS check_returncode,
+            check_mode.duration_ms AS check_duration_ms,
+            check_mode.created_at AS check_created_at
+        FROM automation_operation_log check_mode
+        JOIN latest_candidate candidate
+          ON check_mode.parent_op_id::text = candidate.candidate_op_id
+        WHERE check_mode.operation_type = 'ansible_check_mode_executed'
+        ORDER BY check_mode.created_at DESC
+        LIMIT 1
+    ), latest_apply AS (
+        SELECT
+            apply.op_id::text AS apply_op_id,
+            apply.status AS apply_status,
+            coalesce(
+                apply.output ->> 'returncode',
+                apply.dry_run_result ->> 'returncode'
+            ) AS apply_returncode,
+            apply.duration_ms AS apply_duration_ms,
+            apply.input ->> 'automation_run_id' AS apply_automation_run_id,
+            apply.input -> 'runtime_stage_receipts' AS runtime_stage_receipts,
+            apply.created_at AS apply_created_at
+        FROM automation_operation_log apply
+        JOIN latest_check check_mode
+          ON apply.parent_op_id::text = check_mode.check_op_id
+        WHERE apply.operation_type = 'ansible_apply_executed'
+        ORDER BY apply.created_at DESC
+        LIMIT 1
+    ), runtime_stages AS (
+        SELECT coalesce(
+            array_agg(DISTINCT receipt.value ->> 'stage_id') FILTER (
+                WHERE receipt.value ->> 'stage_id' IS NOT NULL
+                  AND receipt.value ->> 'schema_version' = 'ai_automation_stage_receipt_v1'
+                  AND coalesce((receipt.value ->> 'durable_receipt')::boolean, false)
+            ),
+            ARRAY[]::text[]
+        ) AS stage_ids
+        FROM latest_apply apply
+        LEFT JOIN LATERAL jsonb_array_elements(
+            coalesce(apply.runtime_stage_receipts, '[]'::jsonb)
+        ) receipt(value) ON TRUE
+    ), latest_verifier AS (
+        SELECT
+            evidence.id AS verifier_id,
+            coalesce(evidence.verification_result, 'missing') AS verification_result,
+            evidence.post_execution_state ->> 'automation_run_id' AS verifier_run_id,
+            evidence.collected_at AS verifier_created_at
+        FROM incident_evidence evidence
+        JOIN latest_apply apply
+          ON evidence.post_execution_state ->> 'apply_op_id' = apply.apply_op_id
+        ORDER BY evidence.collected_at DESC
+        LIMIT 1
+    ), latest_km AS (
+        SELECT
+            knowledge.id AS km_entry_id,
+            knowledge.status::text AS km_status,
+            (knowledge.embedding IS NOT NULL) AS km_embedding_persisted,
+            knowledge.created_at AS km_created_at
+        FROM knowledge_entries knowledge
+        JOIN latest_apply apply
+          ON knowledge.path_type = (
+              'ansible_apply_receipt:' || left(apply.apply_op_id, 8)
+          )
+        JOIN latest_candidate candidate
+          ON knowledge.related_incident_id = candidate.incident_id
+        WHERE knowledge.project_id = :project_id
+        ORDER BY knowledge.created_at DESC
+        LIMIT 1
+    ), latest_learning AS (
+        SELECT
+            learning.op_id::text AS learning_op_id,
+            learning.status AS learning_status,
+            coalesce(
+                (
+                    learning.output #>> '{playbook_trust_writeback,durable_write_acknowledged}'
+                )::boolean,
+                false
+            ) AS playbook_trust_acknowledged,
+            learning.created_at AS learning_created_at
+        FROM automation_operation_log learning
+        JOIN latest_apply apply
+          ON learning.parent_op_id::text = apply.apply_op_id
+        WHERE learning.operation_type = 'ansible_learning_writeback_recorded'
+        ORDER BY learning.created_at DESC
+        LIMIT 1
+    ), latest_auto_repair AS (
+        SELECT
+            execution.id::text AS auto_repair_receipt_id,
+            execution.success AS auto_repair_success,
+            execution.created_at AS auto_repair_created_at
+        FROM auto_repair_executions execution
+        JOIN latest_apply apply
+          ON execution.executed_steps::text LIKE ('%' || apply.apply_op_id || '%')
+        WHERE execution.triggered_by = 'ansible_controlled_apply'
+        ORDER BY execution.created_at DESC
+        LIMIT 1
+    ), latest_telegram AS (
+        SELECT
+            outbound.message_id::text AS telegram_receipt_id,
+            outbound.send_status AS telegram_send_status,
+            outbound.sent_at AS telegram_sent_at
+        FROM awooop_outbound_message outbound
+        JOIN latest_candidate candidate
+          ON 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.candidate_op_id
+        WHERE outbound.project_id = :project_id
+          AND outbound.channel_type = 'telegram'
+          AND outbound.source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
+        ORDER BY outbound.queued_at DESC
+        LIMIT 1
+    )
+    SELECT
+        candidate.*,
+        check_mode.*,
+        apply.*,
+        runtime_stages.stage_ids,
+        verifier.*,
+        km.*,
+        learning.*,
+        auto_repair.*,
+        telegram.*
+    FROM latest_candidate candidate
+    LEFT JOIN latest_check check_mode ON TRUE
+    LEFT JOIN latest_apply apply ON TRUE
+    LEFT JOIN runtime_stages ON TRUE
+    LEFT JOIN latest_verifier verifier ON TRUE
+    LEFT JOIN latest_km km ON TRUE
+    LEFT JOIN latest_learning learning ON TRUE
+    LEFT JOIN latest_auto_repair auto_repair ON TRUE
+    LEFT JOIN latest_telegram telegram ON TRUE
+"""
+
+
+async def load_latest_iwooos_wazuh_controlled_executor_runtime(
+    *, project_id: str = "awoooi"
+) -> dict[str, Any]:
+    """Read the latest Wazuh posture executor chain from durable receipts."""
+
+    try:
+        async with get_db_context(project_id) as db:
+            result = await db.execute(
+                text(_LIVE_RUNTIME_SQL),
+                {
+                    "catalog_match": json.dumps([{"catalog_id": CATALOG_ID}]),
+                    "project_id": project_id,
+                },
+            )
+            row = result.mappings().first()
+    except Exception as exc:
+        return build_iwooos_wazuh_controlled_executor_runtime_readback(
+            None,
+            read_error_type=type(exc).__name__,
+        )
+    return build_iwooos_wazuh_controlled_executor_runtime_readback(row)
+
+
+def build_iwooos_wazuh_controlled_executor_runtime_readback(
+    row: Mapping[str, Any] | None,
+    *,
+    read_error_type: str | None = None,
+    executor_enabled: bool | None = None,
+) -> dict[str, Any]:
+    """Build a public-safe runtime summary from one projected DB row."""
+
+    data = dict(row or {})
+    enabled = (
+        settings.ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR
+        if executor_enabled is None
+        else executor_enabled
+    )
+    catalog_ready = get_ansible_catalog_item(CATALOG_ID) is not None
+    candidate_present = bool(data.get("candidate_op_id"))
+    check_passed = (
+        data.get("check_status") == "success"
+        and str(data.get("check_returncode") or "") == "0"
+    )
+    apply_terminal = data.get("apply_status") in {"success", "failed"}
+    apply_passed = (
+        data.get("apply_status") == "success"
+        and str(data.get("apply_returncode") or "") == "0"
+    )
+    verifier_passed = data.get("verification_result") == "success" and str(
+        data.get("verifier_run_id") or ""
+    ) == str(data.get("candidate_op_id") or "")
+    auto_repair_ready = data.get("auto_repair_success") is True
+    km_ready = bool(data.get("km_entry_id"))
+    stage_ids = {str(stage_id) for stage_id in data.get("stage_ids") or [] if stage_id}
+    rag_ready = "rag_writeback" in stage_ids
+    trust_ready = bool(data.get("playbook_trust_acknowledged")) and (
+        "playbook_trust" in stage_ids
+    )
+    telegram_ready = data.get("telegram_send_status") == "sent"
+    incident_closed = all(
+        (
+            apply_passed,
+            verifier_passed,
+            auto_repair_ready,
+            km_ready,
+            telegram_ready,
+        )
+    )
+
+    present_stages = set(stage_ids)
+    for stage_id, present in (
+        ("candidate", candidate_present),
+        ("check_mode", check_passed),
+        ("controlled_apply", apply_passed),
+        ("auto_repair_execution_receipt", auto_repair_ready),
+        ("post_apply_verifier", verifier_passed),
+        ("km_playbook_writeback", km_ready),
+        ("telegram_receipt", telegram_ready),
+        ("incident_closure", incident_closed),
+    ):
+        if present:
+            present_stages.add(stage_id)
+
+    required_stages = list(AI_AUTOMATION_REQUIRED_LOOP_STAGES)
+    eligible_present_stages = [
+        stage_id for stage_id in required_stages if stage_id in present_stages
+    ]
+    missing_stage_ids = [
+        stage_id for stage_id in required_stages if stage_id not in present_stages
+    ]
+    runtime_closed = bool(apply_passed and not missing_stage_ids)
+    required_count = len(required_stages)
+    present_count = len(eligible_present_stages)
+    completion_percent = round((present_count / required_count) * 100)
+    if not runtime_closed:
+        completion_percent = min(completion_percent, 99)
+
+    active_blockers: list[str] = []
+    if read_error_type:
+        active_blockers.append(f"runtime_receipt_read_unavailable:{read_error_type}")
+    if not enabled:
+        active_blockers.append("wazuh_manager_posture_executor_disabled")
+    if not catalog_ready:
+        active_blockers.append("wazuh_manager_posture_catalog_missing")
+    if candidate_present and data.get("check_status") == "failed":
+        active_blockers.append("wazuh_manager_posture_check_mode_failed")
+    if data.get("apply_status") == "failed":
+        active_blockers.append("wazuh_manager_posture_controlled_probe_failed")
+    if apply_passed and not verifier_passed:
+        active_blockers.append("independent_post_verifier_missing_or_failed")
+    if verifier_passed and not (km_ready and rag_ready and trust_ready):
+        active_blockers.append("km_rag_playbook_writeback_incomplete")
+    if verifier_passed and "mcp_context" not in stage_ids:
+        active_blockers.append("same_run_mcp_context_missing")
+    if verifier_passed and not telegram_ready:
+        active_blockers.append("same_run_telegram_receipt_missing")
+
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "status": _status(
+            read_error_type=read_error_type,
+            candidate_present=candidate_present,
+            check_passed=check_passed,
+            apply_terminal=apply_terminal,
+            apply_passed=apply_passed,
+            runtime_closed=runtime_closed,
+        ),
+        "mode": "live_durable_receipt_readback_no_raw_payload_no_secret",
+        "work_item_id": str(data.get("work_item_id") or WORK_ITEM_ID),
+        "trace": {
+            "trace_id": str(data.get("trace_id") or ""),
+            "run_id": str(data.get("run_id") or ""),
+            "automation_run_id": str(
+                data.get("automation_run_id") or data.get("candidate_op_id") or ""
+            ),
+            "same_run_correlation_required": True,
+        },
+        "summary": {
+            "executor_policy_enabled_count": 1 if enabled and catalog_ready else 0,
+            "dispatch_candidate_count": 1 if candidate_present else 0,
+            "check_mode_passed_count": 1 if check_passed else 0,
+            "runtime_execution_performed_count": 1 if apply_terminal else 0,
+            "bounded_posture_execution_passed_count": 1 if apply_passed else 0,
+            "independent_post_verifier_passed_count": 1 if verifier_passed else 0,
+            "auto_repair_execution_receipt_count": 1 if auto_repair_ready else 0,
+            "km_writeback_count": 1 if km_ready else 0,
+            "rag_writeback_count": 1 if rag_ready else 0,
+            "playbook_trust_writeback_count": 1 if trust_ready else 0,
+            "mcp_context_receipt_count": 1 if "mcp_context" in stage_ids else 0,
+            "service_log_evidence_receipt_count": (
+                1 if "service_log_evidence" in stage_ids else 0
+            ),
+            "timeline_receipt_count": 1 if "timeline_projection" in stage_ids else 0,
+            "telegram_receipt_count": 1 if telegram_ready else 0,
+            "same_run_required_stage_count": required_count,
+            "same_run_present_stage_count": present_count,
+            "same_run_missing_stage_count": len(missing_stage_ids),
+            "same_run_completion_percent": completion_percent,
+            "runtime_closed_count": 1 if runtime_closed else 0,
+            "host_write_performed_count": 0,
+            "wazuh_active_response_performed_count": 0,
+            "secret_value_collection_count": 0,
+        },
+        "runtime_chain": [
+            {
+                "stage_id": stage_id,
+                "present": stage_id in present_stages,
+            }
+            for stage_id in required_stages
+        ],
+        "missing_stage_ids": missing_stage_ids,
+        "next_safe_action": _next_action(
+            candidate_present=candidate_present,
+            check_passed=check_passed,
+            apply_terminal=apply_terminal,
+            apply_passed=apply_passed,
+            verifier_passed=verifier_passed,
+            runtime_closed=runtime_closed,
+            missing_stage_ids=missing_stage_ids,
+        ),
+        "active_blockers": active_blockers,
+        "latest_receipt_at": _latest_receipt_at(data),
+        "boundaries": {
+            "runtime_execution_performed": apply_terminal,
+            "bounded_no_write_posture_probe": True,
+            "host_write_performed": False,
+            "wazuh_active_response_performed": False,
+            "live_wazuh_api_query_performed": False,
+            "raw_wazuh_payload_persisted": False,
+            "command_output_returned": False,
+            "inventory_identity_returned": False,
+            "secret_value_collection_allowed": False,
+            "github_api_used": False,
+        },
+        "source_refs": [
+            "infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+            "apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py",
+            "apps/api/src/services/awooop_ansible_check_mode_service.py",
+            "apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py",
+        ],
+    }
+
+
+def _status(
+    *,
+    read_error_type: str | None,
+    candidate_present: bool,
+    check_passed: bool,
+    apply_terminal: bool,
+    apply_passed: bool,
+    runtime_closed: bool,
+) -> str:
+    if read_error_type:
+        return "degraded_runtime_receipt_read_unavailable"
+    if runtime_closed:
+        return "wazuh_manager_posture_same_run_runtime_closed"
+    if apply_terminal and not apply_passed:
+        return "wazuh_manager_posture_execution_failed_waiting_retry_or_repair"
+    if apply_passed:
+        return "wazuh_manager_posture_executed_runtime_writeback_open"
+    if check_passed:
+        return "wazuh_manager_posture_check_passed_waiting_bounded_execution"
+    if candidate_present:
+        return "wazuh_manager_posture_dispatched_waiting_check_mode"
+    return "waiting_for_scheduled_wazuh_manager_posture_dispatch"
+
+
+def _next_action(
+    *,
+    candidate_present: bool,
+    check_passed: bool,
+    apply_terminal: bool,
+    apply_passed: bool,
+    verifier_passed: bool,
+    runtime_closed: bool,
+    missing_stage_ids: list[str],
+) -> str:
+    if runtime_closed:
+        return "keep_scheduled_wazuh_manager_posture_runtime_receipts_fresh"
+    if not candidate_present:
+        return "internal_worker_enqueue_wazuh_manager_posture_candidate"
+    if not check_passed:
+        return "worker_claim_and_run_allowlisted_ansible_check_mode"
+    if not apply_terminal:
+        return "run_bounded_no_write_wazuh_manager_posture_execution"
+    if not apply_passed:
+        return "classify_transport_or_manager_failure_then_bounded_retry"
+    if not verifier_passed:
+        return "write_independent_post_verifier_receipt"
+    return "backfill_same_run_receipts:" + ",".join(missing_stage_ids[:6])
+
+
+def _latest_receipt_at(data: Mapping[str, Any]) -> str | None:
+    values = [
+        value
+        for key, value in data.items()
+        if key.endswith("_at") and value is not None
+    ]
+    if not values:
+        return None
+    latest = max(values)
+    return latest.isoformat() if isinstance(latest, datetime) else str(latest)

apps/api/src/services/iwooos_wazuh_incident_response_closure.py

@@ -43,6 +43,7 @@ _REQUIRED_SAME_RUN_STAGE_IDS = (
 _REQUIRED_PACKET_FIELDS = (
     "trace_id",
     "run_id",
+    "work_item_id",
     "asset_scope_aliases",
     "wazuh_event_receipt_ref",
     "forensics_evidence_ref",