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

fix(sre): verify Telegram learning receipts

Browse Source
This commit is contained in:
Your Name
2026-07-19 05:45:08 +08:00
parent 60ee187bd9
commit eafaa20f33
5 changed files with 361 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py
View File

@@ -452,10 +452,57 @@ async def test_log_controlled_writeback_consumer_maps_ready_bindings_when_alert_
receipt["status"] == "ready_for_ai_loop_context"
for receipt in context["context_receipts"]
)
assert all(
receipt["receipt_kind"] == "log_controlled_writeback_consumer"
and receipt["target_write_performed"] is True
and receipt["consumer_write_receipt"]["status"] == "success"
and receipt["consumer_write_receipt"][
"target_context_receipt_write_performed"
]
is True
and receipt["consumer_write_receipt"]["runtime_target_write_performed"]
is True
and receipt["verifier_write_performed"] is False
for receipt in context["context_receipts"]
)
assert context["operation_boundaries"]["runtime_target_write_performed"] is True
assert context["operation_boundaries"]["raw_payload_included"] is False
@pytest.mark.asyncio
async def test_log_controlled_writeback_consumer_fallback_rejects_unverified_write(
monkeypatch,
):
consumer_rows = _consumer_receipt_rows()
consumer_rows[0]["runtime_target_write_performed"] = "false"
fake_db = _FakeDb(_ledger_rows(), consumer_rows)
monkeypatch.setattr(
consumer_module,
"get_db_context",
lambda project_id: _FakeContext(fake_db),
)
async def alert_registry_timeout(**_kwargs):
raise TimeoutError("alert registry unavailable")
monkeypatch.setattr(
consumer_module,
"list_ai_alert_card_delivery_readback",
alert_registry_timeout,
)
payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback()
assert payload["rollups"]["target_context_receipt_write_count"] == 5
assert payload["rollups"]["runtime_target_write_performed"] is False
assert payload["operation_boundaries"]["runtime_target_write_performed"] is False
context = payload["telegram_alert_learning_context"]
assert context["status"] == (
"blocked_telegram_alert_learning_registry_readback_unavailable"
)
assert context["context_receipt_count"] == 0
def test_log_controlled_writeback_consumer_direct_rows_parse_json_text():
dispatch_row = _ledger_rows()[0]
consumer_row = _consumer_receipt_rows()[0]
@@ -621,6 +668,7 @@ def _assert_consumer_readback(payload: dict):
if receipt["target"] == "ai_agent"
][0]
assert ai_agent_receipt["status"] == "ready_for_ai_loop_context"
assert ai_agent_receipt["receipt_kind"] == "alert_card_learning_registry"
assert ai_agent_receipt["work_item_id"] == "CIR-P0-TG-001"
assert ai_agent_receipt["metadata_only"] is True
assert ai_agent_receipt["target_write_performed"] is False

148
apps/api/tests/test_telegram_alert_learning_context_post_apply_verifier_api.py
View File

