fix(agent): preserve log consumer readback on trace failure

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -3249,6 +3249,9 @@ def _attach_runtime_receipt_readback(
     log_consumer_blockers = log_consumer.get("active_blockers")
     if not isinstance(log_consumer_blockers, list):
         log_consumer_blockers = []
+    log_consumer_dispatch_ledger_count = _int_value(
+        log_consumer_rollups.get("dispatch_ledger_row_count")
+    )
     operation_counts = (readback.get("ansible_operations") or {}).get("counts")
     if not isinstance(operation_counts, Mapping):
         operation_counts = {}
@@ -3373,11 +3376,17 @@ def _attach_runtime_receipt_readback(
             )
         ),
         "live_log_controlled_writeback_dispatch_count": _int_value(
-            log_dispatch_summary.get("total")
+            max(
+                _int_value(log_dispatch_summary.get("total")),
+                log_consumer_dispatch_ledger_count,
+            )
         ),
         "live_log_controlled_writeback_recent_dispatch_count": _int_value(
             log_dispatch_summary.get("recent")
         ),
+        "live_log_controlled_writeback_consumer_dispatch_ledger_count": (
+            log_consumer_dispatch_ledger_count
+        ),
         "live_log_controlled_writeback_consumer_binding_count": _int_value(
             log_consumer_rollups.get("consumer_binding_count")
         ),
@@ -3834,6 +3843,7 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
         "lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
         "limit": max(1, int(limit or 20)),
     }
+    log_controlled_writeback_consumer: dict[str, Any] | None = None
     try:
         async with get_db_context(project_id) as db:
             await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
@@ -3932,24 +3942,30 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
                 "grouped_alert_event_counts",
                 _RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL,
             )
-            log_controlled_writeback_consumer = (
-                await _load_log_controlled_writeback_consumer_readback(
-                    project_id=project_id,
-                )
-            )
     except Exception as exc:
         logger.warning(
             "ai_agent_autonomous_runtime_receipt_readback_failed",
             project_id=project_id,
             error_type=type(exc).__name__,
         )
+        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="unavailable",
             error_type=type(exc).__name__,
+            log_controlled_writeback_consumer=log_controlled_writeback_consumer,
         )
 
+    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"],

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -1,3 +1,6 @@
+import pytest
+
+from src.services import ai_agent_autonomous_runtime_control as runtime_control_module
 from src.services.ai_agent_autonomous_runtime_control import (
     _RUNTIME_ALERT_OPERATION_COUNTS_SQL,
     _RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL,
@@ -10,6 +13,19 @@ from src.services.ai_agent_autonomous_runtime_control import (
 )
 
 
+class _FailingRuntimeDb:
+    async def execute(self, *_args, **_kwargs):
+        raise RuntimeError("runtime trace query unavailable")
+
+
+class _FailingRuntimeDbContext:
+    async def __aenter__(self) -> _FailingRuntimeDb:
+        return _FailingRuntimeDb()
+
+    async def __aexit__(self, _exc_type, _exc, _tb) -> bool:
+        return False
+
+
 def _assert_log_controlled_writeback_executor(payload: dict):
     assert payload["schema_version"] == "ai_agent_log_controlled_writeback_executor_readback_v1"
     assert payload["priority"] == "P1-LOG-KM-RAG-MCP-PLAYBOOK"
@@ -190,6 +206,62 @@ def _assert_log_controlled_writeback_consumer(payload: dict):
     assert payload["operation_boundaries"]["github_api_used"] is False
 
 
+@pytest.mark.asyncio
+async def test_live_runtime_receipt_keeps_consumer_readback_when_trace_query_fails(
+    monkeypatch,
+):
+    async def _fake_consumer_readback(*, project_id: str):
+        assert project_id == "awoooi"
+        return _log_controlled_writeback_consumer_readback()
+
+    monkeypatch.setattr(
+        runtime_control_module,
+        "get_db_context",
+        lambda project_id: _FailingRuntimeDbContext(),
+    )
+    monkeypatch.setattr(
+        runtime_control_module,
+        "_load_log_controlled_writeback_consumer_readback",
+        _fake_consumer_readback,
+    )
+
+    readback = (
+        await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
+    )
+
+    assert readback["db_read_status"] == "unavailable"
+    assert readback["error"]["type"] == "RuntimeError"
+    _assert_log_controlled_writeback_consumer(
+        readback["log_controlled_writeback_consumer"]
+    )
+
+
+def test_runtime_control_rollups_project_consumer_dispatch_ledger_if_trace_missing():
+    payload = build_ai_agent_autonomous_runtime_control()
+    readback = build_runtime_receipt_readback_from_rows(
+        project_id="awoooi",
+        db_read_status="unavailable",
+        log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(),
+    )
+
+    runtime_control_module._attach_runtime_receipt_readback(payload, readback)
+
+    assert payload["rollups"]["live_log_controlled_writeback_dispatch_count"] == 6
+    assert (
+        payload["rollups"][
+            "live_log_controlled_writeback_consumer_dispatch_ledger_count"
+        ]
+        == 6
+    )
+    assert payload["rollups"]["live_log_controlled_writeback_consumer_binding_count"] == 6
+    assert (
+        payload["rollups"]["live_log_controlled_writeback_consumer_ready_binding_count"]
+        == 6
+    )
+    assert payload["rollups"]["live_log_controlled_writeback_consumer_ready_count"] == 1
+    assert payload["rollups"]["live_log_controlled_writeback_consumer_blocker_count"] == 0
+
+
 def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe():
     from src.services.ai_agent_autonomous_runtime_control import (
         _RUNTIME_OPERATION_COUNTS_SQL,