From c9ef737b160314c8eafc6ce7d087b3096dbb5034 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 16:45:49 +0800 Subject: [PATCH] fix(agent): reserve controlled apply receipts --- .../awooop_ansible_check_mode_service.py | 85 +++- apps/api/src/services/telegram_gateway.py | 384 +++++++++++++++++- .../tests/test_ansible_verified_closure.py | 251 ++++++++++++ 3 files changed, 693 insertions(+), 27 deletions(-) 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 d1a220b81..00a1485d6 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -2146,6 +2146,8 @@ async def _read_verified_apply_closure_prerequisites( outbound.source_envelope #>> '{source_refs,incident_ids,0}' ) = :incident_id + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' = :apply_op_id ) AS telegram_receipt, ( :approval_id = '' @@ -2915,6 +2917,7 @@ def _build_retry_runtime_stage_receipt( async def _retry_telegram_receipt_acknowledged( claim: AnsibleCheckModeClaim, *, + apply_op_id: str, project_id: str, ) -> bool: automation_run_id = _automation_run_id_for_claim(claim) @@ -2946,12 +2949,15 @@ async def _retry_telegram_receipt_acknowledged( outbound.source_envelope #>> '{source_refs,incident_ids,0}' ) = :incident_id + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' = :apply_op_id ) """), { "project_id": project_id, "automation_run_id": automation_run_id, "incident_id": claim.incident_id, + "apply_op_id": apply_op_id, }, ) return result.scalar() is True @@ -3040,12 +3046,15 @@ async def _record_retry_runtime_stage_receipt( source_envelope #>> '{source_refs,incident_ids,0}' ) = :incident_id + AND source_envelope #>> + '{callback_reply,apply_op_id}' = :apply_op_id ) """), { "project_id": project_id, "automation_run_id": automation_run_id, "incident_id": claim.incident_id, + "apply_op_id": apply_op_id, }, ) telegram_receipt_acknowledged = telegram_readback.scalar() is True @@ -3597,7 +3606,11 @@ async def _send_controlled_apply_telegram_receipt( learning_written=bool(writeback.get("learning")), project_id=project_id, ) - return bool(response) + return bool( + isinstance(response, Mapping) + and response.get("ok") is True + and response.get("_awooop_outbound_mirror_acknowledged") is True + ) except Exception as exc: logger.warning( "ansible_controlled_apply_telegram_receipt_failed", @@ -3619,14 +3632,47 @@ async def _reconcile_verified_apply_closure_projections( ) -> dict[str, Any]: """Repair terminal projections without ever executing the apply again.""" - if not ( + verified_success = bool( result.returncode == 0 and result.post_verifier_passed is True and writeback.get("verification_passed") is True - ): + ) + if not verified_success: + telegram_receipt = await _retry_telegram_receipt_acknowledged( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + if not telegram_receipt: + receipt_finalized = await _send_controlled_apply_telegram_receipt( + claim, + result, + apply_op_id=apply_op_id, + writeback=writeback, + project_id=project_id, + ) + if receipt_finalized: + telegram_receipt = await _retry_telegram_receipt_acknowledged( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + ) + telegram_lifecycle = False + if telegram_receipt: + telegram_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "TELEGRAM_RESULT_SENT", + apply_op_id=apply_op_id, + success=False, + action_detail="controlled_apply_result_receipt_sent", + project_id=project_id, + post_verifier_passed=result.post_verifier_passed, + ) return { "status": "not_verified_success", "closed": False, + "telegram_receipt_acknowledged": telegram_receipt, + "telegram_lifecycle_written": telegram_lifecycle, "runtime_apply_executed": False, } @@ -4205,6 +4251,8 @@ async def _load_missing_retry_terminal_projection_rows( apply.incident_id::text, apply.input ->> 'incident_id' ) + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' = apply.op_id::text ) AS telegram_receipt_acknowledged FROM automation_operation_log apply JOIN automation_operation_log replay @@ -4338,6 +4386,9 @@ async def _load_missing_retry_terminal_projection_rows( outbound.source_envelope #>> '{source_refs,incident_ids,0}' ) = incident.incident_id + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' + = apply.op_id::text ) OR NOT EXISTS ( SELECT 1 @@ -4643,6 +4694,7 @@ async def backfill_missing_retry_terminal_projections_once( telegram_acknowledged = ( await _retry_telegram_receipt_acknowledged( claim, + apply_op_id=apply_op_id, project_id=project_id, ) ) @@ -5973,24 +6025,8 @@ async def run_controlled_apply_for_claim( ), post_verifier_passed=verified_result.post_verifier_passed, ) - telegram_receipt_sent = await _send_controlled_apply_telegram_receipt( - claim, - result, - apply_op_id=apply_op_id, - writeback=writeback, - project_id=project_id, - ) - lifecycle_telegram_written = False - if telegram_receipt_sent: - lifecycle_telegram_written = await _append_alert_lifecycle_receipt( - claim, - "TELEGRAM_RESULT_SENT", - apply_op_id=apply_op_id, - success=execution_success, - action_detail="controlled_apply_result_receipt_sent", - project_id=project_id, - post_verifier_passed=verified_result.post_verifier_passed, - ) + # Single dispatch entry: `_send_controlled_apply_telegram_receipt` and the + # `"TELEGRAM_RESULT_SENT"` lifecycle are owned by closure reconciliation. try: closure = await _reconcile_verified_apply_closure_projections( claim, @@ -6014,6 +6050,13 @@ async def run_controlled_apply_for_claim( runtime_apply_replay_required=False, ) + telegram_receipt_sent = ( + closure.get("telegram_receipt_acknowledged") is True + ) + lifecycle_telegram_written = ( + closure.get("telegram_lifecycle_written") is True + ) + logger.info( "ansible_controlled_apply_completed", op_id=apply_op_id, diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 5ddad0403..3d0e5d24f 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -131,6 +131,7 @@ _AI_ADVISORY_CALLBACK_RE = re.compile( _CODE_REF_RE = re.compile(r"([0-9a-f]{7,12})", re.IGNORECASE) _TELEGRAM_HTML_CHUNK_LIMIT = 3600 _AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY = "_awooop_source_envelope_extra" +_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS = 3 _HOST_RESOURCE_ALERT_HEADER_RE = re.compile( r"\b(?:WARN|CRIT|INFO)\s+" r"(?P[A-Za-z0-9_.-]+).*?" @@ -2132,6 +2133,41 @@ def _legacy_outbound_run_id(chat_id: str, provider_message_id: str) -> UUID: return uuid5(NAMESPACE_URL, f"awoooi:legacy-telegram:{chat_id}:{provider_message_id}") +def _controlled_apply_result_delivery_identity( + source_envelope_extra: object, +) -> dict[str, str] | None: + """Return the stable identity required before a controlled-result send.""" + if not isinstance(source_envelope_extra, dict): + return None + callback_reply = source_envelope_extra.get("callback_reply") + if not isinstance(callback_reply, dict): + return None + if callback_reply.get("action") != "controlled_apply_result": + return None + + identity = { + "project_id": str(callback_reply.get("project_id") or "").strip(), + "automation_run_id": str( + callback_reply.get("automation_run_id") or "" + ).strip(), + "incident_id": str(callback_reply.get("incident_id") or "").strip(), + "apply_op_id": str(callback_reply.get("apply_op_id") or "").strip(), + } + return identity if all(identity.values()) else None + + +def _controlled_apply_result_delivery_run_id(identity: dict[str, str]) -> UUID: + """Build one stable outbox identity for one controlled apply result.""" + return uuid5( + NAMESPACE_URL, + "awoooi:controlled-apply-result:" + f"{identity['project_id']}:" + f"{identity['automation_run_id']}:" + f"{identity['incident_id']}:" + f"{identity['apply_op_id']}", + ) + + def _infer_outbound_message_type( text: str, payload: dict, @@ -5065,6 +5101,258 @@ class TelegramGateway: """取得 Telegram API URL""" return f"{self.TELEGRAM_API_BASE}/bot{self.bot_token}" + async def _reserve_controlled_apply_result_outbound( + self, + *, + identity: dict[str, str], + payload: dict, + source_envelope_extra: dict[str, object], + ) -> dict[str, object]: + """Persist a send-once reservation before contacting Telegram.""" + from sqlalchemy import text + + from src.db.base import get_db_context + from src.services.channel_hub import record_outbound_message + + project_id = identity["project_id"] + delivery_run_id = _controlled_apply_result_delivery_run_id(identity) + lock_key = f"telegram-controlled-apply-result:{delivery_run_id}" + existing_row: dict[str, object] | None = None + reservation_message_id: str | None = None + try: + async with get_db_context(project_id) as db: + await db.execute( + text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:lock_key, 0) + ) + """), + {"lock_key": lock_key}, + ) + existing = await db.execute( + text(""" + SELECT + message_id::text AS message_id, + send_status, + provider_message_id + FROM awooop_outbound_message + WHERE project_id = :project_id + AND run_id = CAST(:run_id AS uuid) + AND channel_type = 'telegram' + ORDER BY + ( + send_status = 'sent' + AND provider_message_id IS NOT NULL + ) DESC, + queued_at ASC + LIMIT 1 + FOR UPDATE + """), + { + "project_id": project_id, + "run_id": str(delivery_run_id), + }, + ) + row = existing.mappings().one_or_none() + if row is not None: + existing_row = dict(row) + else: + reservation_extra = dict(source_envelope_extra) + callback_reply = dict( + reservation_extra.get("callback_reply") or {} + ) + callback_reply["status"] = ( + "controlled_apply_result_reserved" + ) + reservation_extra["callback_reply"] = callback_reply + reservation_message_id = str( + await record_outbound_message( + db, + project_id=project_id, + run_id=delivery_run_id, + channel_type="telegram", + channel_chat_id=str(payload.get("chat_id") or ""), + message_type=_infer_outbound_message_type( + str(payload.get("text") or ""), + payload, + source_envelope_extra, + ), + content=str(payload.get("text") or ""), + source_envelope=( + _merge_outbound_source_envelope_extra( + _outbound_source_envelope( + "sendMessage", + payload, + ), + reservation_extra, + ) + ), + provider_message_id=None, + send_status="pending", + triggered_by_state="controlled_apply_result", + is_shadow=False, + ) + ) + except Exception as exc: + logger.warning( + "controlled_apply_result_reservation_failed", + automation_run_id=identity["automation_run_id"], + incident_id=identity["incident_id"], + apply_op_id=identity["apply_op_id"], + error_type=type(exc).__name__, + ) + return { + "status": "reservation_failed", + "run_id": str(delivery_run_id), + } + + if existing_row is not None: + provider_message_id = str( + existing_row.get("provider_message_id") or "" + ) + if ( + existing_row.get("send_status") == "sent" + and provider_message_id + ): + return { + "status": "sent", + "run_id": str(delivery_run_id), + "message_id": str(existing_row["message_id"]), + "provider_message_id": provider_message_id, + } + return { + "status": "pending_unknown", + "run_id": str(delivery_run_id), + "message_id": str(existing_row["message_id"]), + } + + if not reservation_message_id: + return { + "status": "reservation_failed", + "run_id": str(delivery_run_id), + } + + try: + async with get_db_context(project_id) as db: + readback = await db.execute( + text(""" + SELECT send_status + FROM awooop_outbound_message + WHERE project_id = :project_id + AND message_id = CAST(:message_id AS uuid) + AND run_id = CAST(:run_id AS uuid) + AND channel_type = 'telegram' + LIMIT 1 + """), + { + "project_id": project_id, + "message_id": reservation_message_id, + "run_id": str(delivery_run_id), + }, + ) + durable_status = readback.scalar_one_or_none() + except Exception as exc: + logger.warning( + "controlled_apply_result_reservation_readback_failed", + automation_run_id=identity["automation_run_id"], + incident_id=identity["incident_id"], + apply_op_id=identity["apply_op_id"], + error_type=type(exc).__name__, + ) + durable_status = None + + return { + "status": ( + "reserved" if durable_status == "pending" else "pending_unknown" + ), + "run_id": str(delivery_run_id), + "message_id": reservation_message_id, + } + + async def _finalize_controlled_apply_result_outbound( + self, + *, + identity: dict[str, str], + reservation: dict[str, object], + provider_message_id: str, + ) -> bool: + """Finalize only the reserved row, retrying DB acknowledgement only.""" + from sqlalchemy import text + + from src.db.base import get_db_context + + message_id = str(reservation.get("message_id") or "") + run_id = str(reservation.get("run_id") or "") + if not message_id or not run_id or not provider_message_id: + return False + + lock_key = f"telegram-controlled-apply-result:{run_id}" + for attempt in range(_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS): + finalized = False + try: + async with get_db_context(identity["project_id"]) as db: + await db.execute( + text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:lock_key, 0) + ) + """), + {"lock_key": lock_key}, + ) + result = await db.execute( + text(""" + UPDATE awooop_outbound_message + SET provider_message_id = :provider_message_id, + send_status = 'sent', + send_error = NULL, + sent_at = coalesce(sent_at, NOW()), + triggered_by_state = 'controlled_apply_result', + source_envelope = jsonb_set( + source_envelope, + '{callback_reply,status}', + to_jsonb('callback_reply_sent'::text), + true + ) + WHERE project_id = :project_id + AND message_id = CAST(:message_id AS uuid) + AND run_id = CAST(:run_id AS uuid) + AND channel_type = 'telegram' + AND send_status IN ('pending', 'sent') + AND ( + provider_message_id IS NULL + OR provider_message_id = :provider_message_id + ) + RETURNING send_status, provider_message_id + """), + { + "project_id": identity["project_id"], + "message_id": message_id, + "run_id": run_id, + "provider_message_id": provider_message_id, + }, + ) + row = result.mappings().one_or_none() + finalized = bool( + row + and row.get("send_status") == "sent" + and str(row.get("provider_message_id") or "") + == provider_message_id + ) + if finalized: + return True + except Exception as exc: + logger.warning( + "controlled_apply_result_finalize_failed", + automation_run_id=identity["automation_run_id"], + incident_id=identity["incident_id"], + apply_op_id=identity["apply_op_id"], + attempt=attempt + 1, + error_type=type(exc).__name__, + ) + if attempt + 1 < _CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS: + await asyncio.sleep(0.05 * (attempt + 1)) + return False + async def _send_request( self, method: str, @@ -5093,6 +5381,66 @@ class TelegramGateway: payload = normalize_telegram_send_message_payload(method, payload) source_envelope_extra = payload.pop(_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY, None) await self._attach_incident_thread_reply(method, payload) + callback_reply = ( + source_envelope_extra.get("callback_reply") + if isinstance(source_envelope_extra, dict) + else None + ) + controlled_apply_result_requested = bool( + isinstance(callback_reply, dict) + and callback_reply.get("action") == "controlled_apply_result" + ) + controlled_delivery_identity = ( + _controlled_apply_result_delivery_identity(source_envelope_extra) + if controlled_apply_result_requested + else None + ) + controlled_reservation: dict[str, object] | None = None + if controlled_apply_result_requested: + if ( + method != "sendMessage" + or controlled_delivery_identity is None + or not isinstance(source_envelope_extra, dict) + ): + logger.warning( + "controlled_apply_result_reservation_identity_missing", + method=method, + ) + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "reservation_identity_missing", + "_awooop_provider_send_performed": False, + } + controlled_reservation = ( + await self._reserve_controlled_apply_result_outbound( + identity=controlled_delivery_identity, + payload=payload, + source_envelope_extra=source_envelope_extra, + ) + ) + reservation_status = str( + controlled_reservation.get("status") or "pending_unknown" + ) + if reservation_status == "sent": + return { + "ok": True, + "result": { + "message_id": controlled_reservation.get( + "provider_message_id" + ) + }, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent_reused", + "_awooop_provider_send_performed": False, + } + if reservation_status != "reserved": + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": reservation_status, + "_awooop_provider_send_performed": False, + } actionable_text = str(payload.get("text") or payload.get("caption") or "") if ( _agent99_ai_alert_card_mirror_configured() @@ -5133,12 +5481,34 @@ class TelegramGateway: result_val = result.get("result") if isinstance(result_val, dict) and "message_id" in result_val: span.set_attribute("telegram.message_id", result_val["message_id"]) - await self._mirror_outbound_message( - method=method, - payload=payload, - provider_message_id=str(result_val["message_id"]), - source_envelope_extra=source_envelope_extra, - ) + provider_message_id = str(result_val["message_id"]) + if ( + controlled_reservation is not None + and controlled_delivery_identity is not None + ): + finalized = ( + await self._finalize_controlled_apply_result_outbound( + identity=controlled_delivery_identity, + reservation=controlled_reservation, + provider_message_id=provider_message_id, + ) + ) + result["_awooop_outbound_mirror_acknowledged"] = finalized + result["_awooop_delivery_status"] = ( + "sent" if finalized else "pending_unknown" + ) + result["_awooop_provider_send_performed"] = True + else: + await self._mirror_outbound_message( + method=method, + payload=payload, + provider_message_id=provider_message_id, + source_envelope_extra=source_envelope_extra, + ) + elif controlled_reservation is not None: + result["_awooop_outbound_mirror_acknowledged"] = False + result["_awooop_delivery_status"] = "pending_unknown" + result["_awooop_provider_send_performed"] = True span.set_status(trace.Status(trace.StatusCode.OK)) return result @@ -9676,6 +10046,8 @@ class TelegramGateway: callback_reply = source_extra.get("callback_reply") if isinstance(callback_reply, dict): callback_reply["automation_run_id"] = automation_run_id + callback_reply["apply_op_id"] = apply_op_id + callback_reply["project_id"] = project_id or "awoooi" source_refs = source_extra.get("source_refs") if isinstance(source_refs, dict): source_refs["automation_run_ids"] = [automation_run_id] diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index d476cad2c..7471b28b3 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -9,6 +9,7 @@ import pytest from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job from src.services import awooop_ansible_check_mode_service as service +from src.services.telegram_gateway import TelegramGateway class _MappingResult: @@ -70,6 +71,26 @@ def _verified_result() -> service.AnsibleRunResult: ) +def _controlled_result_payload() -> dict: + return { + "chat_id": "sre-chat", + "text": "CONTROLLED APPLY RESULT INC-20260711-D037E5", + "_skip_incident_thread_reply": True, + "_awooop_source_envelope_extra": { + "callback_reply": { + "action": "controlled_apply_result", + "status": "callback_reply_sent", + "project_id": "awoooi", + "automation_run_id": ( + "00000000-0000-0000-0000-000000000102" + ), + "incident_id": "INC-20260711-D037E5", + "apply_op_id": "00000000-0000-0000-0000-000000000104", + } + }, + } + + def test_runtime_stage_ids_only_accepts_durable_receipts() -> None: stage_ids = service._runtime_stage_ids( { @@ -346,6 +367,236 @@ async def test_projection_replay_sends_missing_receipt_without_reapplying( telegram_sender.assert_awaited_once() +@pytest.mark.asyncio +async def test_projection_stays_open_until_durable_telegram_readback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + lifecycle = AsyncMock(return_value=True) + monkeypatch.setattr( + service, + "_finalize_controlled_approval_projection", + AsyncMock(return_value=True), + ) + monkeypatch.setattr(service, "_append_alert_lifecycle_receipt", lifecycle) + monkeypatch.setattr( + service, + "_read_verified_apply_closure_prerequisites", + AsyncMock( + side_effect=[ + {"receipts": {"telegram_receipt": False}}, + {"receipts": {"telegram_receipt": False}}, + ] + ), + ) + monkeypatch.setattr( + service, + "_send_controlled_apply_telegram_receipt", + AsyncMock(return_value=False), + ) + monkeypatch.setattr( + service, + "_finalize_verified_apply_closure", + AsyncMock( + return_value={ + "status": "closure_receipts_pending", + "closed": False, + "missing": ["telegram_receipt"], + } + ), + ) + + result = await service._reconcile_verified_apply_closure_projections( + _claim(), + _verified_result(), + apply_op_id="00000000-0000-0000-0000-000000000104", + writeback={ + "verification_passed": True, + "verification_result": "success", + "verification": True, + "learning": True, + }, + project_id="awoooi", + ) + + assert result["closed"] is False + assert result["telegram_receipt_acknowledged"] is False + assert result["telegram_lifecycle_written"] is False + assert lifecycle.await_count == 1 + + +@pytest.mark.asyncio +async def test_controlled_result_pending_reservation_never_resends_provider( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Response: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + return {"ok": True, "result": {"message_id": 456}} + + class _Client: + def __init__(self) -> None: + self.post_count = 0 + + async def post(self, _url, json): + self.post_count += 1 + return _Response() + + gateway = TelegramGateway() + gateway._initialized = True + client = _Client() + gateway._http_client = client + reserve = AsyncMock( + side_effect=[ + { + "status": "reserved", + "run_id": "00000000-0000-0000-0000-000000000201", + "message_id": "00000000-0000-0000-0000-000000000202", + }, + { + "status": "pending_unknown", + "run_id": "00000000-0000-0000-0000-000000000201", + "message_id": "00000000-0000-0000-0000-000000000202", + }, + ] + ) + finalize = AsyncMock(return_value=False) + monkeypatch.setattr( + gateway, + "_reserve_controlled_apply_result_outbound", + reserve, + ) + monkeypatch.setattr( + gateway, + "_finalize_controlled_apply_result_outbound", + finalize, + ) + + first = await gateway._send_request( + "sendMessage", + _controlled_result_payload(), + ) + second = await gateway._send_request( + "sendMessage", + _controlled_result_payload(), + ) + + assert client.post_count == 1 + assert first["_awooop_provider_send_performed"] is True + assert first["_awooop_outbound_mirror_acknowledged"] is False + assert second["ok"] is False + assert second["_awooop_provider_send_performed"] is False + assert second["_awooop_delivery_status"] == "pending_unknown" + finalize.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_controlled_result_reuses_durable_sent_reservation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = TelegramGateway() + gateway._initialized = True + client = AsyncMock() + gateway._http_client = client + monkeypatch.setattr( + gateway, + "_reserve_controlled_apply_result_outbound", + AsyncMock( + return_value={ + "status": "sent", + "run_id": "00000000-0000-0000-0000-000000000201", + "message_id": "00000000-0000-0000-0000-000000000202", + "provider_message_id": "456", + } + ), + ) + + result = await gateway._send_request( + "sendMessage", + _controlled_result_payload(), + ) + + assert result["ok"] is True + assert result["_awooop_outbound_mirror_acknowledged"] is True + assert result["_awooop_provider_send_performed"] is False + client.post.assert_not_awaited() + + +def test_controlled_result_uses_durable_reservation_and_single_reconcile_entry() -> None: + reserve_source = inspect.getsource( + TelegramGateway._reserve_controlled_apply_result_outbound + ) + finalize_source = inspect.getsource( + TelegramGateway._finalize_controlled_apply_result_outbound + ) + gateway_source = inspect.getsource(TelegramGateway._send_request) + live_apply_source = inspect.getsource(service.run_controlled_apply_for_claim) + + assert "pg_advisory_xact_lock" in reserve_source + assert "record_outbound_message" in reserve_source + assert 'send_status="pending"' in reserve_source + assert "reservation_readback_failed" in reserve_source + assert "UPDATE awooop_outbound_message" in finalize_source + assert "_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS" in finalize_source + assert gateway_source.index( + "_reserve_controlled_apply_result_outbound" + ) < gateway_source.index("_http_client.post") + assert ( + "telegram_receipt_sent = await _send_controlled_apply_telegram_receipt" + not in live_apply_source + ) + assert live_apply_source.count( + "_reconcile_verified_apply_closure_projections" + ) == 1 + + +@pytest.mark.asyncio +async def test_retry_telegram_readback_binds_exact_apply_operation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + db = _SequenceDB(_MappingResult(scalar=True)) + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + monkeypatch.setattr(service, "get_db_context", fake_get_db_context) + apply_op_id = "00000000-0000-0000-0000-000000000104" + + acknowledged = await service._retry_telegram_receipt_acknowledged( + _claim(), + apply_op_id=apply_op_id, + project_id="awoooi", + ) + + assert acknowledged is True + assert "{callback_reply,apply_op_id}" in db.statements[0] + assert db.parameters[0]["apply_op_id"] == apply_op_id + + +def test_all_terminal_telegram_readbacks_require_exact_apply_operation() -> None: + closure_source = inspect.getsource( + service._read_verified_apply_closure_prerequisites + ) + retry_source = inspect.getsource( + service._retry_telegram_receipt_acknowledged + ) + retry_receipt_source = inspect.getsource( + service._record_retry_runtime_stage_receipt + ) + projection_source = inspect.getsource( + service._load_missing_retry_terminal_projection_rows + ) + + assert closure_source.count("{callback_reply,apply_op_id}") == 1 + assert retry_source.count("{callback_reply,apply_op_id}") == 1 + assert retry_receipt_source.count("{callback_reply,apply_op_id}") == 1 + assert projection_source.count("{callback_reply,apply_op_id}") == 2 + assert projection_source.count("= apply.op_id::text") >= 2 + assert "coalesce(\n outbound.source_envelope #>>\n '{callback_reply,apply_op_id}'" not in projection_source + + def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -> None: proposal = candidate_job._build_backfill_proposal( {