fix(agent): hydrate complete runtime receipt chain

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,

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",
+    ]