@@ -80,6 +80,92 @@ def _context_receipts() -> list[dict]:
]
def _fallback_context_receipts() -> list[dict]:
targets = [
("km", "km_consumer_context"),
("rag", "rag_consumer_context"),
("playbook", "playbook_consumer_context"),
("mcp", "mcp_consumer_context"),
("verifier", "verifier_consumer_context"),
("ai_agent", "ai_agent_consumer_context"),
]
return [
{
"receipt_kind": "log_controlled_writeback_consumer",
"receipt_id": (
"telegram_alert_learning_context_fallback::"
f"log_controlled_writeback_consumed::{target}"
),
"source_receipt_id": f"log_controlled_writeback_consumed::{target}",
"source_run_id": "",
"source_message_id": "",
"work_item_id": "CIR-P0-TG-001",
"target": target,
"consumer_surface": surface,
"source_ref": f"log_controlled_writeback_dispatched::{target}",
"ai_agent_context_ref": (
"ai-agent://awoooi/log-controlled-writeback/"
f"log_controlled_writeback_consumed::{target}"
),
"status": "ready_for_ai_loop_context",
"metadata_only": True,
"target_write_performed": True,
"consumer_write_receipt": {
"consumer_receipt_id": (
f"log_controlled_writeback_consumed::{target}"
),
"status": "success",
"target_context_receipt_write_performed": True,
"runtime_target_write_performed": True,
},
"verifier_write_performed": False,
"raw_payload_included": False,
"target_selector": {
"project_id": "awoooi",
"dispatch_receipt_id": (
f"log_controlled_writeback_dispatched::{target}"
),
"consumer_receipt_id": (
f"log_controlled_writeback_consumed::{target}"
),
"target": target,
"source_ref": f"log_controlled_writeback_dispatched::{target}",
},
"source_of_truth_diff": {
"current_state": (
"telegram_alert_registry_unavailable_or_not_ready"
),
"desired_state": "ai_loop_agent_context_receipt_available",
"delta_kind": (
f"log_controlled_writeback_{target}_context_fallback"
),
"raw_payload_included": False,
},
"check_mode": {
"enabled": True,
"checks": [
"consumer_binding_ready",
"consumer_apply_receipt_present",
"metadata_only_raw_payload_absent",
"post_apply_verifier_refs_present",
],
},
"rollback": {
"required": True,
"rollback_ref": (
"rollback://telegram-alert-learning-context-fallback/"
f"log_controlled_writeback_consumed::{target}"
),
},
"post_apply_verifier": {
"required": True,
"verifier_refs": [f"post-write-verifier://{target}"],
},
}
for target, surface in targets
]
def _consumer_readback(*, receipts: list[dict] | None = None) -> dict:
return {
"schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1",
@@ -123,6 +209,7 @@ def test_telegram_alert_learning_context_post_apply_verifier_passes_receipts():
assert rollups["verified_context_receipt_count"] == 6
assert rollups["verified_target_count"] == 6
assert rollups["verified_ai_agent_context_receipt_count"] == 1
assert rollups["verified_target_context_write_count"] == 0
assert rollups["raw_payload_included_count"] == 0
assert rollups["runtime_write_performed_count"] == 0
@@ -141,6 +228,48 @@ def test_telegram_alert_learning_context_post_apply_verifier_passes_receipts():
assert "raw Telegram payload" not in serialized
def test_telegram_alert_learning_context_post_apply_verifier_passes_fallback_receipts():
payload = build_telegram_alert_learning_context_post_apply_verifier(
consumer_readback=_consumer_readback(receipts=_fallback_context_receipts())
)
assert payload["status"] == (
"telegram_alert_learning_context_post_apply_verified"
)
assert payload["active_blockers"] == []
assert payload["rollups"]["verified_context_receipt_count"] == 6
assert payload["rollups"]["verified_target_context_write_count"] == 6
assert {
result["receipt_kind"] for result in payload["verifier_results"]
} == {"log_controlled_writeback_consumer"}
assert all(
result["target_write_contract_verified"] is True
and result["verified_target_context_write_performed"] is True
and result["verifier_write_performed"] is False
for result in payload["verifier_results"]
)
def test_telegram_alert_learning_context_post_apply_verifier_blocks_bad_fallback_receipt():
receipts = _fallback_context_receipts()
receipts[0]["consumer_write_receipt"]["runtime_target_write_performed"] = False
payload = build_telegram_alert_learning_context_post_apply_verifier(
consumer_readback=_consumer_readback(receipts=receipts)
)
assert payload["status"] == (
"blocked_telegram_alert_learning_context_post_apply_verifier"
)
assert payload["rollups"]["failed_context_receipt_count"] == 1
failed = [
result
for result in payload["verifier_results"]
if result["verifier_status"] == "failed"
]
assert failed[0]["failed_checks"] == ["consumer_write_receipt_verified"]
def test_telegram_alert_learning_context_post_apply_verifier_blocks_bad_receipt():
receipts = _context_receipts()
receipts[0]["rollback"] = {"required": True, "rollback_ref": ""}
@@ -159,6 +288,25 @@ def test_telegram_alert_learning_context_post_apply_verifier_blocks_bad_receipt(
)
def test_telegram_alert_learning_context_post_apply_verifier_rejects_unknown_kind():
receipts = _context_receipts()
receipts[0]["receipt_kind"] = "unsupported_learning_receipt"
payload = build_telegram_alert_learning_context_post_apply_verifier(
consumer_readback=_consumer_readback(receipts=receipts)
)
failed = [
result
for result in payload["verifier_results"]
if result["verifier_status"] == "failed"
]
assert payload["status"] == (
"blocked_telegram_alert_learning_context_post_apply_verifier"
)
assert failed[0]["failed_checks"] == ["receipt_kind_supported"]
def test_telegram_alert_learning_context_post_apply_verifier_endpoint(monkeypatch):
async def fake_loader():
return build_telegram_alert_learning_context_post_apply_verifier(