diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py
index 4a32a73cb..f5f6fa058 100644
--- a/apps/api/src/services/awooop_ansible_audit_service.py
+++ b/apps/api/src/services/awooop_ansible_audit_service.py
@@ -1771,6 +1771,7 @@ def build_ansible_decision_audit_payload(
"auto_apply_enabled": row["auto_apply_enabled"],
"approval_required": row["approval_required"],
"risk_level": row["risk_level"],
+ "no_write_only": row.get("no_write_only") is True,
"catalog_revision": row.get("catalog_revision"),
"canonical_asset_id": row.get("canonical_asset_id"),
"typed_domain": row.get("typed_domain"),
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 762cf8e2a..26edec07d 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -346,6 +346,7 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
"auto_apply_enabled": bool(catalog_item.get("auto_apply_enabled") is True),
"approval_required": bool(catalog_item.get("approval_required") is True),
"break_glass_required": bool(catalog_item.get("break_glass_required") is True),
+ "no_write_only": bool(catalog_item.get("no_write_only") is True),
"canonical_asset_id": (
typed_route.get("canonical_asset_id")
if typed_schema
@@ -589,6 +590,13 @@ def _allowed_controlled_apply_risks() -> set[str]:
return {item.strip().lower() for item in raw.split(",") if item.strip()}
+def _claim_is_no_write_observation(claim: AnsibleCheckModeClaim) -> bool:
+ """Classify read-only catalog execution from the current allowlist."""
+
+ catalog_item = get_ansible_catalog_item(claim.catalog_id) or {}
+ return catalog_item.get("no_write_only") is True
+
+
def _controlled_apply_allowed(candidate: dict[str, Any]) -> tuple[bool, str | None]:
risk = str(candidate.get("risk_level") or "").strip().lower()
if settings.ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY is not True:
@@ -764,6 +772,8 @@ def build_ansible_check_mode_claim_input(
"check_mode_playbook_path": safe["check_mode_playbook_path"],
"inventory_hosts": list(safe["inventory_hosts"]),
"risk_level": safe["risk_level"],
+ "no_write_only": safe["no_write_only"],
+ "runtime_write_expected": not safe["no_write_only"],
"canonical_asset_id": safe["canonical_asset_id"],
"typed_domain": safe["typed_domain"],
"target_scope_validated": safe["target_scope_validated"],
@@ -2767,6 +2777,7 @@ async def _read_verified_apply_closure_prerequisites(
automation_run_id = _automation_run_id_for_claim(claim)
approval_id = _approval_id_for_claim(claim)
canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id)
+ telegram_shadow_allowed = _claim_is_no_write_observation(claim)
async with get_db_context(project_id) as db:
result = await db.execute(
text("""
@@ -2835,8 +2846,26 @@ async def _read_verified_apply_closure_prerequisites(
FROM awooop_outbound_message outbound
WHERE outbound.project_id = :project_id
AND outbound.channel_type = 'telegram'
- AND outbound.send_status = 'sent'
- AND outbound.provider_message_id IS NOT NULL
+ AND (
+ (
+ outbound.send_status = 'sent'
+ AND outbound.provider_message_id IS NOT NULL
+ )
+ OR (
+ :telegram_shadow_allowed
+ AND outbound.send_status = 'shadow'
+ AND outbound.provider_message_id IS NULL
+ AND outbound.source_envelope #>>
+ '{notification_policy,disposition}'
+ = 'suppressed'
+ AND coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ )
+ )
AND outbound.source_envelope #>>
'{callback_reply,action}'
= 'controlled_apply_result'
@@ -2888,8 +2917,17 @@ async def _read_verified_apply_closure_prerequisites(
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = :incident_id
- AND lifecycle.event_type::text
- = 'TELEGRAM_RESULT_SENT'
+ AND (
+ lifecycle.event_type::text
+ = 'TELEGRAM_RESULT_SENT'
+ OR (
+ :telegram_shadow_allowed
+ AND lifecycle.event_type::text
+ = 'NOTIFICATION_CLASSIFIED'
+ AND lifecycle.action_detail
+ = 'no_write_posture_result_receipt_suppressed'
+ )
+ )
AND lifecycle.success IS TRUE
AND lifecycle.context ->> 'apply_op_id'
= :apply_op_id
@@ -2916,6 +2954,7 @@ async def _read_verified_apply_closure_prerequisites(
"approval_id": approval_id,
"canonical_playbook_id": canonical_playbook_id,
"km_path_type": f"ansible_apply_receipt:{apply_op_id[:8]}",
+ "telegram_shadow_allowed": telegram_shadow_allowed,
},
)
row = result.mappings().one_or_none()
@@ -3216,6 +3255,7 @@ def _incident_terminal_disposition_payload(
retry_op_id: str | None,
telegram_receipt_acknowledged: bool,
) -> dict[str, Any]:
+ no_write_observation = _claim_is_no_write_observation(claim)
return {
"schema_version": "ansible_incident_terminal_disposition_v1",
"automation_run_id": _automation_run_id_for_claim(claim),
@@ -3225,8 +3265,13 @@ def _incident_terminal_disposition_payload(
"incident_status": "RESOLVED" if success_terminal else "MITIGATING",
"incident_resolved": success_terminal,
"no_write_terminal": not success_terminal,
+ "observation_only": no_write_observation,
+ "repair_performed": success_terminal and not no_write_observation,
+ "runtime_write_performed": success_terminal and not no_write_observation,
"safe_next_action": (
- "keep_verified_repair_and_monitor_recurrence"
+ "retain_posture_receipt_and_notify_only_on_state_transition"
+ if success_terminal and no_write_observation
+ else "keep_verified_repair_and_monitor_recurrence"
if success_terminal
else "queue_ai_playbook_or_transport_repair_candidate"
),
@@ -3249,7 +3294,11 @@ async def _commit_verified_incident_closure(
**_incident_terminal_disposition_payload(
claim,
apply_op_id=apply_op_id,
- terminal_type="controlled_apply_verified_closed",
+ terminal_type=(
+ "no_write_posture_verified_closed"
+ if _claim_is_no_write_observation(claim)
+ else "controlled_apply_verified_closed"
+ ),
success_terminal=True,
retry_op_id=None,
telegram_receipt_acknowledged=True,
@@ -3831,6 +3880,7 @@ async def _retry_telegram_receipt_acknowledged(
project_id: str,
) -> bool:
automation_run_id = _automation_run_id_for_claim(claim)
+ telegram_shadow_allowed = _claim_is_no_write_observation(claim)
try:
async with get_db_context(project_id) as db:
result = await db.execute(
@@ -3840,8 +3890,26 @@ async def _retry_telegram_receipt_acknowledged(
FROM awooop_outbound_message outbound
WHERE outbound.project_id = :project_id
AND outbound.channel_type = 'telegram'
- AND outbound.send_status = 'sent'
- AND outbound.provider_message_id IS NOT NULL
+ AND (
+ (
+ outbound.send_status = 'sent'
+ AND outbound.provider_message_id IS NOT NULL
+ )
+ OR (
+ :telegram_shadow_allowed
+ AND outbound.send_status = 'shadow'
+ AND outbound.provider_message_id IS NULL
+ AND outbound.source_envelope #>>
+ '{notification_policy,disposition}'
+ = 'suppressed'
+ AND coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ )
+ )
AND outbound.source_envelope #>>
'{callback_reply,action}'
= 'controlled_apply_result'
@@ -3868,6 +3936,7 @@ async def _retry_telegram_receipt_acknowledged(
"automation_run_id": automation_run_id,
"incident_id": claim.incident_id,
"apply_op_id": apply_op_id,
+ "telegram_shadow_allowed": telegram_shadow_allowed,
},
)
return result.scalar() is True
@@ -6202,10 +6271,22 @@ async def _send_controlled_apply_telegram_receipt(
try:
from src.services.telegram_gateway import get_telegram_gateway
+ no_write_observation = _claim_is_no_write_observation(claim)
+ verified_success = bool(
+ result.returncode == 0
+ and str(writeback.get("verification_result") or "") == "success"
+ )
+ effective_execution_kind = (
+ "no_write_posture_readback"
+ if no_write_observation and execution_kind == "controlled_apply"
+ else execution_kind
+ )
effective_provider_delivery = str(
claim.input_payload.get("telegram_provider_delivery")
or provider_delivery
)
+ if no_write_observation and verified_success:
+ effective_provider_delivery = "shadow_only"
response = await get_telegram_gateway().send_controlled_apply_result_receipt(
automation_run_id=str(
claim.input_payload.get("automation_run_id")
@@ -6222,7 +6303,7 @@ async def _send_controlled_apply_telegram_receipt(
duration_ms=result.duration_ms,
verifier_written=bool(writeback.get("verification")),
learning_written=bool(writeback.get("learning")),
- execution_kind=execution_kind,
+ execution_kind=effective_execution_kind,
provider_delivery=effective_provider_delivery,
project_id=project_id,
)
@@ -6279,12 +6360,21 @@ async def _reconcile_verified_apply_closure_projections(
)
telegram_lifecycle = False
if telegram_receipt:
+ notification_classified = _claim_is_no_write_observation(claim)
telegram_lifecycle = await _append_alert_lifecycle_receipt(
claim,
- "TELEGRAM_RESULT_SENT",
+ (
+ "NOTIFICATION_CLASSIFIED"
+ if notification_classified
+ else "TELEGRAM_RESULT_SENT"
+ ),
apply_op_id=apply_op_id,
success=False,
- action_detail="controlled_apply_result_receipt_sent",
+ action_detail=(
+ "no_write_posture_failure_digest_recorded"
+ if notification_classified
+ else "controlled_apply_result_receipt_sent"
+ ),
project_id=project_id,
post_verifier_passed=result.post_verifier_passed,
)
@@ -6307,7 +6397,11 @@ async def _reconcile_verified_apply_closure_projections(
"EXECUTION_COMPLETED",
apply_op_id=apply_op_id,
success=True,
- action_detail=f"controlled_apply_completed:{claim.catalog_id}",
+ action_detail=(
+ f"no_write_posture_readback_completed:{claim.catalog_id}"
+ if _claim_is_no_write_observation(claim)
+ else f"controlled_apply_completed:{claim.catalog_id}"
+ ),
project_id=project_id,
post_verifier_passed=True,
)
@@ -6339,12 +6433,21 @@ async def _reconcile_verified_apply_closure_projections(
telegram_lifecycle = False
if telegram_receipt:
+ notification_suppressed = _claim_is_no_write_observation(claim)
telegram_lifecycle = await _append_alert_lifecycle_receipt(
claim,
- "TELEGRAM_RESULT_SENT",
+ (
+ "NOTIFICATION_CLASSIFIED"
+ if notification_suppressed
+ else "TELEGRAM_RESULT_SENT"
+ ),
apply_op_id=apply_op_id,
success=True,
- action_detail="controlled_apply_result_receipt_sent",
+ action_detail=(
+ "no_write_posture_result_receipt_suppressed"
+ if notification_suppressed
+ else "controlled_apply_result_receipt_sent"
+ ),
project_id=project_id,
post_verifier_passed=True,
)
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 d09bb0c4c..0ac1858f1 100644
--- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
@@ -302,6 +302,18 @@ _LIVE_RUNTIME_SQL = """
SELECT
outbound.message_id::text AS telegram_receipt_id,
outbound.send_status AS telegram_send_status,
+ outbound.provider_message_id AS telegram_provider_message_id,
+ coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->> 'execution_kind'
+ ) AS telegram_execution_kind,
+ outbound.source_envelope #>>
+ '{notification_policy,disposition}'
+ AS telegram_notification_disposition,
+ outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ AS telegram_provider_delivery,
outbound.sent_at AS telegram_sent_at
FROM awooop_outbound_message outbound
JOIN latest_candidate candidate
@@ -610,7 +622,20 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
trust_ready = bool(data.get("playbook_trust_acknowledged")) and (
"playbook_trust" in stage_ids
)
- telegram_ready = data.get("telegram_send_status") == "sent"
+ telegram_provider_sent = bool(
+ data.get("telegram_send_status") == "sent"
+ and data.get("telegram_provider_message_id")
+ )
+ telegram_notification_suppressed = bool(
+ not ingress_mode
+ and data.get("telegram_send_status") == "shadow"
+ and not data.get("telegram_provider_message_id")
+ and data.get("telegram_execution_kind")
+ == "no_write_posture_readback"
+ and data.get("telegram_notification_disposition") == "suppressed"
+ and data.get("telegram_provider_delivery") == "shadow_only"
+ )
+ telegram_ready = telegram_provider_sent or telegram_notification_suppressed
incident_closed = all(
(
apply_passed,
@@ -748,6 +773,10 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
),
"timeline_receipt_count": 1 if "timeline_projection" in stage_ids else 0,
"telegram_receipt_count": 1 if telegram_ready else 0,
+ "telegram_provider_send_count": 1 if telegram_provider_sent else 0,
+ "telegram_notification_suppressed_count": (
+ 1 if telegram_notification_suppressed else 0
+ ),
"same_run_identity_chain_verified_count": (
1 if identity_chain_verified else 0
),
@@ -833,6 +862,12 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
"wazuh_active_response_performed": False,
"live_wazuh_api_query_performed": False,
"raw_wazuh_payload_persisted": False,
+ "healthy_posture_telegram_provider_send_performed": (
+ telegram_provider_sent if not ingress_mode else False
+ ),
+ "healthy_posture_notification_suppressed": (
+ telegram_notification_suppressed
+ ),
"command_output_returned": False,
"inventory_identity_returned": False,
"secret_value_collection_allowed": False,
diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
index 80c9daef7..a03a6c80b 100644
--- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
+++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
@@ -226,8 +226,10 @@ _RUNTIME_LIFECYCLE_SQL = """
OR (
telegram_result.event_type
= 'NOTIFICATION_CLASSIFIED'
- AND telegram_result.action_detail
- = 'stdin_boundary_no_write_result_receipt_suppressed'
+ AND telegram_result.action_detail IN (
+ 'stdin_boundary_no_write_result_receipt_suppressed',
+ 'no_write_posture_result_receipt_suppressed'
+ )
)
)
AND telegram_result.context
@@ -253,8 +255,10 @@ _RUNTIME_LIFECYCLE_SQL = """
WHERE event_type = 'TELEGRAM_RESULT_SENT'
OR (
event_type = 'NOTIFICATION_CLASSIFIED'
- AND action_detail
- = 'stdin_boundary_no_write_result_receipt_suppressed'
+ AND action_detail IN (
+ 'stdin_boundary_no_write_result_receipt_suppressed',
+ 'no_write_posture_result_receipt_suppressed'
+ )
)
) AS result_acknowledged_count,
MAX(created_at) AS latest_at
@@ -1000,6 +1004,20 @@ def build_telegram_alert_monitoring_coverage_readback(
runtime_summary.get("automation_runtime_closure_percent") or 0.0
),
"runtime_automation_closure_ready": runtime_closure_ready,
+ "wazuh_posture_provider_sent_count_1h": _int(
+ runtime_summary.get("wazuh_posture_provider_sent_count_1h")
+ ),
+ "wazuh_posture_notification_suppressed_count_1h": _int(
+ runtime_summary.get(
+ "wazuh_posture_notification_suppressed_count_1h"
+ )
+ ),
+ "latest_wazuh_posture_provider_sent_at": runtime_summary.get(
+ "latest_wazuh_posture_provider_sent_at"
+ ),
+ "latest_wazuh_posture_suppressed_at": runtime_summary.get(
+ "latest_wazuh_posture_suppressed_at"
+ ),
"verified_context_receipt_count": _int(
verifier_rollups.get("verified_context_receipt_count")
),
@@ -1274,6 +1292,8 @@ async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]:
"automation_open_run_count_7d": 0,
"automation_runtime_closure_percent": 0.0,
"automation_runtime_closure_ready": False,
+ "wazuh_posture_provider_sent_count_1h": 0,
+ "wazuh_posture_notification_suppressed_count_1h": 0,
"error_type": type(exc).__name__,
},
"event_type_counts_7d": {},
@@ -1310,6 +1330,52 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]:
WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb)
? 'source_refs'
) AS outbound_source_refs_total,
+ COUNT(*) FILTER (
+ WHERE queued_at >= NOW() - INTERVAL '1 hour'
+ AND send_status = 'sent'
+ AND provider_message_id IS NOT NULL
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS wazuh_posture_provider_sent_count_1h,
+ COUNT(*) FILTER (
+ WHERE queued_at >= NOW() - INTERVAL '1 hour'
+ AND send_status = 'shadow'
+ AND provider_message_id IS NULL
+ AND source_envelope #>>
+ '{notification_policy,disposition}' = 'suppressed'
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS wazuh_posture_notification_suppressed_count_1h,
+ MAX(COALESCE(sent_at, queued_at)) FILTER (
+ WHERE send_status = 'sent'
+ AND provider_message_id IS NOT NULL
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS latest_wazuh_posture_provider_sent_at,
+ MAX(queued_at) FILTER (
+ WHERE send_status = 'shadow'
+ AND provider_message_id IS NULL
+ AND source_envelope #>>
+ '{notification_policy,disposition}' = 'suppressed'
+ AND source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ ) AS latest_wazuh_posture_suppressed_at,
MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at
FROM awooop_outbound_message
WHERE project_id = :project_id
@@ -1387,6 +1453,52 @@ async def _load_runtime_log_readback_direct(
WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb)
? 'source_refs'
) AS outbound_source_refs_total,
+ COUNT(*) FILTER (
+ WHERE queued_at >= NOW() - INTERVAL '1 hour'
+ AND send_status = 'sent'
+ AND provider_message_id IS NOT NULL
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS wazuh_posture_provider_sent_count_1h,
+ COUNT(*) FILTER (
+ WHERE queued_at >= NOW() - INTERVAL '1 hour'
+ AND send_status = 'shadow'
+ AND provider_message_id IS NULL
+ AND source_envelope #>>
+ '{notification_policy,disposition}' = 'suppressed'
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS wazuh_posture_notification_suppressed_count_1h,
+ MAX(COALESCE(sent_at, queued_at)) FILTER (
+ WHERE send_status = 'sent'
+ AND provider_message_id IS NOT NULL
+ AND (
+ source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ OR content LIKE
+ '%ansible:wazuh-manager-posture-readback%'
+ )
+ ) AS latest_wazuh_posture_provider_sent_at,
+ MAX(queued_at) FILTER (
+ WHERE send_status = 'shadow'
+ AND provider_message_id IS NULL
+ AND source_envelope #>>
+ '{notification_policy,disposition}' = 'suppressed'
+ AND source_envelope #>>
+ '{callback_reply,catalog_id}' =
+ 'ansible:wazuh-manager-posture-readback'
+ ) AS latest_wazuh_posture_suppressed_at,
MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at
FROM awooop_outbound_message
WHERE project_id = $1
@@ -1501,6 +1613,24 @@ def _runtime_log_readback_from_rows(
"outbound_source_refs_total_7d": _int(
outbound_row.get("outbound_source_refs_total")
),
+ "wazuh_posture_provider_sent_count_1h": _int(
+ outbound_row.get("wazuh_posture_provider_sent_count_1h")
+ ),
+ "wazuh_posture_notification_suppressed_count_1h": _int(
+ outbound_row.get(
+ "wazuh_posture_notification_suppressed_count_1h"
+ )
+ ),
+ "latest_wazuh_posture_provider_sent_at": (
+ str(outbound_row.get("latest_wazuh_posture_provider_sent_at"))
+ if outbound_row.get("latest_wazuh_posture_provider_sent_at")
+ else None
+ ),
+ "latest_wazuh_posture_suppressed_at": (
+ str(outbound_row.get("latest_wazuh_posture_suppressed_at"))
+ if outbound_row.get("latest_wazuh_posture_suppressed_at")
+ else None
+ ),
"latest_alert_operation_log_at": latest_event_at,
"latest_outbound_telegram_at": str(outbound_row.get("latest_outbound_at"))
if outbound_row.get("latest_outbound_at")
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index a92f66dcf..b3cf76e9f 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -12653,29 +12653,32 @@ class TelegramGateway:
"""Send and mirror the AI Agent controlled-apply result receipt."""
no_write_replay = execution_kind == "no_write_replay"
+ no_write_posture = execution_kind == "no_write_posture_readback"
success = (
not no_write_replay
and verification_result == "success"
and returncode == 0
)
- next_step = (
- "monitor_for_regression"
- if success
- else (
- "queue_ai_transport_or_playbook_repair"
- if no_write_replay
- else "queue_ai_rollback_or_playbook_repair"
+ if no_write_posture:
+ next_step = (
+ "record_healthy_posture_without_notification"
+ if success
+ else "queue_ai_transport_or_playbook_repair"
)
- )
- title = (
- "CONTROLLED APPLY RESULT｜AI Agent 受控執行完成"
- if success
- else (
- "AI 自動修復摘要｜PlayBook 檢查待修復"
- if no_write_replay
- else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復"
+ title = (
+ "AI 姿態讀回｜健康狀態已驗證"
+ if success
+ else "AI 自動修復摘要｜姿態檢查待修復"
)
- )
+ elif success:
+ next_step = "monitor_for_regression"
+ title = "CONTROLLED APPLY RESULT｜AI Agent 受控執行完成"
+ elif no_write_replay:
+ next_step = "queue_ai_transport_or_playbook_repair"
+ title = "AI 自動修復摘要｜PlayBook 檢查待修復"
+ else:
+ next_step = "queue_ai_rollback_or_playbook_repair"
+ title = "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復"
failure_fingerprint = (
_no_write_replay_failure_fingerprint(
catalog_id=catalog_id,
@@ -12684,7 +12687,7 @@ class TelegramGateway:
returncode=returncode,
next_step=next_step,
)
- if no_write_replay
+ if no_write_replay or (no_write_posture and not success)
else ""
)
truth_chain: dict[str, object] | None = None
@@ -12732,10 +12735,45 @@ class TelegramGateway:
callback_reply["apply_op_id"] = apply_op_id
callback_reply["project_id"] = project_id or "awoooi"
callback_reply["execution_kind"] = execution_kind
+ callback_reply["catalog_id"] = catalog_id
+ callback_reply["runtime_write_performed"] = not (
+ no_write_replay or no_write_posture
+ )
source_refs = source_extra.get("source_refs")
if isinstance(source_refs, dict):
source_refs["automation_run_ids"] = [automation_run_id]
- if no_write_replay:
+ 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",
+ ],
+ }
+ elif no_write_replay:
effective_provider_delivery = (
"shadow_only"
if provider_delivery == "shadow_only"
@@ -12775,7 +12813,36 @@ class TelegramGateway:
"km_rag_playbook",
],
}
- if no_write_replay:
+ if no_write_posture:
+ lines = [
+ f"{html.escape(title)}",
+ "階段: controlled_observe",
+ f"目標: {html.escape(str(catalog_id or '--'))}",
+ (
+ "狀態: posture "
+ f"{html.escape(str(verification_result or 'failed'))} "
+ f"/ rc {html.escape(str(returncode))}"
+ ),
+ (
+ "AI 處置: healthy_receipt_recorded"
+ if success
+ else "AI 處置: transport_or_playbook_repair_queued"
+ ),
+ "Runtime write: 0",
+ (
+ "Receipts: verifier "
+ f"{html.escape(_bool_code(verifier_written))} / KM "
+ f"{html.escape(_bool_code(learning_written))}"
+ ),
+ (
+ "證據: "
+ f"{html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}"
+ ),
+ (
+ "通知: 健康週期讀回不外送；異常相同原因 30 分鐘只通知一次。"
+ ),
+ ]
+ elif no_write_replay:
lines = [
f"{html.escape(title)}",
"階段: controlled_check",
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 e49d73361..87b0ee865 100644
--- a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
+++ b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
@@ -381,3 +381,67 @@ async def test_backfill_repairs_orphan_incident_and_reuses_verifier(
assert result["incident_closure_written"] == 1
incident_ledger.assert_awaited_once()
assert writeback.await_args.kwargs["durable_verifier_receipt"] == receipt
+
+
+@pytest.mark.asyncio
+async def test_wazuh_posture_success_receipt_is_shadow_only_no_write(
+ 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",
+ }
+ monkeypatch.setattr(
+ "src.services.telegram_gateway.get_telegram_gateway",
+ lambda: gateway,
+ )
+
+ acknowledged = await service._send_controlled_apply_telegram_receipt(
+ _wazuh_claim(),
+ service.AnsibleRunResult(
+ returncode=0,
+ stdout="",
+ stderr="",
+ duration_ms=5,
+ post_verifier_passed=True,
+ ),
+ apply_op_id="00000000-0000-0000-0000-000000000203",
+ writeback={
+ "verification_result": "success",
+ "verification": True,
+ "learning": True,
+ },
+ project_id="awoooi",
+ )
+
+ 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"
+
+
+def test_wazuh_posture_shadow_receipt_is_a_terminal_notification_receipt() -> None:
+ closure_source = service._read_verified_apply_closure_prerequisites.__doc__ or ""
+ assert service._claim_is_no_write_observation(_wazuh_claim()) is True
+ assert "durable prerequisite" in closure_source
+
+ import inspect
+
+ readback_source = inspect.getsource(
+ service._read_verified_apply_closure_prerequisites
+ )
+ retry_source = inspect.getsource(
+ service._retry_telegram_receipt_acknowledged
+ )
+ reconcile_source = inspect.getsource(
+ service._reconcile_verified_apply_closure_projections
+ )
+ for source in (readback_source, retry_source):
+ assert "telegram_shadow_allowed" in source
+ assert "no_write_posture_readback" in source
+ assert "send_status = 'shadow'" in source
+ assert "no_write_posture_result_receipt_suppressed" in reconcile_source
+ assert "no_write_posture_failure_digest_recorded" in reconcile_source
+ assert '"NOTIFICATION_CLASSIFIED"' in reconcile_source
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 3eb060dc8..58e021b27 100644
--- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
@@ -122,6 +122,7 @@ def _complete_row() -> dict[str, object]:
"auto_repair_success": True,
"telegram_receipt_id": "telegram-303",
"telegram_send_status": "sent",
+ "telegram_provider_message_id": "303",
"stage_ids": [
"mcp_context",
"service_log_evidence",
@@ -165,6 +166,36 @@ def test_wazuh_runtime_readback_closes_only_with_complete_same_run_receipts() ->
assert payload["boundaries"]["command_output_returned"] is False
+def test_wazuh_posture_shadow_receipt_closes_without_provider_send() -> None:
+ row = _complete_row()
+ row.update(
+ {
+ "telegram_send_status": "shadow",
+ "telegram_provider_message_id": None,
+ "telegram_execution_kind": "no_write_posture_readback",
+ "telegram_notification_disposition": "suppressed",
+ "telegram_provider_delivery": "shadow_only",
+ }
+ )
+
+ payload = build_iwooos_wazuh_controlled_executor_runtime_readback(
+ row,
+ executor_enabled=True,
+ )
+
+ assert payload["summary"]["runtime_closed_count"] == 1
+ assert payload["summary"]["telegram_receipt_count"] == 1
+ assert payload["summary"]["telegram_provider_send_count"] == 0
+ assert payload["summary"]["telegram_notification_suppressed_count"] == 1
+ assert (
+ payload["boundaries"][
+ "healthy_posture_telegram_provider_send_performed"
+ ]
+ is False
+ )
+ assert payload["boundaries"]["healthy_posture_notification_suppressed"] is True
+
+
def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None:
row = _complete_row()
row["catalog_id"] = "ansible:wazuh-alertmanager-integration"
diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
index 2a78f32db..95eb291ca 100644
--- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
+++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
@@ -788,7 +788,13 @@ def test_telegram_alert_runtime_log_calculates_incident_closure() -> None:
"latest_triggered_at": "2026-07-11T02:00:00+00:00",
"latest_closed_at": "2026-07-11T01:59:00+00:00",
},
- outbound_row={},
+ outbound_row={
+ "wazuh_posture_provider_sent_count_1h": 0,
+ "wazuh_posture_notification_suppressed_count_1h": 4,
+ "latest_wazuh_posture_suppressed_at": (
+ "2026-07-18T04:32:00+00:00"
+ ),
+ },
readback_source="test",
)
@@ -798,6 +804,11 @@ def test_telegram_alert_runtime_log_calculates_incident_closure() -> None:
assert summary["automation_open_run_count_7d"] == 1
assert summary["automation_runtime_closure_percent"] == 66.7
assert summary["automation_runtime_closure_ready"] is False
+ assert summary["wazuh_posture_provider_sent_count_1h"] == 0
+ assert summary["wazuh_posture_notification_suppressed_count_1h"] == 4
+ assert summary["latest_wazuh_posture_suppressed_at"] == (
+ "2026-07-18T04:32:00+00:00"
+ )
def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None:
@@ -835,6 +846,7 @@ def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None:
assert "= 'TELEGRAM_RESULT_SENT'" in sql
assert "= 'NOTIFICATION_CLASSIFIED'" in sql
assert "stdin_boundary_no_write_result_receipt_suppressed" in sql
+ assert "no_write_posture_result_receipt_suppressed" in sql
assert "result_acknowledged_count" in sql
assert "completed_success_count" not in sql
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 66498f73a..e3715b42c 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/apps/api/tests/test_telegram_message_templates.py
@@ -1441,6 +1441,130 @@ async def test_no_write_replay_receipt_never_claims_runtime_apply(monkeypatch) -
]
+@pytest.mark.asyncio
+async def test_healthy_posture_readback_is_suppressed_and_never_claims_apply(
+ 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": "suppressed",
+ }
+
+ 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-000000000003",
+ incident_id="IWZ-POSTURE-2026071806",
+ catalog_id="ansible:wazuh-manager-posture-readback",
+ apply_op_id="93b7a95c-3652-4c0d-bb4c-729e500acedb",
+ playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+ verification_result="success",
+ returncode=0,
+ duration_ms=8649,
+ verifier_written=True,
+ learning_written=True,
+ execution_kind="no_write_posture_readback",
+ project_id="awoooi",
+ )
+
+ assert result["ok"] is True
+ method, payload = sent_requests[0]
+ assert method == "sendMessage"
+ assert "AI 姿態讀回｜健康狀態已驗證" in payload["text"]
+ assert "Runtime write: 0" in payload["text"]
+ assert "CONTROLLED APPLY RESULT" not in payload["text"]
+ assert "Runtime apply: 1" not in payload["text"]
+ source_extra = payload["_awooop_source_envelope_extra"]
+ callback = source_extra["callback_reply"]
+ assert callback["execution_kind"] == "no_write_posture_readback"
+ assert callback["catalog_id"] == "ansible:wazuh-manager-posture-readback"
+ assert callback["runtime_write_performed"] is False
+ policy = source_extra["notification_policy"]
+ assert policy["policy_version"] == "telegram_posture_state_digest_v1"
+ assert policy["provider_delivery"] == "shadow_only"
+ assert policy["disposition"] == "suppressed"
+ assert policy["provider_send_performed"] is False
+ assert policy["runtime_write_performed"] is False
+
+
+@pytest.mark.asyncio
+async def test_failed_posture_readback_is_digest_deduplicated_and_queues_repair(
+ monkeypatch,
+) -> None:
+ gateway = TelegramGateway()
+ sent_requests = []
+
+ async def fake_send_request(method, payload):
+ sent_requests.append((method, payload))
+ return {"ok": True, "result": {"message_id": 12347}}
+
+ 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,
+ )
+
+ await gateway.send_controlled_apply_result_receipt(
+ automation_run_id="00000000-0000-0000-0000-000000000004",
+ incident_id="IWZ-POSTURE-2026071812",
+ catalog_id="ansible:wazuh-manager-posture-readback",
+ apply_op_id="a3b7a95c-3652-4c0d-bb4c-729e500acedb",
+ playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
+ verification_result="failed",
+ returncode=2,
+ duration_ms=9000,
+ verifier_written=True,
+ learning_written=True,
+ execution_kind="no_write_posture_readback",
+ project_id="awoooi",
+ )
+
+ payload = sent_requests[0][1]
+ assert "AI 自動修復摘要｜姿態檢查待修復" in payload["text"]
+ assert "transport_or_playbook_repair_queued" in payload["text"]
+ assert "Runtime write: 0" in payload["text"]
+ policy = payload["_awooop_source_envelope_extra"]["notification_policy"]
+ assert policy["provider_delivery"] == "digest"
+ assert policy["digest_window_minutes"] == 30
+ assert len(policy["failure_fingerprint"]) == 24
+
+
def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None:
"""Callback evidence 要保存當下 AwoooP 狀態鏈，不只保存 live query 結果。"""
snapshot = telegram_gateway_module._callback_reply_awooop_status_chain_snapshot(