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 7da5d5ce6..ec740cb2e 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -1094,12 +1094,9 @@ def _claim_is_no_write_observation(claim: AnsibleCheckModeClaim) -> bool:
def _telegram_shadow_completion_allowed(claim: AnsibleCheckModeClaim) -> bool:
- """Keep denoise receipts separate from Wazuh recipient-visible closure."""
+ """Accept a policy-complete no-send receipt for no-write observations."""
- return bool(
- _claim_is_no_write_observation(claim)
- and claim.catalog_id != _WAZUH_MANAGER_POSTURE_CATALOG_ID
- )
+ return _claim_is_no_write_observation(claim)
def _controlled_apply_allowed(candidate: dict[str, Any]) -> tuple[bool, str | None]:
@@ -3913,6 +3910,29 @@ async def _read_verified_apply_closure_prerequisites(
(
outbound.send_status = 'sent'
AND outbound.provider_message_id IS NOT NULL
+ AND (
+ :telegram_shadow_allowed IS FALSE
+ OR (
+ coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ AND outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_lifecycle_transition_v1'
+ AND outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'state_transition'
+ AND outbound.source_envelope #>>
+ '{notification_policy,lifecycle_state}'
+ = 'recovered'
+ AND outbound.source_envelope #>>
+ '{notification_policy,disposition}'
+ = 'sent'
+ )
+ )
)
OR (
:telegram_shadow_allowed
@@ -3921,21 +3941,36 @@ async def _read_verified_apply_closure_prerequisites(
AND outbound.source_envelope #>>
'{notification_policy,disposition}'
= 'suppressed'
- AND nullif(
- outbound.source_envelope #>>
- '{notification_policy,policy_version}',
- ''
- ) IS NOT NULL
AND outbound.source_envelope #>>
- '{notification_policy,provider_delivery}'
- IN (
- 'shadow_only',
- 'state_transition',
- 'digest'
- )
+ '{notification_policy,lifecycle_state}'
+ = 'recovered'
AND outbound.source_envelope #>>
'{notification_policy,provider_send_performed}'
= 'false'
+ AND (
+ (
+ outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_lifecycle_transition_v1'
+ AND outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'state_transition'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'duplicate_lifecycle_state'
+ )
+ OR (
+ outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_state_digest_v1'
+ AND outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'shadow_only'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'historical_projection_backfill'
+ )
+ )
AND coalesce(
outbound.source_envelope #>>
'{callback_reply,execution_kind}',
@@ -4981,25 +5016,76 @@ async def _retry_telegram_receipt_acknowledged(
AND outbound.source_envelope #>>
'{notification_policy,disposition}'
= 'suppressed'
- AND nullif(
- outbound.source_envelope #>>
- '{notification_policy,policy_version}',
- ''
- ) IS NOT NULL
- AND outbound.source_envelope #>>
- '{notification_policy,provider_delivery}'
- IN ('shadow_only', 'digest')
AND outbound.source_envelope #>>
'{notification_policy,provider_send_performed}'
= 'false'
- AND coalesce(
- outbound.source_envelope #>>
- '{callback_reply,execution_kind}',
- outbound.source_envelope ->>
- 'execution_kind'
- ) IN (
- 'no_write_posture_readback',
- 'no_write_replay'
+ AND outbound.source_envelope #>>
+ '{notification_policy,lifecycle_state}'
+ IN ('recovered', 'failed')
+ AND (
+ (
+ coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ AND outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_lifecycle_transition_v1'
+ AND outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'state_transition'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'duplicate_lifecycle_state'
+ )
+ OR (
+ coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ AND outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_state_digest_v1'
+ AND outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'shadow_only'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'historical_projection_backfill'
+ )
+ OR (
+ coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_replay'
+ AND outbound.source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_failure_digest_v2'
+ AND (
+ (
+ outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'digest'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'duplicate_failure_fingerprint_window'
+ )
+ OR (
+ outbound.source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'shadow_only'
+ AND outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'historical_projection_backfill'
+ )
+ )
+ )
)
)
)
@@ -5114,24 +5200,76 @@ async def _record_retry_runtime_stage_receipt(
AND source_envelope #>>
'{notification_policy,disposition}'
= 'suppressed'
- AND nullif(
- source_envelope #>>
- '{notification_policy,policy_version}',
- ''
- ) IS NOT NULL
- AND source_envelope #>>
- '{notification_policy,provider_delivery}'
- IN ('shadow_only', 'digest')
AND source_envelope #>>
'{notification_policy,provider_send_performed}'
= 'false'
- AND coalesce(
- source_envelope #>>
- '{callback_reply,execution_kind}',
- source_envelope ->> 'execution_kind'
- ) IN (
- 'no_write_posture_readback',
- 'no_write_replay'
+ AND source_envelope #>>
+ '{notification_policy,lifecycle_state}'
+ IN ('recovered', 'failed')
+ AND (
+ (
+ coalesce(
+ source_envelope #>>
+ '{callback_reply,execution_kind}',
+ source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ AND source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_lifecycle_transition_v1'
+ AND source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'state_transition'
+ AND source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'duplicate_lifecycle_state'
+ )
+ OR (
+ coalesce(
+ source_envelope #>>
+ '{callback_reply,execution_kind}',
+ source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_posture_readback'
+ AND source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_posture_state_digest_v1'
+ AND source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'shadow_only'
+ AND source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'historical_projection_backfill'
+ )
+ OR (
+ coalesce(
+ source_envelope #>>
+ '{callback_reply,execution_kind}',
+ source_envelope ->>
+ 'execution_kind'
+ ) = 'no_write_replay'
+ AND source_envelope #>>
+ '{notification_policy,policy_version}'
+ = 'telegram_failure_digest_v2'
+ AND (
+ (
+ source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'digest'
+ AND source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'duplicate_failure_fingerprint_window'
+ )
+ OR (
+ source_envelope #>>
+ '{notification_policy,provider_delivery}'
+ = 'shadow_only'
+ AND source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ = 'historical_projection_backfill'
+ )
+ )
+ )
)
)
)
@@ -7448,16 +7586,21 @@ async def _send_controlled_apply_telegram_receipt(
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 = (
- "state_transition"
- if claim.catalog_id == _WAZUH_MANAGER_POSTURE_CATALOG_ID
- else "shadow_only"
+ requested_provider_delivery = str(provider_delivery or "").strip()
+ effective_provider_delivery = (
+ "shadow_only"
+ if requested_provider_delivery == "shadow_only"
+ else str(
+ claim.input_payload.get("telegram_provider_delivery")
+ or requested_provider_delivery
+ or "digest"
)
+ )
+ if no_write_observation and effective_provider_delivery != "shadow_only":
+ if claim.catalog_id == _WAZUH_MANAGER_POSTURE_CATALOG_ID:
+ effective_provider_delivery = "state_transition"
+ elif verified_success:
+ effective_provider_delivery = "shadow_only"
receipt_readback = dict(closure_receipts or {})
response = await get_telegram_gateway().send_controlled_apply_result_receipt(
automation_run_id=str(
@@ -7520,6 +7663,7 @@ async def _reconcile_verified_apply_closure_projections(
apply_op_id: str,
writeback: dict[str, Any],
project_id: str,
+ provider_delivery: str = "digest",
) -> dict[str, Any]:
"""Repair terminal projections without ever executing the apply again."""
@@ -7541,6 +7685,7 @@ async def _reconcile_verified_apply_closure_projections(
apply_op_id=apply_op_id,
writeback=writeback,
project_id=project_id,
+ provider_delivery=provider_delivery,
)
if receipt_finalized:
telegram_receipt = await _retry_telegram_receipt_acknowledged(
@@ -7561,7 +7706,13 @@ async def _reconcile_verified_apply_closure_projections(
apply_op_id=apply_op_id,
success=False,
action_detail=(
- "no_write_posture_failure_digest_recorded"
+ "no_write_posture_failure_transition_recorded"
+ if (
+ notification_classified
+ and claim.catalog_id
+ == _WAZUH_MANAGER_POSTURE_CATALOG_ID
+ )
+ else "no_write_posture_failure_digest_recorded"
if notification_classified
else "controlled_apply_result_receipt_sent"
),
@@ -7615,6 +7766,7 @@ async def _reconcile_verified_apply_closure_projections(
apply_op_id=apply_op_id,
writeback=writeback,
project_id=project_id,
+ provider_delivery=provider_delivery,
closure_receipts=receipt_readback,
)
readback = await _read_verified_apply_closure_prerequisites(
@@ -8009,6 +8161,7 @@ async def backfill_missing_auto_repair_execution_receipts_once(
apply_op_id=str(row.get("op_id") or ""),
writeback=writeback,
project_id=project_id,
+ provider_delivery="shadow_only",
)
if closure.get("telegram_receipt_acknowledged") is True:
stats["telegram_receipt_acknowledged"] += 1
@@ -8699,6 +8852,7 @@ async def backfill_missing_retry_terminal_projections_once(
},
project_id=project_id,
execution_kind="no_write_replay",
+ provider_delivery="shadow_only",
)
if receipt_sent:
telegram_acknowledged = (
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 ac614bf7d..15005384d 100644
--- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
@@ -565,6 +565,12 @@ _LIVE_RUNTIME_SQL = """
outbound.source_envelope #>>
'{notification_policy,policy_version}'
AS telegram_notification_policy_version,
+ outbound.source_envelope #>>
+ '{notification_policy,lifecycle_state}'
+ AS telegram_notification_lifecycle_state,
+ outbound.source_envelope #>>
+ '{notification_policy,suppression_reason}'
+ AS telegram_notification_suppression_reason,
outbound.source_envelope #>>
'{notification_policy,provider_send_performed}'
= 'false' AS telegram_provider_send_suppressed,
@@ -1230,6 +1236,20 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
telegram_provider_sent = bool(
data.get("telegram_send_status") == "sent"
and data.get("telegram_provider_message_id")
+ and (
+ ingress_mode
+ or (
+ data.get("telegram_execution_kind")
+ == "no_write_posture_readback"
+ and data.get("telegram_notification_policy_version")
+ == "telegram_posture_lifecycle_transition_v1"
+ and data.get("telegram_provider_delivery")
+ == "state_transition"
+ and data.get("telegram_notification_lifecycle_state")
+ == "recovered"
+ and data.get("telegram_notification_disposition") == "sent"
+ )
+ )
)
telegram_notification_suppressed = bool(
not ingress_mode
@@ -1238,15 +1258,29 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
and data.get("telegram_execution_kind")
== "no_write_posture_readback"
and data.get("telegram_notification_disposition") == "suppressed"
- and str(data.get("telegram_notification_policy_version") or "")
- and data.get("telegram_provider_delivery")
- in {"shadow_only", "state_transition", "digest"}
and data.get("telegram_provider_send_suppressed") is True
+ and data.get("telegram_notification_lifecycle_state") == "recovered"
+ and (
+ (
+ data.get("telegram_notification_policy_version")
+ == "telegram_posture_lifecycle_transition_v1"
+ and data.get("telegram_provider_delivery")
+ == "state_transition"
+ and data.get("telegram_notification_suppression_reason")
+ == "duplicate_lifecycle_state"
+ )
+ or (
+ data.get("telegram_notification_policy_version")
+ == "telegram_posture_state_digest_v1"
+ and data.get("telegram_provider_delivery") == "shadow_only"
+ and data.get("telegram_notification_suppression_reason")
+ == "historical_projection_backfill"
+ )
+ )
)
- # This P0 slice requires a same-run, recipient-visible provider receipt.
- # A durable suppression row remains useful denoise evidence, but it cannot
- # substitute for Telegram delivery or close the incident.
- telegram_ready = telegram_provider_sent
+ # A policy-complete shadow is the terminal receipt for unchanged healthy
+ # posture. Only actual lifecycle transitions are provider-visible.
+ telegram_ready = telegram_provider_sent or telegram_notification_suppressed
incident_ledger_closure_verified = bool(
str(data.get("incident_status") or "").upper()
in {"RESOLVED", "CLOSED"}
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index bd1b6c3a0..7847f41cd 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -418,6 +418,9 @@ def _telegram_provider_destination_verification_method(
_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS = 3
_NO_WRITE_REPLAY_DIGEST_WINDOW_MINUTES = 30
+_WAZUH_MANAGER_POSTURE_CATALOG_ID = (
+ "ansible:wazuh-manager-posture-readback"
+)
_CONTROLLED_APPLY_LIFECYCLE_PENDING_LEASE_MINUTES = 10
_CONTROLLED_APPLY_LIFECYCLE_DELIVERY_DEADLINE_SECONDS = 90.0
_CONTROLLED_APPLY_LIFECYCLE_DELIVERY_LOCK_TTL_SECONDS = 120
@@ -6943,6 +6946,17 @@ class TelegramGateway:
WHERE project_id = :project_id
AND run_id = CAST(:run_id AS uuid)
AND channel_type = 'telegram'
+ AND (
+ :lifecycle_scope = ''
+ OR (
+ source_envelope #>>
+ '{notification_policy,lifecycle_scope}'
+ = :lifecycle_scope
+ AND source_envelope #>>
+ '{notification_policy,lifecycle_state}'
+ = :lifecycle_state
+ )
+ )
ORDER BY
(
send_status = 'sent'
@@ -6955,6 +6969,8 @@ class TelegramGateway:
{
"project_id": project_id,
"run_id": str(delivery_run_id),
+ "lifecycle_scope": lifecycle_scope,
+ "lifecycle_state": lifecycle_state,
},
)
row = existing.mappings().one_or_none()
@@ -6986,7 +7002,14 @@ class TelegramGateway:
AS lifecycle_state
FROM awooop_outbound_message
WHERE project_id = :project_id
- AND run_id <> CAST(:run_id AS uuid)
+ AND (
+ run_id <> CAST(:run_id AS uuid)
+ OR coalesce(
+ source_envelope #>>
+ '{notification_policy,lifecycle_state}',
+ ''
+ ) <> :lifecycle_state
+ )
AND channel_type = 'telegram'
AND (
send_status = 'sent'
@@ -7011,6 +7034,7 @@ class TelegramGateway:
"project_id": project_id,
"run_id": str(delivery_run_id),
"lifecycle_scope": lifecycle_scope,
+ "lifecycle_state": lifecycle_state,
"pending_lease_minutes": (
_CONTROLLED_APPLY_LIFECYCLE_PENDING_LEASE_MINUTES
),
@@ -13854,6 +13878,12 @@ class TelegramGateway:
no_write_replay = execution_kind == "no_write_replay"
no_write_posture = execution_kind == "no_write_posture_readback"
+ if (
+ no_write_posture
+ and catalog_id == _WAZUH_MANAGER_POSTURE_CATALOG_ID
+ and provider_delivery != "shadow_only"
+ ):
+ provider_delivery = "state_transition"
success = (
not no_write_replay
and verification_result == "success"
@@ -13882,20 +13912,11 @@ class TelegramGateway:
if str(catalog_id).startswith("ansible:")
else "controlled_automation"
)
- same_run_provider_receipt = bool(
- no_write_posture
- and success
- and provider_delivery == "state_transition"
- and catalog_id == "ansible:wazuh-manager-posture-readback"
- )
lifecycle_scope = _controlled_apply_lifecycle_scope(
alert_category=normalized_category,
canonical_asset_id=canonical_asset_id,
catalog_id=catalog_id,
playbook_path=playbook_path,
- scope_discriminator=(
- automation_run_id if same_run_provider_receipt else ""
- ),
)
lifecycle_state = (
"recovered"
@@ -14020,31 +14041,19 @@ class TelegramGateway:
if provider_delivery == "state_transition":
source_extra["notification_policy"] = {
"policy_version": (
- "telegram_wazuh_posture_same_run_receipt_v1"
- if same_run_provider_receipt
- else "telegram_posture_lifecycle_transition_v1"
+ "telegram_posture_lifecycle_transition_v1"
),
"canonical_category": normalized_category,
"lifecycle_scope": lifecycle_scope,
"lifecycle_state": lifecycle_state,
- "group_scope": (
- "controlled_asset_same_run"
- if same_run_provider_receipt
- else "controlled_asset_lifecycle"
- ),
+ "group_scope": "controlled_asset_lifecycle",
"provider_delivery": "state_transition",
- "disposition": (
- "send_once_per_run"
- if same_run_provider_receipt
- else "send_on_transition"
- ),
+ "disposition": "send_on_transition",
"pending_lease_minutes": (
_CONTROLLED_APPLY_LIFECYCLE_PENDING_LEASE_MINUTES
),
"delivery_semantics": (
- "at_least_once_per_run_after_unknown_ack_lease"
- if same_run_provider_receipt
- else "at_least_once_after_unknown_ack_lease"
+ "at_least_once_after_unknown_ack_lease"
),
"provider_send_performed": None,
"runtime_write_performed": False,
@@ -14063,6 +14072,7 @@ class TelegramGateway:
source_extra["notification_policy"] = {
"policy_version": "telegram_posture_state_digest_v1",
"failure_fingerprint": failure_fingerprint,
+ "lifecycle_state": lifecycle_state,
"group_scope": (
"scheduled_posture_healthy"
if success
@@ -14094,6 +14104,7 @@ class TelegramGateway:
source_extra["notification_policy"] = {
"policy_version": "telegram_failure_digest_v2",
"failure_fingerprint": failure_fingerprint,
+ "lifecycle_state": lifecycle_state,
"group_scope": (
"historical_projection_backfill"
if effective_provider_delivery == "shadow_only"
@@ -14176,9 +14187,7 @@ class TelegramGateway:
f"{html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}"
),
(
- "通知: 本次 run 僅外送一次，並保存 recipient/provider durable receipt。"
- if same_run_provider_receipt
- else "通知: 狀態轉換僅外送一次；相同狀態由 durable receipt 去重。"
+ "通知: 狀態轉換僅外送一次；相同狀態由 durable receipt 去重。"
if provider_delivery == "state_transition"
else "通知: 健康週期讀回不外送；異常相同原因 30 分鐘只通知一次。"
),
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 2d8eae810..c8aa67d3e 100644
--- a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
+++ b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py
@@ -484,10 +484,91 @@ async def test_wazuh_posture_success_receipt_is_recipient_visible_transition(
assert call["provider_delivery"] == "state_transition"
-def test_wazuh_posture_shadow_receipt_remains_denoise_not_terminal() -> None:
+@pytest.mark.asyncio
+async def test_wazuh_posture_failure_receipt_is_lifecycle_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": "sent",
+ }
+ monkeypatch.setattr(
+ "src.services.telegram_gateway.get_telegram_gateway",
+ lambda: gateway,
+ )
+
+ acknowledged = await service._send_controlled_apply_telegram_receipt(
+ _wazuh_claim(),
+ service.AnsibleRunResult(
+ returncode=2,
+ stdout="",
+ stderr="failed posture",
+ duration_ms=5,
+ post_verifier_passed=False,
+ ),
+ apply_op_id="00000000-0000-0000-0000-000000000203",
+ writeback={
+ "verification_result": "failed",
+ "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"] == "state_transition"
+
+
+@pytest.mark.asyncio
+async def test_wazuh_posture_backfill_preserves_explicit_shadow_delivery(
+ 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,
+ )
+
+ claim = _wazuh_claim()
+ claim.input_payload["telegram_provider_delivery"] = "state_transition"
+ acknowledged = await service._send_controlled_apply_telegram_receipt(
+ 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",
+ provider_delivery="shadow_only",
+ )
+
+ 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_no_write_terminal() -> None:
closure_source = service._read_verified_apply_closure_prerequisites.__doc__ or ""
assert service._claim_is_no_write_observation(_wazuh_claim()) is True
- assert service._telegram_shadow_completion_allowed(_wazuh_claim()) is False
+ assert service._telegram_shadow_completion_allowed(_wazuh_claim()) is True
assert "durable prerequisite" in closure_source
import inspect
@@ -506,13 +587,23 @@ def test_wazuh_posture_shadow_receipt_remains_denoise_not_terminal() -> None:
assert "no_write_posture_readback" in source
assert "send_status = 'shadow'" in source
assert "_telegram_shadow_completion_allowed" in source
+ assert "telegram_posture_lifecycle_transition_v1" in source
+ assert "duplicate_lifecycle_state" in source
+ assert "telegram_posture_state_digest_v1" in source
+ assert "historical_projection_backfill" in source
+ assert "'{notification_policy,lifecycle_state}'" in readback_source
+ assert "= 'recovered'" in readback_source
+ assert ":telegram_shadow_allowed IS FALSE" in readback_source
+ assert "'{notification_policy,disposition}'" in readback_source
+ assert "= 'sent'" in readback_source
assert "no_write_posture_result_receipt_suppressed" in reconcile_source
assert "no_write_posture_failure_digest_recorded" in reconcile_source
+ assert "no_write_posture_failure_transition_recorded" in reconcile_source
assert '"NOTIFICATION_CLASSIFIED"' in reconcile_source
@pytest.mark.asyncio
-async def test_wazuh_posture_retry_rejects_shadow_receipt_allowance(
+async def test_wazuh_posture_retry_accepts_shadow_receipt_allowance(
monkeypatch: pytest.MonkeyPatch,
) -> None:
db = _SequenceDB(_ScalarResult(False))
@@ -530,4 +621,4 @@ async def test_wazuh_posture_retry_rejects_shadow_receipt_allowance(
)
assert acknowledged is False
- assert db.parameters[0]["telegram_shadow_allowed"] is False
+ assert db.parameters[0]["telegram_shadow_allowed"] is True
diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py
index 9eb09e6c7..1d89eefd1 100644
--- a/apps/api/tests/test_ansible_verified_closure.py
+++ b/apps/api/tests/test_ansible_verified_closure.py
@@ -54,8 +54,14 @@ class _SequenceDB:
self.parameters: list[dict] = []
async def execute(self, statement, parameters=None):
+ provided = dict(parameters or {})
+ required = set(getattr(statement, "_bindparams", {}))
+ missing = required - set(provided)
+ assert not missing, (
+ f"missing SQL bind parameters: {sorted(missing)}"
+ )
self.statements.append(str(statement))
- self.parameters.append(dict(parameters or {}))
+ self.parameters.append(provided)
return self._results.pop(0) if self._results else _MappingResult()
@@ -1844,7 +1850,7 @@ async def test_no_write_group_reservation_persists_shadow_suppression(
@pytest.mark.asyncio
-async def test_same_controlled_apply_lifecycle_state_persists_shadow_receipt(
+async def test_same_wazuh_lifecycle_state_across_runs_persists_shadow_receipt(
monkeypatch: pytest.MonkeyPatch,
) -> None:
db = _SequenceDB(
@@ -1875,17 +1881,28 @@ async def test_same_controlled_apply_lifecycle_state_persists_shadow_receipt(
identity = {
"project_id": "awoooi",
"automation_run_id": "00000000-0000-0000-0000-000000000402",
- "incident_id": "INC-20260722-LIFECYCLE",
+ "incident_id": "IWZ-POSTURE-2026072206",
"apply_op_id": "00000000-0000-0000-0000-000000000404",
}
+ lifecycle_scope = (
+ telegram_gateway_module._controlled_apply_lifecycle_scope(
+ alert_category="security",
+ canonical_asset_id="service:wazuh-manager",
+ catalog_id="ansible:wazuh-manager-posture-readback",
+ playbook_path=(
+ "infra/ansible/playbooks/wazuh-manager-posture-readback.yml"
+ ),
+ )
+ )
source_extra = {
"callback_reply": {
"action": "controlled_apply_result",
+ "execution_kind": "no_write_posture_readback",
**identity,
},
"notification_policy": {
- "policy_version": "telegram_controlled_apply_lifecycle_v1",
- "lifecycle_scope": "assetlifecycle00000000001",
+ "policy_version": "telegram_posture_lifecycle_transition_v1",
+ "lifecycle_scope": lifecycle_scope,
"lifecycle_state": "recovered",
"provider_delivery": "state_transition",
"disposition": "send_on_transition",
@@ -1899,9 +1916,7 @@ async def test_same_controlled_apply_lifecycle_state_persists_shadow_receipt(
)
assert reservation["status"] == "suppressed"
- assert db.parameters[0]["lock_key"].endswith(
- ":assetlifecycle00000000001"
- )
+ assert db.parameters[0]["lock_key"].endswith(f":{lifecycle_scope}")
assert "lifecycle_scope" in db.statements[2]
assert "(send_status = 'pending') DESC" in db.statements[2]
assert "queued_at DESC" in db.statements[2]
@@ -1918,10 +1933,13 @@ async def test_same_controlled_apply_lifecycle_state_persists_shadow_receipt(
assert policy["lifecycle_state"] == "recovered"
assert policy["suppression_reason"] == "duplicate_lifecycle_state"
assert policy["provider_send_performed"] is False
+ assert record_kwargs["source_envelope"]["callback_reply"][
+ "execution_kind"
+ ] == "no_write_posture_readback"
@pytest.mark.asyncio
-async def test_controlled_apply_lifecycle_transition_reserves_provider_send(
+async def test_same_apply_wazuh_lifecycle_transition_reserves_new_provider_send(
monkeypatch: pytest.MonkeyPatch,
) -> None:
db = _SequenceDB(
@@ -1952,17 +1970,28 @@ async def test_controlled_apply_lifecycle_transition_reserves_provider_send(
identity = {
"project_id": "awoooi",
"automation_run_id": "00000000-0000-0000-0000-000000000502",
- "incident_id": "INC-20260722-RECOVERY",
+ "incident_id": "IWZ-POSTURE-2026072206",
"apply_op_id": "00000000-0000-0000-0000-000000000504",
}
+ lifecycle_scope = (
+ telegram_gateway_module._controlled_apply_lifecycle_scope(
+ alert_category="security",
+ canonical_asset_id="service:wazuh-manager",
+ catalog_id="ansible:wazuh-manager-posture-readback",
+ playbook_path=(
+ "infra/ansible/playbooks/wazuh-manager-posture-readback.yml"
+ ),
+ )
+ )
source_extra = {
"callback_reply": {
"action": "controlled_apply_result",
+ "execution_kind": "no_write_posture_readback",
**identity,
},
"notification_policy": {
- "policy_version": "telegram_controlled_apply_lifecycle_v1",
- "lifecycle_scope": "assetlifecycle00000000001",
+ "policy_version": "telegram_posture_lifecycle_transition_v1",
+ "lifecycle_scope": lifecycle_scope,
"lifecycle_state": "recovered",
"provider_delivery": "state_transition",
"disposition": "send_on_transition",
@@ -1976,6 +2005,13 @@ async def test_controlled_apply_lifecycle_transition_reserves_provider_send(
)
assert reservation["status"] == "reserved"
+ assert "'{notification_policy,lifecycle_scope}'" in db.statements[1]
+ assert "'{notification_policy,lifecycle_state}'" in db.statements[1]
+ assert db.parameters[1]["lifecycle_scope"] == lifecycle_scope
+ assert db.parameters[1]["lifecycle_state"] == "recovered"
+ assert "run_id <> CAST(:run_id AS uuid)" in db.statements[2]
+ assert "<> :lifecycle_state" in db.statements[2]
+ assert db.parameters[2]["lifecycle_state"] == "recovered"
record_kwargs = record_outbound.await_args.kwargs
assert record_kwargs["send_status"] == "pending"
assert record_kwargs["is_shadow"] is False
@@ -2522,7 +2558,7 @@ async def test_retry_telegram_readback_binds_exact_apply_operation(
[
("ansible:generic-posture-readback", False, False),
("ansible:generic-posture-readback", True, True),
- ("ansible:wazuh-manager-posture-readback", True, False),
+ ("ansible:wazuh-manager-posture-readback", True, True),
],
)
async def test_retry_stage_receipt_binds_shadow_allowance(
@@ -2596,7 +2632,7 @@ async def test_retry_stage_receipt_binds_shadow_allowance(
@pytest.mark.asyncio
-async def test_wazuh_retry_telegram_readback_rejects_shadow_completion(
+async def test_wazuh_retry_telegram_readback_binds_shadow_completion(
monkeypatch: pytest.MonkeyPatch,
) -> None:
db = _SequenceDB(_MappingResult(scalar=False))
@@ -2624,7 +2660,7 @@ async def test_wazuh_retry_telegram_readback_rejects_shadow_completion(
)
assert acknowledged is False
- assert db.parameters[0]["telegram_shadow_allowed"] is False
+ assert db.parameters[0]["telegram_shadow_allowed"] is True
def test_all_terminal_telegram_readbacks_require_exact_apply_operation() -> None:
@@ -2937,6 +2973,9 @@ async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
assert telegram_sender.await_args.kwargs["execution_kind"] == (
"no_write_replay"
)
+ assert telegram_sender.await_args.kwargs["provider_delivery"] == (
+ "shadow_only"
+ )
telegram_readback.assert_awaited_once()
@@ -3039,6 +3078,7 @@ async def test_retry_projection_does_not_starve_verified_success_closure(
assert result["incident_closure_written"] == 1
assert result["telegram_receipt_acknowledged"] == 1
closure.assert_awaited_once()
+ assert closure.await_args.kwargs["provider_delivery"] == "shadow_only"
@pytest.mark.asyncio
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 66a3eea25..129e45fdd 100644
--- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
+++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
@@ -63,6 +63,8 @@ def test_wazuh_runtime_sql_reads_schema_safe_packet_fallback() -> None:
assert "critical_secret_reference_provisioning_receipt_verified" not in (
_LIVE_RUNTIME_SQL
)
+ assert "telegram_notification_lifecycle_state" in _LIVE_RUNTIME_SQL
+ assert "telegram_notification_suppression_reason" in _LIVE_RUNTIME_SQL
def _durable_scheduled_wazuh_source_recurrence(
@@ -136,6 +138,15 @@ def _complete_row() -> dict[str, object]:
"telegram_receipt_id": "telegram-303",
"telegram_send_status": "sent",
"telegram_provider_message_id": "303",
+ "telegram_execution_kind": "no_write_posture_readback",
+ "telegram_notification_disposition": "sent",
+ "telegram_provider_delivery": "state_transition",
+ "telegram_notification_policy_version": (
+ "telegram_posture_lifecycle_transition_v1"
+ ),
+ "telegram_notification_lifecycle_state": "recovered",
+ "telegram_notification_suppression_reason": None,
+ "telegram_provider_send_suppressed": False,
"incident_status": "resolved",
"incident_resolved_at": "2026-07-11T03:36:30+00:00",
"incident_terminal_automation_run_id": run_id,
@@ -280,7 +291,7 @@ async def test_wazuh_runtime_loader_keeps_ingress_and_posture_lanes_visible(
assert lanes["posture"]["summary"]["runtime_closed_count"] == 1
-def test_wazuh_posture_shadow_receipt_cannot_replace_provider_send() -> None:
+def test_wazuh_posture_shadow_receipt_closes_unchanged_healthy_state() -> None:
row = _complete_row()
row.update(
{
@@ -292,6 +303,10 @@ def test_wazuh_posture_shadow_receipt_cannot_replace_provider_send() -> None:
"telegram_notification_policy_version": (
"telegram_posture_state_digest_v1"
),
+ "telegram_notification_lifecycle_state": "recovered",
+ "telegram_notification_suppression_reason": (
+ "historical_projection_backfill"
+ ),
"telegram_provider_send_suppressed": True,
}
)
@@ -301,14 +316,14 @@ def test_wazuh_posture_shadow_receipt_cannot_replace_provider_send() -> None:
executor_enabled=True,
)
- assert payload["summary"]["runtime_closed_count"] == 0
- assert payload["summary"]["telegram_receipt_count"] == 0
+ 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["recipient_receipt"]["status"] == "suppressed_unchanged"
assert payload["recipient_receipt"]["provider_message_ref"] is None
- assert "telegram_receipt" in payload["missing_stage_ids"]
- assert "incident_closure" in payload["missing_stage_ids"]
+ assert "telegram_receipt" not in payload["missing_stage_ids"]
+ assert "incident_closure" not in payload["missing_stage_ids"]
assert (
payload["boundaries"][
"healthy_posture_telegram_provider_send_performed"
@@ -318,7 +333,7 @@ def test_wazuh_posture_shadow_receipt_cannot_replace_provider_send() -> None:
assert payload["boundaries"]["healthy_posture_notification_suppressed"] is True
-def test_wazuh_posture_duplicate_lifecycle_shadow_is_not_recipient_receipt() -> None:
+def test_wazuh_posture_duplicate_lifecycle_shadow_is_terminal_receipt() -> None:
row = _complete_row()
row.update(
{
@@ -330,6 +345,42 @@ def test_wazuh_posture_duplicate_lifecycle_shadow_is_not_recipient_receipt() ->
"telegram_notification_policy_version": (
"telegram_posture_lifecycle_transition_v1"
),
+ "telegram_notification_lifecycle_state": "recovered",
+ "telegram_notification_suppression_reason": (
+ "duplicate_lifecycle_state"
+ ),
+ "telegram_provider_send_suppressed": True,
+ }
+ )
+
+ 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["recipient_receipt"]["status"] == "suppressed_unchanged"
+
+
+def test_wazuh_failure_digest_shadow_cannot_close_recovered_posture() -> 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": "digest",
+ "telegram_notification_policy_version": (
+ "telegram_posture_state_digest_v1"
+ ),
+ "telegram_notification_lifecycle_state": "failed",
+ "telegram_notification_suppression_reason": (
+ "duplicate_failure_fingerprint_window"
+ ),
"telegram_provider_send_suppressed": True,
}
)
@@ -342,8 +393,24 @@ def test_wazuh_posture_duplicate_lifecycle_shadow_is_not_recipient_receipt() ->
assert payload["summary"]["runtime_closed_count"] == 0
assert payload["summary"]["telegram_receipt_count"] == 0
assert payload["summary"]["telegram_provider_send_count"] == 0
- assert payload["summary"]["telegram_notification_suppressed_count"] == 1
- assert payload["recipient_receipt"]["status"] == "suppressed_unchanged"
+ assert "telegram_receipt" in payload["missing_stage_ids"]
+ assert "incident_closure" in payload["missing_stage_ids"]
+
+
+def test_wazuh_sent_failed_transition_cannot_close_recovered_posture() -> None:
+ row = _complete_row()
+ row["telegram_notification_lifecycle_state"] = "failed"
+
+ payload = build_iwooos_wazuh_controlled_executor_runtime_readback(
+ row,
+ executor_enabled=True,
+ )
+
+ assert payload["summary"]["runtime_closed_count"] == 0
+ assert payload["summary"]["telegram_receipt_count"] == 0
+ assert payload["summary"]["telegram_provider_send_count"] == 0
+ assert "telegram_receipt" in payload["missing_stage_ids"]
+ assert "incident_closure" in payload["missing_stage_ids"]
def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None:
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 96b077bd7..cfa31e657 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/apps/api/tests/test_telegram_message_templates.py
@@ -1575,7 +1575,7 @@ 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(
+async def test_historical_healthy_posture_is_shadowed_and_never_claims_apply(
monkeypatch,
) -> None:
gateway = TelegramGateway()
@@ -1621,7 +1621,11 @@ async def test_healthy_posture_readback_is_suppressed_and_never_claims_apply(
duration_ms=8649,
verifier_written=True,
learning_written=True,
+ rag_written=True,
+ mcp_context_ready=True,
+ playbook_trust_written=True,
execution_kind="no_write_posture_readback",
+ provider_delivery="shadow_only",
project_id="awoooi",
)
@@ -1642,11 +1646,12 @@ async def test_healthy_posture_readback_is_suppressed_and_never_claims_apply(
assert policy["provider_delivery"] == "shadow_only"
assert policy["disposition"] == "suppressed"
assert policy["provider_send_performed"] is False
+ assert policy["lifecycle_state"] == "recovered"
assert policy["runtime_write_performed"] is False
@pytest.mark.asyncio
-async def test_healthy_wazuh_posture_is_recipient_visible_once_per_run(
+async def test_healthy_wazuh_posture_uses_stable_lifecycle_transition(
monkeypatch,
) -> None:
gateway = TelegramGateway()
@@ -1702,16 +1707,16 @@ async def test_healthy_wazuh_posture_is_recipient_visible_once_per_run(
method, payload = sent_requests[0]
assert method == "sendMessage"
assert "Runtime write: 0" in payload["text"]
- assert "本次 run 僅外送一次" in payload["text"]
+ assert "狀態轉換僅外送一次" in payload["text"]
policy = payload["_awooop_source_envelope_extra"]["notification_policy"]
assert policy["policy_version"] == (
- "telegram_wazuh_posture_same_run_receipt_v1"
+ "telegram_posture_lifecycle_transition_v1"
)
assert policy["provider_delivery"] == "state_transition"
- assert policy["disposition"] == "send_once_per_run"
- assert policy["group_scope"] == "controlled_asset_same_run"
+ assert policy["disposition"] == "send_on_transition"
+ assert policy["group_scope"] == "controlled_asset_lifecycle"
assert policy["delivery_semantics"] == (
- "at_least_once_per_run_after_unknown_ack_lease"
+ "at_least_once_after_unknown_ack_lease"
)
assert policy["lifecycle_state"] == "recovered"
assert policy["runtime_write_performed"] is False
@@ -1723,7 +1728,6 @@ async def test_healthy_wazuh_posture_is_recipient_visible_once_per_run(
playbook_path=(
"infra/ansible/playbooks/wazuh-manager-posture-readback.yml"
),
- scope_discriminator="00000000-0000-0000-0000-000000000013",
)
)
@@ -1762,7 +1766,7 @@ def test_controlled_apply_lifecycle_scope_preserves_legacy_and_separates_runs(
@pytest.mark.asyncio
-async def test_failed_posture_readback_is_digest_deduplicated_and_queues_repair(
+async def test_failed_wazuh_posture_uses_lifecycle_transition_and_queues_repair(
monkeypatch,
) -> None:
gateway = TelegramGateway()
@@ -1808,10 +1812,16 @@ async def test_failed_posture_readback_is_digest_deduplicated_and_queues_repair(
assert "AI 自動修復摘要｜姿態檢查待修復" in payload["text"]
assert "transport_or_playbook_repair_queued" in payload["text"]
assert "Runtime write: 0" in payload["text"]
+ assert "狀態轉換僅外送一次" 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
+ 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"] == "failed"
+ assert "digest_window_minutes" not in policy
+ assert "failure_fingerprint" not in policy
def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None: