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 e5a94d1eb..2ea6d56f3 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -849,6 +849,7 @@ def _build_runtime_receipt_readback_recovery( db_read_status: str, log_integration_taxonomy: Mapping[str, Any], error_type: str | None, + partial_query_failures: Iterable[Mapping[str, Any]] = (), ) -> dict[str, Any]: """Expose the next controlled action when live receipt/log evidence is absent.""" @@ -867,9 +868,22 @@ def _build_runtime_receipt_readback_recovery( recent_classified_event_total = _int_value( taxonomy_rollups.get("recent_classified_event_total") ) - db_unavailable = db_read_status != "ok" + db_unavailable = db_read_status in {"not_queried", "unavailable"} + partial_failures = [dict(item) for item in partial_query_failures] + db_partial = db_read_status == "partial" no_live_log_events = classified_event_total <= 0 - if db_unavailable: + if db_partial and not no_live_log_events: + status = "degraded_runtime_receipt_partial_query_failures" + safe_next_action_id = ( + "repair_partial_runtime_receipt_queries_then_rerun_log_integration_taxonomy" + ) + safe_next_action_stage = "runtime_receipt_partial_query_recovery" + safe_next_action = ( + "Optimize or fallback the failed receipt queries while keeping successful " + "LOG source families visible to KM/RAG/MCP/PlayBook consumers." + ) + blocker_fields = ["partial_query_failures"] + elif db_unavailable: status = "blocked_runtime_receipt_db_readback_unavailable" safe_next_action_id = ( "repair_runtime_receipt_db_readback_then_rerun_log_integration_taxonomy" @@ -918,6 +932,7 @@ def _build_runtime_receipt_readback_recovery( taxonomy_rollups.get("active_source_family_count") ), "inactive_source_family_ids": inactive_source_family_ids, + "partial_query_failures": partial_failures, "blocker_fields": blocker_fields, "operation_boundaries": { "readback_only": True, @@ -2141,7 +2156,7 @@ def _build_work_item_progress( "completed" if runtime_truth_complete else "blocked" - if db_read_status != "ok" + if db_read_status in {"not_queried", "unavailable"} else "in_progress" ), "exit_criteria": "production API reports db_read_status=ok and live executor receipts", @@ -2149,7 +2164,9 @@ def _build_work_item_progress( if runtime_truth_complete else ( "runtime_receipt_db_readback_unavailable" - if db_read_status != "ok" + if db_read_status in {"not_queried", "unavailable"} + else "runtime_receipt_partial_query_failures" + if db_read_status == "partial" else "live_log_classified_event_total_zero" ), "db_read_status": db_read_status, @@ -3043,6 +3060,7 @@ def build_runtime_receipt_readback_from_rows( grouped_alert_event_count_rows: Iterable[Mapping[str, Any] | Any] = (), log_controlled_writeback_consumer: Mapping[str, Any] | None = None, error_type: str | None = None, + partial_query_failures: Iterable[Mapping[str, Any]] = (), ) -> dict[str, Any]: """Build the live executor receipt readback from already-fetched rows.""" @@ -3135,6 +3153,7 @@ def build_runtime_receipt_readback_from_rows( db_read_status=db_read_status, log_integration_taxonomy=log_integration_taxonomy, error_type=error_type, + partial_query_failures=partial_query_failures, ) agent_decision_wiring = _build_agent_decision_wiring( operation_summary=operation_summary, @@ -3198,6 +3217,7 @@ def build_runtime_receipt_readback_from_rows( "project_id": project_id, "lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)), "db_read_status": db_read_status, + "partial_query_failures": [dict(item) for item in partial_query_failures], "writes_on_read": False, "ansible_operations": { "counts": operation_summary, @@ -3972,11 +3992,15 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( "limit": max(1, int(limit or 20)), } log_controlled_writeback_consumer: dict[str, Any] | None = None + partial_query_failures: list[dict[str, str]] = [] try: async with get_db_context(project_id) as db: - await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + async def _set_statement_timeout() -> None: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) - async def _safe_aux_rows( + await _set_statement_timeout() + + async def _safe_rows( query_name: str, sql: str, fallback_sql: str | None = None, @@ -3984,89 +4008,131 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( try: return (await db.execute(text(sql), params)).mappings().all() except Exception as exc: # pragma: no cover - depends on live schema drift + partial_query_failures.append({ + "query_name": query_name, + "error_type": type(exc).__name__, + }) logger.warning( - "ai_agent_autonomous_runtime_trace_aux_read_failed", + "ai_agent_autonomous_runtime_trace_read_failed", project_id=project_id, query_name=query_name, error_type=type(exc).__name__, ) + rollback = getattr(db, "rollback", None) + if callable(rollback): + try: + await rollback() + await _set_statement_timeout() + except Exception as reset_exc: # pragma: no cover - live DB state + logger.warning( + "ai_agent_autonomous_runtime_trace_reset_failed", + project_id=project_id, + query_name=query_name, + error_type=type(reset_exc).__name__, + ) if fallback_sql: try: return (await db.execute(text(fallback_sql), params)).mappings().all() except Exception as fallback_exc: # pragma: no cover - live schema drift + partial_query_failures.append({ + "query_name": f"{query_name}_fallback", + "error_type": type(fallback_exc).__name__, + }) logger.warning( "ai_agent_autonomous_runtime_trace_aux_fallback_failed", project_id=project_id, query_name=query_name, error_type=type(fallback_exc).__name__, ) + rollback = getattr(db, "rollback", None) + if callable(rollback): + try: + await rollback() + await _set_statement_timeout() + except Exception as reset_exc: # pragma: no cover + logger.warning( + "ai_agent_autonomous_runtime_trace_reset_failed", + project_id=project_id, + query_name=f"{query_name}_fallback", + error_type=type(reset_exc).__name__, + ) return [] - operation_counts = ( - await db.execute(text(_RUNTIME_OPERATION_COUNTS_SQL), params) - ).mappings().all() - operation_latest = ( - await db.execute(text(_RUNTIME_OPERATION_LATEST_SQL), params) - ).mappings().all() - auto_repair_counts = ( - await db.execute(text(_RUNTIME_AUTO_REPAIR_COUNTS_SQL), params) - ).mappings().all() - auto_repair_latest = ( - await db.execute(text(_RUNTIME_AUTO_REPAIR_LATEST_SQL), params) - ).mappings().all() - verifier_counts = ( - await db.execute(text(_RUNTIME_VERIFIER_COUNTS_SQL), params) - ).mappings().all() - verifier_latest = ( - await db.execute(text(_RUNTIME_VERIFIER_LATEST_SQL), params) - ).mappings().all() - km_counts = ( - await db.execute(text(_RUNTIME_KM_COUNTS_SQL), params) - ).mappings().all() - km_latest = ( - await db.execute(text(_RUNTIME_KM_LATEST_SQL), params) - ).mappings().all() - telegram_counts = ( - await db.execute(text(_RUNTIME_TELEGRAM_COUNTS_SQL), params) - ).mappings().all() - telegram_latest = ( - await db.execute(text(_RUNTIME_TELEGRAM_LATEST_SQL), params) - ).mappings().all() - mcp_gateway_counts = await _safe_aux_rows( + operation_counts = await _safe_rows( + "operation_counts", + _RUNTIME_OPERATION_COUNTS_SQL, + ) + operation_latest = await _safe_rows( + "operation_latest", + _RUNTIME_OPERATION_LATEST_SQL, + ) + auto_repair_counts = await _safe_rows( + "auto_repair_counts", + _RUNTIME_AUTO_REPAIR_COUNTS_SQL, + ) + auto_repair_latest = await _safe_rows( + "auto_repair_latest", + _RUNTIME_AUTO_REPAIR_LATEST_SQL, + ) + verifier_counts = await _safe_rows( + "verifier_counts", + _RUNTIME_VERIFIER_COUNTS_SQL, + ) + verifier_latest = await _safe_rows( + "verifier_latest", + _RUNTIME_VERIFIER_LATEST_SQL, + ) + km_counts = await _safe_rows( + "km_counts", + _RUNTIME_KM_COUNTS_SQL, + ) + km_latest = await _safe_rows( + "km_latest", + _RUNTIME_KM_LATEST_SQL, + ) + telegram_counts = await _safe_rows( + "telegram_counts", + _RUNTIME_TELEGRAM_COUNTS_SQL, + ) + telegram_latest = await _safe_rows( + "telegram_latest", + _RUNTIME_TELEGRAM_LATEST_SQL, + ) + mcp_gateway_counts = await _safe_rows( "mcp_gateway_counts", _RUNTIME_MCP_GATEWAY_COUNTS_SQL, ) - legacy_mcp_counts = await _safe_aux_rows( + legacy_mcp_counts = await _safe_rows( "legacy_mcp_counts", _RUNTIME_LEGACY_MCP_COUNTS_SQL, ) - service_log_counts = await _safe_aux_rows( + service_log_counts = await _safe_rows( "service_log_counts", _RUNTIME_SERVICE_LOG_COUNTS_SQL, ) - executor_log_counts = await _safe_aux_rows( + executor_log_counts = await _safe_rows( "executor_log_counts", _RUNTIME_EXECUTOR_LOG_COUNTS_SQL, ) - timeline_counts = await _safe_aux_rows( + timeline_counts = await _safe_rows( "timeline_counts", _RUNTIME_TIMELINE_COUNTS_SQL, _RUNTIME_TIMELINE_COUNTS_FALLBACK_SQL, ) - playbook_trust_counts = await _safe_aux_rows( + playbook_trust_counts = await _safe_rows( "playbook_trust_counts", _RUNTIME_PLAYBOOK_TRUST_COUNTS_SQL, _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL, ) - alert_operation_counts = await _safe_aux_rows( + alert_operation_counts = await _safe_rows( "alert_operation_counts", _RUNTIME_ALERT_OPERATION_COUNTS_SQL, ) - alertmanager_event_counts = await _safe_aux_rows( + alertmanager_event_counts = await _safe_rows( "alertmanager_event_counts", _RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL, ) - grouped_alert_event_counts = await _safe_aux_rows( + grouped_alert_event_counts = await _safe_rows( "grouped_alert_event_counts", _RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL, ) @@ -4097,7 +4163,7 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( return build_runtime_receipt_readback_from_rows( project_id=project_id, lookback_hours=params["lookback_hours"], - db_read_status="ok", + db_read_status="partial" if partial_query_failures else "ok", operation_count_rows=operation_counts, operation_latest_rows=operation_latest, auto_repair_count_rows=auto_repair_counts, @@ -4118,6 +4184,8 @@ async def load_ai_agent_autonomous_runtime_receipt_readback( alertmanager_event_count_rows=alertmanager_event_counts, grouped_alert_event_count_rows=grouped_alert_event_counts, log_controlled_writeback_consumer=log_controlled_writeback_consumer, + error_type="partial_query_failures" if partial_query_failures else None, + partial_query_failures=partial_query_failures, ) 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 f3e1ee0eb..956ac4934 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -981,6 +981,40 @@ def test_runtime_receipt_recovery_flags_zero_live_log_events(): assert readback["work_item_progress"]["rollups"]["not_started_count"] == 10 +def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible(): + readback = build_runtime_receipt_readback_from_rows( + project_id="awoooi", + db_read_status="partial", + service_log_count_rows=[ + {"status": "sanitized_recent_logs", "total": 12, "recent": 3}, + ], + partial_query_failures=[ + {"query_name": "km_counts", "error_type": "TimeoutError"}, + ], + error_type="partial_query_failures", + ) + + recovery = readback["runtime_receipt_readback_recovery"] + assert recovery["status"] == "degraded_runtime_receipt_partial_query_failures" + assert recovery["safe_next_action_id"] == ( + "repair_partial_runtime_receipt_queries_then_rerun_log_integration_taxonomy" + ) + assert recovery["partial_query_failures"] == [ + {"query_name": "km_counts", "error_type": "TimeoutError"} + ] + assert recovery["classified_event_total"] == 12 + assert recovery["active_source_family_count"] == 1 + + progress_items = { + item["work_item_id"]: item + for item in readback["work_item_progress"]["ordered_items"] + } + assert progress_items["P0-A-runtime-truth"]["status"] == "in_progress" + assert progress_items["P0-A-runtime-truth"]["blocker"] == ( + "runtime_receipt_partial_query_failures" + ) + + def test_runtime_receipt_work_items_use_learning_receipts_without_latest_telegram_gate(): apply_op_id = "2f8ef5c8-fd4e-4950-99e9-dc9e61150cab" incident_id = "INC-20260629-LEARNING"