wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(agent): reserve controlled apply receipts
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 2m45s
Details
CD Pipeline / build-and-deploy (push) Failing after 16s
Details
CD Pipeline / post-deploy-checks (push) Has been skipped
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 16:45:49 +08:00
parent 673b0d57a4
commit c9ef737b16
3 changed files with 693 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -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,

384
apps/api/src/services/telegram_gateway.py
View File

@@ -131,6 +131,7 @@ _AI_ADVISORY_CALLBACK_RE = re.compile(
_CODE_REF_RE = re.compile(r"<code>([0-9a-f]{7,12})</code>", 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<target>[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]

251
apps/api/tests/test_ansible_verified_closure.py
View File

@@ -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(
{