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

fix(agents): bound telegram alert receipt readback
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / tests (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-09 16:21:25 +08:00
parent 8e6e5c7e80
commit 45d2357892
3 changed files with 226 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
apps/api/src/services/telegram_alert_ai_automation_matrix.py
View File

@@ -24,6 +24,12 @@ _OWNER_ACCEPTANCE = "telegram-notification-egress-owner-response-acceptance.snap
_MIGRATION_PLAN = "telegram-notification-egress-migration-plan-draft.snapshot.json"
_READABILITY_GUARD = "telegram-alert-readability-guard.snapshot.json"
_DIGEST_POLICY_PATTERN = "ai_agent_telegram_action_required_digest_policy_*.json"
_AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE = (
"apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx"
)
_AWOOOP_VERIFIED_CONTEXT_SURFACE_MARKER = (
"telegram-alert-post-apply-verifier-readback"
)
def load_latest_telegram_alert_ai_automation_matrix(
@@ -656,8 +662,8 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, Any]:
"telegram_alert_learning_context_post_apply_verifier_v1",
),
"telegram_alert_post_apply_verifier_awooop_surface_present": (
"apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx",
"telegram-alert-post-apply-verifier-readback",
_AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE,
_AWOOOP_VERIFIED_CONTEXT_SURFACE_MARKER,
),
}
result: dict[str, int | bool] = {}
@@ -701,6 +707,11 @@ def _read_source_contract_text(repo_root: Path, relative_path: str) -> str:
api_prefix = "apps/api/"
if relative_path.startswith(api_prefix):
candidates.append(repo_root / relative_path.removeprefix(api_prefix))
if (
relative_path == _AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE
and not (repo_root / "apps").exists()
):
candidates.append(repo_root / "src/services/telegram_alert_ai_automation_matrix.py")
for path in candidates:
if path.exists():

164
apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py
View File

