diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 4a04027d2..a7d2ca6e1 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -4402,6 +4402,30 @@ async def _load_ai_agent_autonomous_runtime_receipt_readback_uncached( if isinstance(exc, TimeoutError) and db_context_stage == "db_context_enter" else type(exc).__name__ ) + direct_core_rows = ( + await _load_core_runtime_receipt_rows_direct( + project_id=project_id, + lookback_hours=params["lookback_hours"], + ) + if error_type == "RuntimeReceiptDbContextTimeout" + else None + ) + if direct_core_rows is not None: + log_controlled_writeback_consumer = ( + await _load_log_controlled_writeback_consumer_readback( + project_id=project_id, + ) + ) + return build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=params["lookback_hours"], + db_read_status="ok", + operation_count_rows=direct_core_rows["operation_counts"], + verifier_count_rows=direct_core_rows["verifier_counts"], + km_count_rows=direct_core_rows["km_counts"], + telegram_count_rows=direct_core_rows["telegram_counts"], + log_controlled_writeback_consumer=log_controlled_writeback_consumer, + ) logger.warning( "ai_agent_autonomous_runtime_receipt_readback_failed", project_id=project_id, @@ -4455,6 +4479,76 @@ async def _load_ai_agent_autonomous_runtime_receipt_readback_uncached( ) +async def _load_core_runtime_receipt_rows_direct( + *, + project_id: str, + lookback_hours: int, +) -> dict[str, list[dict[str, Any]]] | None: + """Read P0 core receipt counters through a short-lived connection.""" + + db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") + try: + import asyncpg + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_RUNTIME_RECEIPT_DB_CONTEXT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_RUNTIME_RECEIPT_SINGLE_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_RUNTIME_RECEIPT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_RUNTIME_RECEIPT_SINGLE_QUERY_TIMEOUT_SECONDS, + ) + + async def _fetch(query_name: str, sql: str, *args: Any) -> list[dict[str, Any]]: + records = await asyncio.wait_for( + conn.fetch(sql, *args), + timeout=_RUNTIME_RECEIPT_SINGLE_QUERY_TIMEOUT_SECONDS, + ) + return [dict(record) for record in records] + + return { + "operation_counts": await _fetch( + "operation_counts", + _RUNTIME_OPERATION_COUNTS_DIRECT_SQL, + lookback_hours, + ), + "verifier_counts": await _fetch( + "verifier_counts", + _RUNTIME_VERIFIER_COUNTS_DIRECT_SQL, + lookback_hours, + ), + "km_counts": await _fetch( + "km_counts", + _RUNTIME_KM_COUNTS_DIRECT_SQL, + project_id, + lookback_hours, + ), + "telegram_counts": await _fetch( + "telegram_counts", + _RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL, + project_id, + lookback_hours, + ), + } + finally: + await conn.close() + except Exception as exc: # pragma: no cover - live fallback only + logger.warning( + "ai_agent_autonomous_runtime_direct_core_readback_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return None + + async def build_ai_agent_autonomous_runtime_control_with_live_readback( *, project_id: str = _DEFAULT_PROJECT_ID, @@ -4522,6 +4616,39 @@ _RUNTIME_OPERATION_COUNTS_SQL = """ ORDER BY 1, status """ +_RUNTIME_OPERATION_COUNTS_DIRECT_SQL = """ + SELECT + CASE + WHEN operation_type = 'km_linked' + AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' + THEN 'log_controlled_writeback_dispatched' + ELSE operation_type + END AS operation_type, + status, + count(*)::int AS total, + count(*) FILTER ( + WHERE created_at >= NOW() - ($1 * INTERVAL '1 hour') + )::int AS recent + FROM automation_operation_log + WHERE ( + operation_type IN ( + 'ansible_candidate_matched', + 'ansible_check_mode_executed', + 'ansible_apply_executed', + 'ansible_learning_writeback_recorded', + 'ansible_rollback_executed', + 'ansible_execution_skipped', + 'log_controlled_writeback_dispatched' + ) + OR ( + operation_type = 'km_linked' + AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' + ) + ) + GROUP BY 1, status + ORDER BY 1, status +""" + _RUNTIME_OPERATION_LATEST_SQL = """ SELECT @@ -4613,6 +4740,19 @@ _RUNTIME_VERIFIER_COUNTS_SQL = """ ORDER BY verification_result """ +_RUNTIME_VERIFIER_COUNTS_DIRECT_SQL = """ + SELECT + coalesce(verification_result, 'missing') AS verification_result, + count(*)::int AS total, + count(*) FILTER ( + WHERE collected_at >= NOW() - ($1 * INTERVAL '1 hour') + )::int AS recent + FROM incident_evidence + WHERE post_execution_state ->> 'apply_op_id' IS NOT NULL + GROUP BY coalesce(verification_result, 'missing') + ORDER BY verification_result +""" + _RUNTIME_VERIFIER_LATEST_SQL = """ SELECT @@ -4649,6 +4789,23 @@ _RUNTIME_KM_COUNTS_SQL = """ ORDER BY status """ +_RUNTIME_KM_COUNTS_DIRECT_SQL = """ + SELECT + status, + count(*)::int AS total, + count(*) FILTER ( + WHERE created_at >= NOW() - ($2 * INTERVAL '1 hour') + )::int AS recent + FROM knowledge_entries + WHERE project_id = $1 + AND ( + path_type LIKE 'ansible_apply_receipt:%' + OR tags::text LIKE '%ansible_controlled_apply%' + ) + GROUP BY status + ORDER BY status +""" + _RUNTIME_KM_LATEST_SQL = """ SELECT @@ -4689,6 +4846,24 @@ _RUNTIME_TELEGRAM_COUNTS_SQL = """ ORDER BY send_status """ +_RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL = """ + SELECT + send_status, + count(*)::int AS total, + count(*) FILTER ( + WHERE queued_at >= NOW() - ($2 * INTERVAL '1 hour') + )::int AS recent + FROM awooop_outbound_message + WHERE project_id = $1 + AND channel_type = 'telegram' + AND ( + source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' + OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' + ) + GROUP BY send_status + ORDER BY send_status +""" + _RUNTIME_TELEGRAM_LATEST_SQL = """ SELECT diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 4b803aec7..6c18aecf0 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -306,6 +306,11 @@ async def test_live_runtime_receipt_db_context_timeout_returns_degraded_payload( assert project_id == "awoooi" return _log_controlled_writeback_consumer_readback() + async def _no_direct_core_readback(*, project_id: str, lookback_hours: int): + assert project_id == "awoooi" + assert lookback_hours == 24 + return None + monkeypatch.setattr( runtime_control_module, "_RUNTIME_RECEIPT_DB_CONTEXT_TIMEOUT_SECONDS", @@ -321,6 +326,11 @@ async def test_live_runtime_receipt_db_context_timeout_returns_degraded_payload( "_load_log_controlled_writeback_consumer_readback", _fake_consumer_readback, ) + monkeypatch.setattr( + runtime_control_module, + "_load_core_runtime_receipt_rows_direct", + _no_direct_core_readback, + ) try: loader = getattr( @@ -338,6 +348,81 @@ async def test_live_runtime_receipt_db_context_timeout_returns_degraded_payload( ) +@pytest.mark.asyncio +async def test_live_runtime_receipt_db_context_timeout_uses_direct_core_fallback( + monkeypatch, +): + runtime_control_module._clear_runtime_receipt_readback_cache() + + async def _fake_consumer_readback(*, project_id: str): + assert project_id == "awoooi" + return _log_controlled_writeback_consumer_readback() + + async def _fake_direct_core_readback(*, project_id: str, lookback_hours: int): + assert project_id == "awoooi" + assert lookback_hours == 24 + return { + "operation_counts": [ + { + "operation_type": "ansible_apply_executed", + "status": "success", + "total": 5, + "recent": 2, + } + ], + "verifier_counts": [ + {"verification_result": "success", "total": 5, "recent": 2} + ], + "km_counts": [ + {"status": "linked", "total": 5, "recent": 2}, + ], + "telegram_counts": [ + {"send_status": "sent", "total": 5, "recent": 2}, + ], + } + + monkeypatch.setattr( + runtime_control_module, + "_RUNTIME_RECEIPT_DB_CONTEXT_TIMEOUT_SECONDS", + 0.01, + ) + monkeypatch.setattr( + runtime_control_module, + "get_db_context", + lambda project_id: _SlowEnterRuntimeDbContext(), + ) + monkeypatch.setattr( + runtime_control_module, + "_load_log_controlled_writeback_consumer_readback", + _fake_consumer_readback, + ) + monkeypatch.setattr( + runtime_control_module, + "_load_core_runtime_receipt_rows_direct", + _fake_direct_core_readback, + ) + + try: + loader = getattr( + runtime_control_module, + "_load_ai_agent_autonomous_runtime_receipt_readback_uncached", + ) + readback = await loader() + finally: + runtime_control_module._clear_runtime_receipt_readback_cache() + + assert readback["db_read_status"] == "ok" + assert readback["ansible_apply_executed"]["total"] == 5 + assert readback["runtime_receipt_readback_recovery"]["status"] == ( + "completed_live_runtime_receipts_observed" + ) + progress_items = { + item["work_item_id"]: item + for item in readback["work_item_progress"]["ordered_items"] + } + assert progress_items["P0-A-runtime-truth"]["status"] == "completed" + + @pytest.mark.asyncio async def test_live_runtime_receipt_lock_timeout_returns_degraded_payload( monkeypatch,