fix(telegram): suppress healthy Wazuh posture spam

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

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"

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
 

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: <code>0</code>" in payload["text"]
+    assert "CONTROLLED APPLY RESULT" not in payload["text"]
+    assert "Runtime apply: <code>1</code>" 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: <code>0</code>" 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(