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

fix(sre): bound Telegram receipt readbacks

Browse Source
This commit is contained in:
Your Name
2026-07-19 05:30:42 +08:00
parent c28c2b1598
commit 60ee187bd9
5 changed files with 495 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
apps/api/tests/test_awooop_operator_timeline_labels.py
View File

@@ -1752,8 +1752,8 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts
async def __aexit__(self, exc_type, exc, tb):
return False
async def passthrough_store(_kind, _key, value, **_kwargs):
return value
async def forbidden_store(*_args, **_kwargs):
raise AssertionError("read-only delivery readback must not write cache")
monkeypatch.setattr(
platform_operator_service,
@@ -1763,7 +1763,7 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts
monkeypatch.setattr(
platform_operator_service,
"store_operator_summary_async",
passthrough_store,
forbidden_store,
)
payload = await platform_operator_service.list_ai_alert_card_delivery_readback(
@@ -1773,6 +1773,7 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts
page=1,
per_page=5,
refresh=True,
cache_write=False,
)
summary_query = next(query for query in queries if "COUNT(*) AS total" in query)
@@ -1790,6 +1791,7 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts
assert "LEFT JOIN agent99_runs ar" in list_query
assert "ON ar.message_id = m.message_id" in list_query
assert "FROM agent99_runs ars" not in list_query
assert payload["summary"]["cache_write_status"] == "disabled_read_only"
assert list_query.index("LIMIT :limit OFFSET :offset") < list_query.index(
"candidate_refs AS MATERIALIZED"
)

246
apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
View File

@@ -471,6 +471,62 @@ def test_telegram_alert_monitoring_coverage_blocks_open_runtime_lifecycle():
pipeline = {stage["stage_id"]: stage for stage in payload["alert_receipt_pipeline"]}
assert pipeline["runtime_ai_automation_lifecycle_closure"]["ready"] is False
assert pipeline["runtime_ai_automation_lifecycle_closure"]["gap_count"] == 7
lifecycle_candidate = next(
item
for item in payload["ai_controlled_gap_queue"]
if item["blocker"] == "runtime_ai_automation_lifecycle_open:7"
)
assert lifecycle_candidate["status"] == "blocked_critical_break_glass_required"
assert lifecycle_candidate["risk_level"] == "critical"
assert lifecycle_candidate["critical_break_glass_required"] is True
assert lifecycle_candidate["operation_boundaries"]["requires_runtime_send"] is True
def test_telegram_alert_monitoring_gap_queue_separates_read_only_and_runtime_write():
payload = _payload(
runtime_log_readback={
"status": "unavailable",
"summary": {
"alert_operation_log_total_7d": 0,
"automation_triggered_run_count_7d": 0,
"automation_closed_run_count_7d": 0,
"automation_open_run_count_7d": 0,
"automation_runtime_closure_percent": 0.0,
"automation_runtime_closure_ready": False,
},
"event_type_counts_7d": {},
},
ai_alert_card_delivery_readback={
"status": "unavailable",
"summary": {
"total": 0,
"learning_writeback_ready_total": 0,
"db_read_status": "unavailable",
},
},
monitoring_live_receipt_apply_readback=_metadata_live_receipt_apply_readback(),
)
queue = {item["blocker"]: item for item in payload["ai_controlled_gap_queue"]}
for blocker in (
"runtime_alert_operation_log_db_readback_unavailable",
"awooop_ai_alert_card_delivery_db_readback_unavailable",
):
candidate = queue[blocker]
assert candidate["status"] == "ready_ai_read_only_retry"
assert candidate["risk_level"] == "low"
assert candidate["execution_mode"] == "automatic_bounded_readback_retry"
assert candidate["ai_controlled_execution_allowed"] is True
assert candidate["critical_break_glass_required"] is False
assert candidate["operation_boundaries"]["database_read_only"] is True
assert candidate["operation_boundaries"]["requires_runtime_send"] is False
runtime_canary = queue["runtime_ai_automation_trigger_receipt_missing"]
assert runtime_canary["status"] == "blocked_critical_break_glass_required"
assert runtime_canary["risk_level"] == "critical"
assert runtime_canary["ai_controlled_execution_allowed"] is False
assert runtime_canary["critical_break_glass_required"] is True
assert runtime_canary["operation_boundaries"]["requires_runtime_send"] is True
def test_telegram_alert_monitoring_coverage_blocks_partial_surface_contracts():
@@ -668,6 +724,189 @@ def test_telegram_alert_monitoring_coverage_endpoint(monkeypatch):
assert data["operation_boundaries"]["bot_api_call_performed"] is False
@pytest.mark.asyncio
async def test_telegram_alert_monitoring_live_loader_serializes_db_readbacks(
monkeypatch: pytest.MonkeyPatch,
):
active_db_reads = 0
max_active_db_reads = 0
read_order: list[str] = []
async def db_read(name: str, payload: dict) -> dict:
nonlocal active_db_reads, max_active_db_reads
active_db_reads += 1
max_active_db_reads = max(max_active_db_reads, active_db_reads)
read_order.append(f"{name}:start")
await asyncio.sleep(0)
read_order.append(f"{name}:end")
active_db_reads -= 1
return payload
async def matrix_readback(*, project_id: str) -> dict:
assert project_id == "awoooi"
return {
"schema_version": "telegram_alert_ai_automation_matrix_v1",
"status": "telegram_alert_ai_automation_matrix_ready",
"summary": {
"telegram_alert_surface_count": 9,
"known_direct_send_gap_count": 0,
"db_or_log_receipt_ready_surface_count": 9,
"ai_route_ready_surface_count": 9,
"controlled_queue_ready_surface_count": 9,
"post_verifier_ready_surface_count": 9,
"learning_writeback_ready_surface_count": 9,
},
}
async def consumer_readback(*, project_id: str) -> dict:
assert project_id == "awoooi"
return await db_read("consumer", _consumer_context_readback())
async def live_receipt_readback(*, project_id: str) -> dict:
assert project_id == "awoooi"
return await db_read("live_receipt", _metadata_live_receipt_apply_readback())
async def ai_card_readback(*, project_id: str) -> dict:
assert project_id == "awoooi"
return await db_read(
"ai_card",
{
"status": "ok",
"summary": {"total": 6, "learning_writeback_ready_total": 6},
},
)
async def runtime_log_readback(*, project_id: str) -> dict:
assert project_id == "awoooi"
return await db_read(
"runtime_log",
{
"status": "ok",
"summary": {
"alert_operation_log_total_7d": 4,
"automation_triggered_run_count_7d": 1,
"automation_closed_run_count_7d": 1,
"automation_open_run_count_7d": 0,
"automation_runtime_closure_percent": 100.0,
"automation_runtime_closure_ready": True,
},
"event_type_counts_7d": {"ALERT_RECEIVED": 1},
},
)
monkeypatch.setattr(
coverage_service, "_load_telegram_matrix_readback", matrix_readback
)
monkeypatch.setattr(
coverage_service,
"_load_log_controlled_writeback_consumer_readback",
consumer_readback,
)
monkeypatch.setattr(
coverage_service,
"_load_monitoring_live_receipt_apply_readback",
live_receipt_readback,
)
monkeypatch.setattr(
coverage_service, "_load_ai_alert_card_delivery", ai_card_readback
)
monkeypatch.setattr(
coverage_service, "_load_runtime_log_readback", runtime_log_readback
)
payload = (
await coverage_service.load_latest_telegram_alert_monitoring_coverage_readback()
)
assert max_active_db_reads == 1
assert read_order == [
"consumer:start",
"consumer:end",
"live_receipt:start",
"live_receipt:end",
"ai_card:start",
"ai_card:end",
"runtime_log:start",
"runtime_log:end",
]
assert payload["summary"]["runtime_automation_closure_ready"] is True
@pytest.mark.asyncio
async def test_telegram_alert_monitoring_serial_db_budget_is_below_ui_deadline(
monkeypatch: pytest.MonkeyPatch,
):
async def slow_stage() -> dict:
await asyncio.sleep(1)
return {"status": "unexpected"}
monkeypatch.setattr(coverage_service, "_SERIAL_DB_STAGE_TIMEOUT_SECONDS", 0.01)
started_at = asyncio.get_running_loop().time()
results = [
await coverage_service._load_serial_db_stage(
stage=stage,
loader=slow_stage,
)
for stage in ("consumer", "live_receipt", "ai_alert_card", "runtime_log")
]
elapsed_seconds = asyncio.get_running_loop().time() - started_at
assert coverage_service._SERIAL_DB_TOTAL_BUDGET_SECONDS == 8.0
assert coverage_service._FULL_READBACK_TOTAL_BUDGET_SECONDS == 9.5
assert coverage_service._FULL_READBACK_TOTAL_BUDGET_SECONDS < 12.0
assert elapsed_seconds < 0.2
assert [result["status"] for result in results] == [
"blocked_waiting_controlled_writeback_consumer_db_readback",
"unavailable",
"unavailable",
"unavailable",
]
@pytest.mark.asyncio
async def test_telegram_alert_monitoring_full_loader_honors_shared_deadline(
monkeypatch: pytest.MonkeyPatch,
):
async def slow_stage(**_kwargs) -> dict:
await asyncio.sleep(1)
return {"status": "unexpected"}
monkeypatch.setattr(coverage_service, "_PRELUDE_STAGE_TIMEOUT_SECONDS", 0.01)
monkeypatch.setattr(coverage_service, "_SERIAL_DB_STAGE_TIMEOUT_SECONDS", 0.01)
monkeypatch.setattr(
coverage_service, "_load_telegram_matrix_readback", slow_stage
)
monkeypatch.setattr(
coverage_service,
"_load_log_controlled_writeback_consumer_readback",
slow_stage,
)
monkeypatch.setattr(
coverage_service,
"_load_monitoring_live_receipt_apply_readback",
slow_stage,
)
monkeypatch.setattr(
coverage_service, "_load_ai_alert_card_delivery", slow_stage
)
monkeypatch.setattr(coverage_service, "_load_runtime_log_readback", slow_stage)
monkeypatch.setattr(
coverage_service, "_inspect_source_contract", lambda _repo_root: _source_contract()
)
started_at = asyncio.get_running_loop().time()
payload = (
await coverage_service.load_latest_telegram_alert_monitoring_coverage_readback()
)
elapsed_seconds = asyncio.get_running_loop().time() - started_at
assert elapsed_seconds < 0.2
assert payload["status"] == (
"blocked_telegram_alert_monitoring_coverage_gaps_present"
)
assert payload["summary"]["runtime_automation_closure_ready"] is False
@pytest.mark.asyncio
async def test_telegram_alert_monitoring_live_readbacks_are_bounded(
monkeypatch: pytest.MonkeyPatch,
@@ -855,7 +1094,10 @@ def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None:
async def test_telegram_alert_monitoring_marks_ai_alert_cards_source_unavailable(
monkeypatch: pytest.MonkeyPatch,
):
async def source_unavailable_ai_alert_cards(**_kwargs):
call_kwargs: dict = {}
async def source_unavailable_ai_alert_cards(**kwargs):
call_kwargs.update(kwargs)
return (
platform_operator_service
._ai_alert_card_delivery_source_unavailable_response(
@@ -877,6 +1119,8 @@ async def test_telegram_alert_monitoring_marks_ai_alert_cards_source_unavailable
ai_cards = await coverage_service._load_ai_alert_card_delivery(project_id="awoooi")
assert ai_cards["status"] == "unavailable"
assert call_kwargs["refresh"] is True
assert call_kwargs["cache_write"] is False
assert ai_cards["summary"]["db_read_status"] == "unavailable"
assert ai_cards["learning_registry"]["status"] == (
"source_unavailable_ai_controlled_repair_required"