diff --git a/apps/api/src/services/telegram_alert_ai_automation_matrix.py b/apps/api/src/services/telegram_alert_ai_automation_matrix.py index 5f46c868a..91d65e255 100644 --- a/apps/api/src/services/telegram_alert_ai_automation_matrix.py +++ b/apps/api/src/services/telegram_alert_ai_automation_matrix.py @@ -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(): diff --git a/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py b/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py index b5ae81acb..7adfabf01 100644 --- a/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py +++ b/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py @@ -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], diff --git a/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py b/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py index b87629d66..974124f3a 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py @@ -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(