@@ -34,6 +34,9 @@ POST_VERIFIER_ROUTE = "/api/v1/agents/telegram-alert-monitoring-coverage-readbac
APPLY_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply"
READBACK_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-readback"
_DB_RETRY_DELAYS_SECONDS = (0.0, 0.25, 1.0)
_COVERAGE_READ_TIMEOUT_SECONDS = 12.0
_DB_APPLY_TIMEOUT_SECONDS = 12.0
_DB_READBACK_TIMEOUT_SECONDS = 8.0
logger = get_logger(__name__)
@@ -43,9 +46,24 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
) -> dict[str, Any]:
"""Persist idempotent metadata live receipt rows for coverage batches."""
coverage = await load_latest_telegram_alert_monitoring_coverage_readback(
project_id=project_id,
)
try:
coverage = await asyncio.wait_for(
load_latest_telegram_alert_monitoring_coverage_readback(
project_id=project_id,
),
timeout=_COVERAGE_READ_TIMEOUT_SECONDS,
)
except Exception as exc: # pragma: no cover - live dependency pressure
logger.warning(
"telegram_monitoring_live_receipt_apply_coverage_unavailable",
project_id=project_id,
error_type=type(exc).__name__,
)
return _fallback_apply_receipt(
project_id=project_id,
error_type=type(exc).__name__,
)
receipt = build_telegram_alert_monitoring_live_receipt_apply_receipt(coverage)
if receipt["active_blockers"]:
receipt["apply_result"] = {
@@ -59,9 +77,12 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
return receipt
try:
ledger_operation_type, rows = await _write_live_receipts_with_retry(
project_id=project_id,
receipt=receipt,
ledger_operation_type, rows = await asyncio.wait_for(
_write_live_receipts_with_retry(
project_id=project_id,
receipt=receipt,
),
timeout=_DB_APPLY_TIMEOUT_SECONDS,
)
except Exception as exc: # pragma: no cover - live DB pool pressure
logger.warning(
@@ -111,11 +132,30 @@ async def load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
) -> dict[str, Any]:
"""Read back metadata live receipt rows and map them to coverage batches."""
coverage = await load_latest_telegram_alert_monitoring_coverage_readback(
project_id=project_id,
)
try:
rows = await _load_live_receipt_rows_with_retry(project_id=project_id)
coverage = await asyncio.wait_for(
load_latest_telegram_alert_monitoring_coverage_readback(
project_id=project_id,
),
timeout=_COVERAGE_READ_TIMEOUT_SECONDS,
)
except Exception as exc: # pragma: no cover - live dependency pressure
logger.warning(
"telegram_monitoring_live_receipt_apply_readback_coverage_unavailable",
project_id=project_id,
error_type=type(exc).__name__,
)
return _fallback_apply_readback(
coverage=_coverage_unavailable(project_id=project_id),
project_id=project_id,
error_type=type(exc).__name__,
)
try:
rows = await asyncio.wait_for(
_load_live_receipt_rows_with_retry(project_id=project_id),
timeout=_DB_READBACK_TIMEOUT_SECONDS,
)
except Exception as exc: # pragma: no cover - live DB pool pressure
logger.warning(
"telegram_monitoring_live_receipt_apply_readback_db_unavailable",
@@ -214,6 +254,79 @@ def build_telegram_alert_monitoring_live_receipt_apply_receipt(
}
def _fallback_apply_receipt(
*,
project_id: str,
error_type: str,
) -> dict[str, Any]:
return {
"schema_version": APPLY_SCHEMA_VERSION,
"priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE",
"scope": "telegram_alert_monitoring_live_receipt_apply",
"project_id": project_id,
"status": "blocked_waiting_telegram_monitoring_live_receipt_coverage_readback",
"readback": {
"workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY",
"workplan_title": (
"Telegram monitoring live receipt metadata apply waits for "
"bounded coverage readback"
),
"source_schema_version": None,
"source_status": "unavailable",
"source_error_type": error_type,
"operation_type": OPERATION_TYPE,
"executor_route": EXECUTOR_ROUTE,
"safe_next_step": "retry_coverage_readback_then_apply_metadata_live_receipts",
},
"live_receipt_apply_receipts": [],
"rollups": {
"source_live_receipt_batch_count": 0,
"ready_source_live_receipt_batch_count": 0,
"source_surface_count": 0,
"source_write_capable_surface_count": 0,
"metadata_only_receipt_count": 0,
"post_apply_verifier_ref_count": 0,
"controlled_live_receipt_apply_ready": False,
"runtime_target_write_performed": False,
"live_receipt_apply_row_count": 0,
"inserted_live_receipt_apply_row_count": 0,
"existing_live_receipt_apply_row_count": 0,
},
"active_blockers": [
"telegram_monitoring_live_receipt_coverage_readback_unavailable"
],
"apply_result": {
"operation_type": OPERATION_TYPE,
"ledger_operation_type": None,
"semantic_operation_type": OPERATION_TYPE,
"db_write_status": "not_attempted",
"error_type": error_type,
"live_receipt_apply_row_count": 0,
"rows": [],
},
"operation_boundaries": {
"metadata_ledger_write_only": True,
"monitoring_live_receipt_metadata_write_performed": False,
"runtime_target_write_performed": False,
"monitoring_inventory_source_mutated": False,
"monitoring_live_receipt_acceptance_mutated": False,
"km_write_performed": False,
"rag_index_write_performed": False,
"playbook_trust_write_performed": False,
"mcp_tool_call_performed": False,
"agent_runtime_action_performed": False,
"telegram_send_performed": False,
"bot_api_call_performed": False,
"alertmanager_reload_performed": False,
"workflow_trigger_performed": False,
"raw_alert_payload_stored": False,
"raw_payload_included": False,
"secret_value_read": False,
"github_api_used": False,
},
}
def build_telegram_alert_monitoring_live_receipt_apply_readback(
*,
coverage: Mapping[str, Any],
@@ -784,6 +897,37 @@ def _readback_active_blockers(
return _unique(blockers)
def _coverage_unavailable(*, project_id: str) -> dict[str, Any]:
return {
"schema_version": "telegram_alert_monitoring_coverage_readback_v1",
"project_id": project_id,
"status": "blocked_telegram_alert_monitoring_coverage_readback_unavailable",
"summary": {
"monitoring_surface_count": 0,
"monitoring_live_receipt_gap_count": 0,
"ai_controlled_live_receipt_batch_count": 0,
"source_contract_ready": False,
},
"ai_controlled_live_receipt_batches": [],
"active_blockers": [
"telegram_monitoring_live_receipt_coverage_readback_unavailable"
],
"operation_boundaries": {
"metadata_read_performed": False,
"runtime_db_read_performed": False,
"telegram_send_performed": False,
"bot_api_call_performed": False,
"gateway_queue_write_performed": False,
"receiver_route_changed": False,
"alertmanager_reload_performed": False,
"runtime_write_performed": False,
"raw_alert_payload_stored": False,
"secret_value_read": False,
"github_api_used": False,
},
}
def _fallback_apply_readback(
*,
coverage: Mapping[str, Any],

59
apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py
View File

@@ -16,6 +16,7 @@ from src.services.telegram_alert_monitoring_live_receipt_apply import (
apply_latest_telegram_alert_monitoring_live_receipts,
build_telegram_alert_monitoring_live_receipt_apply_readback,
build_telegram_alert_monitoring_live_receipt_apply_receipt,
load_latest_telegram_alert_monitoring_live_receipt_apply_readback,
)
@@ -352,6 +353,64 @@ async def test_telegram_monitoring_live_receipt_apply_degrades_on_db_write_press
assert payload["operation_boundaries"]["runtime_target_write_performed"] is False
@pytest.mark.asyncio
async def test_telegram_monitoring_live_receipt_apply_degrades_on_coverage_pressure(
monkeypatch,
):
async def failing_coverage(*, project_id: str):
assert project_id == "awoooi"
raise TimeoutError("coverage readback timeout")
monkeypatch.setattr(
apply_module,
"load_latest_telegram_alert_monitoring_coverage_readback",
failing_coverage,
)
payload = await apply_latest_telegram_alert_monitoring_live_receipts(
project_id="awoooi"
)
assert payload["status"] == (
"blocked_waiting_telegram_monitoring_live_receipt_coverage_readback"
)
assert "telegram_monitoring_live_receipt_coverage_readback_unavailable" in payload[
"active_blockers"
]
assert payload["apply_result"]["db_write_status"] == "not_attempted"
assert payload["operation_boundaries"]["runtime_target_write_performed"] is False
assert payload["operation_boundaries"]["telegram_send_performed"] is False
@pytest.mark.asyncio
async def test_telegram_monitoring_live_receipt_apply_readback_degrades_on_coverage_pressure(
monkeypatch,
):
async def failing_coverage(*, project_id: str):
assert project_id == "awoooi"
raise TimeoutError("coverage readback timeout")
monkeypatch.setattr(
apply_module,
"load_latest_telegram_alert_monitoring_coverage_readback",
failing_coverage,
)
payload = await load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
project_id="awoooi"
)
assert payload["status"] == (
"blocked_waiting_telegram_monitoring_live_receipt_apply_db_readback"
)
assert "telegram_monitoring_live_receipt_apply_db_readback_unavailable" in payload[
"active_blockers"
]
assert payload["rollups"]["expected_live_receipt_batch_count"] == 0
assert payload["operation_boundaries"]["metadata_ledger_read_performed"] is False
assert payload["operation_boundaries"]["telegram_send_performed"] is False
def test_telegram_monitoring_live_receipt_apply_endpoint_returns_receipt(monkeypatch):
async def fake_apply():
payload = build_telegram_alert_monitoring_live_receipt_apply_receipt(