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 389fa3087..371bace60 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -3228,6 +3228,58 @@ def _operation_by_id( return None +def _missing_runtime_operation_chain_ref_ids( + rows: Iterable[Mapping[str, Any] | Any], +) -> list[str]: + """Return exact apply/check/candidate receipt ids missing from a bounded read.""" + + operation_rows = [_row_mapping(row) for row in rows] + latest_apply = _first_operation(operation_rows, "ansible_apply_executed") + if latest_apply is None: + return [] + + check_mode_op_id = str( + latest_apply.get("check_mode_op_id") + or latest_apply.get("parent_op_id") + or "" + ) + latest_check = _operation_by_id(operation_rows, check_mode_op_id) + candidate_op_id = str( + (latest_check or {}).get("parent_op_id") + or (latest_check or {}).get("source_candidate_op_id") + or latest_apply.get("source_candidate_op_id") + or "" + ) + present_ids = { + str(row.get("op_id") or "") + for row in operation_rows + if str(row.get("op_id") or "") + } + refs = [ + str(latest_apply.get("op_id") or ""), + check_mode_op_id, + candidate_op_id, + ] + return list(dict.fromkeys(ref for ref in refs if ref and ref not in present_ids)) + + +def _merge_runtime_operation_rows( + primary_rows: Iterable[Mapping[str, Any] | Any], + supplemental_rows: Iterable[Mapping[str, Any] | Any], +) -> list[dict[str, Any]]: + merged: list[dict[str, Any]] = [] + seen_ids: set[str] = set() + for raw_row in (*list(primary_rows), *list(supplemental_rows)): + row = _row_mapping(raw_row) + op_id = str(row.get("op_id") or "") + if op_id and op_id in seen_ids: + continue + if op_id: + seen_ids.add(op_id) + merged.append(row) + return merged + + def _stage_status(row: Mapping[str, Any] | None, *, fallback_status: str | None = None) -> str: if row is None: return fallback_status or "missing" @@ -5297,6 +5349,30 @@ async def _load_ai_agent_autonomous_runtime_receipt_readback_uncached( "operation_latest", _RUNTIME_OPERATION_LATEST_SQL, ) + for chain_pass in range(2): + missing_chain_refs = _missing_runtime_operation_chain_ref_ids( + operation_latest + ) + if not missing_chain_refs: + break + padded_refs = [*missing_chain_refs[:3], "", ""][:3] + params.update( + { + "operation_chain_ref_1": padded_refs[0], + "operation_chain_ref_2": padded_refs[1], + "operation_chain_ref_3": padded_refs[2], + } + ) + operation_chain_rows = await _safe_rows( + f"operation_chain_{chain_pass + 1}", + _RUNTIME_OPERATION_CHAIN_SQL, + ) + if not operation_chain_rows: + break + operation_latest = _merge_runtime_operation_rows( + operation_latest, + operation_chain_rows, + ) auto_repair_counts = await _safe_rows( "auto_repair_counts", _RUNTIME_AUTO_REPAIR_COUNTS_SQL, @@ -5518,17 +5594,37 @@ async def _load_core_runtime_receipt_rows_direct( ) return [] + operation_latest = await _fetch( + "operation_latest", + _RUNTIME_OPERATION_LATEST_DIRECT_SQL, + 20, + ) + for chain_pass in range(2): + missing_chain_refs = _missing_runtime_operation_chain_ref_ids( + operation_latest + ) + if not missing_chain_refs: + break + padded_refs = [*missing_chain_refs[:3], "", ""][:3] + operation_chain_rows = await _fetch( + f"operation_chain_{chain_pass + 1}", + _RUNTIME_OPERATION_CHAIN_DIRECT_SQL, + *padded_refs, + ) + if not operation_chain_rows: + break + operation_latest = _merge_runtime_operation_rows( + operation_latest, + operation_chain_rows, + ) + return { "operation_counts": await _fetch( "operation_counts", _RUNTIME_OPERATION_COUNTS_DIRECT_SQL, lookback_hours, ), - "operation_latest": await _fetch( - "operation_latest", - _RUNTIME_OPERATION_LATEST_DIRECT_SQL, - 20, - ), + "operation_latest": operation_latest, "auto_repair_counts": await _fetch( "auto_repair_counts", _RUNTIME_AUTO_REPAIR_COUNTS_DIRECT_SQL, @@ -5836,6 +5932,78 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """ """ +_RUNTIME_OPERATION_CHAIN_SQL = """ + SELECT + op_id::text AS op_id, + parent_op_id::text AS parent_op_id, + 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, + actor, + CASE + WHEN operation_type = 'ansible_candidate_matched' + THEN op_id::text + ELSE input ->> 'automation_run_id' + END AS automation_run_id, + coalesce(incident_id::text, input ->> 'incident_id') AS incident_id, + input ->> 'catalog_id' AS catalog_id, + coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path, + input ->> 'execution_mode' AS execution_mode, + input ->> 'source_candidate_op_id' AS source_candidate_op_id, + input ->> 'check_mode_op_id' AS check_mode_op_id, + input ->> 'risk_level' AS risk_level, + input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, + duration_ms, + created_at + FROM automation_operation_log + WHERE op_id::text IN ( + :operation_chain_ref_1, + :operation_chain_ref_2, + :operation_chain_ref_3 + ) + ORDER BY created_at DESC +""" + + +_RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """ + SELECT + op_id::text AS op_id, + parent_op_id::text AS parent_op_id, + 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, + actor, + CASE + WHEN operation_type = 'ansible_candidate_matched' + THEN op_id::text + ELSE input ->> 'automation_run_id' + END AS automation_run_id, + coalesce(incident_id::text, input ->> 'incident_id') AS incident_id, + input ->> 'catalog_id' AS catalog_id, + coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path, + input ->> 'execution_mode' AS execution_mode, + input ->> 'source_candidate_op_id' AS source_candidate_op_id, + input ->> 'check_mode_op_id' AS check_mode_op_id, + input ->> 'risk_level' AS risk_level, + input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, + duration_ms, + created_at + FROM automation_operation_log + WHERE op_id::text IN ($1, $2, $3) + ORDER BY created_at DESC +""" + + _RUNTIME_AUTO_REPAIR_COUNTS_SQL = """ SELECT CASE WHEN success THEN 'success' ELSE 'failed' END AS result_status, 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 d73fa5776..932e690f5 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -2412,3 +2412,41 @@ def test_runtime_execution_loop_ledger_does_not_mix_unrelated_check_mode_rows(): assert ledger["stages"][1]["ref_id"] == "expected-check-mode-op" assert ledger["stages"][1]["status"] == "missing" assert "candidate" in ledger["missing_stage_ids"] + + +def test_runtime_operation_chain_ref_ids_recover_rows_outside_latest_window(): + rows = [ + { + "op_id": "apply-op", + "parent_op_id": "check-op", + "operation_type": "ansible_apply_executed", + "check_mode_op_id": "check-op", + "source_candidate_op_id": "candidate-op", + } + ] + + missing_refs = runtime_control_module._missing_runtime_operation_chain_ref_ids( + rows + ) + assert missing_refs == ["check-op", "candidate-op"] + + merged = runtime_control_module._merge_runtime_operation_rows( + rows, + [ + { + "op_id": "check-op", + "parent_op_id": "candidate-op", + "operation_type": "ansible_check_mode_executed", + }, + { + "op_id": "candidate-op", + "operation_type": "ansible_candidate_matched", + }, + ], + ) + assert runtime_control_module._missing_runtime_operation_chain_ref_ids(merged) == [] + assert [row["op_id"] for row in merged] == [ + "apply-op", + "check-op", + "candidate-op", + ]