fix(ansible): gate shadow receipt acknowledgements

apps/api/tests/test_ansible_incident_ledger_churn_guard.py

@@ -507,3 +507,25 @@ def test_wazuh_posture_shadow_receipt_is_a_terminal_notification_receipt() -> No
     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
+
+
+@pytest.mark.asyncio
+async def test_wazuh_posture_retry_binds_shadow_receipt_allowance(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    db = _SequenceDB(_ScalarResult(True))
+
+    @asynccontextmanager
+    async def fake_get_db_context(_project_id):
+        yield db
+
+    monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
+
+    acknowledged = await service._retry_telegram_receipt_acknowledged(
+        _wazuh_claim(),
+        apply_op_id="00000000-0000-0000-0000-000000000203",
+        project_id="awoooi",
+    )
+
+    assert acknowledged is True
+    assert db.parameters[0]["telegram_shadow_allowed"] is True

apps/api/tests/test_ansible_verified_closure.py

@@ -2235,6 +2235,77 @@ async def test_retry_telegram_readback_binds_exact_apply_operation(
     assert "no_write_replay" in db.statements[0]
     assert "provider_send_performed" in db.statements[0]
     assert db.parameters[0]["apply_op_id"] == apply_op_id
+    assert db.parameters[0]["telegram_shadow_allowed"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("shadow_allowed", [False, True])
+async def test_retry_stage_receipt_binds_shadow_allowance(
+    monkeypatch: pytest.MonkeyPatch,
+    shadow_allowed: bool,
+) -> None:
+    claim = _claim()
+    apply_op_id = "00000000-0000-0000-0000-000000000104"
+    replay_op_id = "00000000-0000-0000-0000-000000000105"
+    db = _SequenceDB(
+        _MappingResult(
+            row={
+                "status": "success",
+                "input": {
+                    "automation_run_id": claim.input_payload[
+                        "automation_run_id"
+                    ],
+                    "retry_of_apply_op_id": apply_op_id,
+                },
+                "output": {
+                    "runtime_apply_executed": False,
+                    "terminal_disposition": (
+                        "no_write_replay_passed_waiting_repair_candidate"
+                    ),
+                },
+                "dry_run_result": {},
+            }
+        ),
+        _MappingResult(scalar=True),
+    )
+
+    @asynccontextmanager
+    async def fake_get_db_context(_project_id):
+        yield db
+
+    monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
+    monkeypatch.setattr(
+        service,
+        "_claim_is_no_write_observation",
+        lambda _claim: shadow_allowed,
+    )
+    monkeypatch.setattr(
+        service,
+        "_record_incident_terminal_disposition",
+        AsyncMock(return_value=None),
+    )
+    monkeypatch.setattr(
+        service,
+        "_append_runtime_stage_receipts_to_apply",
+        AsyncMock(return_value=True),
+    )
+
+    recorded = await service._record_retry_runtime_stage_receipt(
+        claim,
+        service.AnsibleRunResult(
+            returncode=0,
+            stdout="",
+            stderr="",
+            duration_ms=1,
+        ),
+        apply_op_id=apply_op_id,
+        replay_op_id=replay_op_id,
+        project_id="awoooi",
+    )
+
+    assert recorded is True
+    assert ":telegram_shadow_allowed" in db.statements[1]
+    assert db.parameters[1]["telegram_shadow_allowed"] is shadow_allowed
 
 
 def test_all_terminal_telegram_readbacks_require_exact_apply_operation() -> None: