fix(agents): parse direct log consumer receipts

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py

@@ -11,6 +11,7 @@ prove this route has rolled out.
 from __future__ import annotations
 
 import asyncio
+import json
 from collections.abc import Mapping
 from typing import Any
 
@@ -250,7 +251,10 @@ async def _load_consumer_rows_direct(
                 conn.fetch(_CONSUMER_APPLY_DIRECT_SQL, CONSUMER_OPERATION_TYPE),
                 timeout=_CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS,
             )
-            return _result_rows(dispatch_rows), _result_rows(consumer_rows)
+            return (
+                _direct_result_rows(dispatch_rows, row_kind="dispatch"),
+                _direct_result_rows(consumer_rows, row_kind="consumer"),
+            )
         finally:
             try:
                 await asyncio.wait_for(
@@ -279,33 +283,8 @@ _CONSUMER_DISPATCH_DIRECT_SQL = """
         actor,
         status,
         created_at,
-        input ->> 'semantic_operation_type' AS semantic_operation_type,
-        input ->> 'ledger_operation_type' AS ledger_operation_type,
-        input ->> 'dispatch_receipt_id' AS dispatch_receipt_id,
-        input ->> 'batch_id' AS batch_id,
-        input ->> 'target' AS target,
-        input ->> 'target_surface' AS target_surface,
-        input ->> 'risk_tier' AS risk_tier,
-        input ->> 'project_id' AS project_id,
-        input ->> 'raw_payload_included' AS raw_payload_included,
-        output ->> 'next_action' AS next_action,
-        output ->> 'km_write_performed' AS km_write_performed,
-        output ->> 'rag_index_write_performed' AS rag_index_write_performed,
-        output ->> 'playbook_trust_write_performed' AS playbook_trust_write_performed,
-        output ->> 'mcp_tool_call_performed' AS mcp_tool_call_performed,
-        output ->> 'telegram_send_performed' AS telegram_send_performed,
-        COALESCE(
-            ARRAY(
-                SELECT jsonb_array_elements_text(
-                    CASE
-                        WHEN jsonb_typeof(output -> 'post_apply_verifier_refs') = 'array'
-                        THEN output -> 'post_apply_verifier_refs'
-                        ELSE '[]'::jsonb
-                    END
-                )
-            ),
-            ARRAY[]::text[]
-        ) AS post_apply_verifier_refs
+        input::text AS input_json,
+        output::text AS output_json
     FROM automation_operation_log
     WHERE coalesce(input ->> 'semantic_operation_type', operation_type) = $1
     ORDER BY created_at DESC, op_id DESC
@@ -319,32 +298,8 @@ _CONSUMER_APPLY_DIRECT_SQL = """
         actor,
         status,
         created_at,
-        input ->> 'semantic_operation_type' AS semantic_operation_type,
-        input ->> 'ledger_operation_type' AS ledger_operation_type,
-        input ->> 'consumer_receipt_id' AS consumer_receipt_id,
-        input ->> 'source_dispatch_receipt_id' AS source_dispatch_receipt_id,
-        input ->> 'target' AS target,
-        input ->> 'target_surface' AS target_surface,
-        input ->> 'consumer_surface' AS consumer_surface,
-        input ->> 'runtime_target_write_performed' AS runtime_target_write_performed,
-        input ->> 'raw_payload_included' AS raw_payload_included,
-        output ->> 'consumer_context_receipt_recorded'
-            AS consumer_context_receipt_recorded,
-        output ->> 'target_context_receipt_write_performed'
-            AS target_context_receipt_write_performed,
-        output ->> 'next_action' AS next_action,
-        COALESCE(
-            ARRAY(
-                SELECT jsonb_array_elements_text(
-                    CASE
-                        WHEN jsonb_typeof(output -> 'post_apply_verifier_refs') = 'array'
-                        THEN output -> 'post_apply_verifier_refs'
-                        ELSE '[]'::jsonb
-                    END
-                )
-            ),
-            ARRAY[]::text[]
-        ) AS post_apply_verifier_refs
+        input::text AS input_json,
+        output::text AS output_json
     FROM automation_operation_log
     WHERE coalesce(input ->> 'semantic_operation_type', operation_type) = $1
     ORDER BY created_at DESC, op_id DESC
@@ -352,6 +307,85 @@ _CONSUMER_APPLY_DIRECT_SQL = """
 """
 
 
+def _direct_result_rows(rows: Any, *, row_kind: str) -> list[dict[str, Any]]:
+    return [
+        _direct_row_from_json_payloads(_row_mapping(row), row_kind=row_kind)
+        for row in rows
+    ]
+
+
+def _direct_row_from_json_payloads(
+    row: Mapping[str, Any],
+    *,
+    row_kind: str,
+) -> dict[str, Any]:
+    input_payload = _json_dict(row.get("input_json"))
+    output_payload = _json_dict(row.get("output_json"))
+    base = {
+        "op_id": str(row.get("op_id") or ""),
+        "operation_type": str(row.get("operation_type") or ""),
+        "actor": str(row.get("actor") or ""),
+        "status": str(row.get("status") or ""),
+        "created_at": row.get("created_at"),
+        "semantic_operation_type": str(
+            input_payload.get("semantic_operation_type")
+            or row.get("operation_type")
+            or ""
+        ),
+        "ledger_operation_type": str(input_payload.get("ledger_operation_type") or ""),
+        "target": str(input_payload.get("target") or ""),
+        "target_surface": str(input_payload.get("target_surface") or ""),
+        "raw_payload_included": _json_scalar_string(
+            input_payload.get("raw_payload_included")
+        ),
+        "next_action": str(output_payload.get("next_action") or ""),
+        "post_apply_verifier_refs": _list(
+            output_payload.get("post_apply_verifier_refs")
+        ),
+    }
+    if row_kind == "consumer":
+        base.update({
+            "consumer_receipt_id": str(input_payload.get("consumer_receipt_id") or ""),
+            "source_dispatch_receipt_id": str(
+                input_payload.get("source_dispatch_receipt_id") or ""
+            ),
+            "consumer_surface": str(input_payload.get("consumer_surface") or ""),
+            "runtime_target_write_performed": _json_scalar_string(
+                input_payload.get("runtime_target_write_performed")
+            ),
+            "consumer_context_receipt_recorded": _json_scalar_string(
+                output_payload.get("consumer_context_receipt_recorded")
+            ),
+            "target_context_receipt_write_performed": _json_scalar_string(
+                output_payload.get("target_context_receipt_write_performed")
+            ),
+        })
+        return base
+
+    base.update({
+        "dispatch_receipt_id": str(input_payload.get("dispatch_receipt_id") or ""),
+        "batch_id": str(input_payload.get("batch_id") or ""),
+        "risk_tier": str(input_payload.get("risk_tier") or ""),
+        "project_id": str(input_payload.get("project_id") or ""),
+        "km_write_performed": _json_scalar_string(
+            output_payload.get("km_write_performed")
+        ),
+        "rag_index_write_performed": _json_scalar_string(
+            output_payload.get("rag_index_write_performed")
+        ),
+        "playbook_trust_write_performed": _json_scalar_string(
+            output_payload.get("playbook_trust_write_performed")
+        ),
+        "mcp_tool_call_performed": _json_scalar_string(
+            output_payload.get("mcp_tool_call_performed")
+        ),
+        "telegram_send_performed": _json_scalar_string(
+            output_payload.get("telegram_send_performed")
+        ),
+    })
+    return base
+
+
 def _build_consumer_readback(
     *,
     rows: list[dict[str, Any]],
@@ -1037,6 +1071,28 @@ def _dict(value: Any) -> dict[str, Any]:
     return {}
 
 
+def _json_dict(value: Any) -> dict[str, Any]:
+    if isinstance(value, Mapping):
+        return dict(value)
+    if isinstance(value, (bytes, bytearray)):
+        value = value.decode("utf-8", errors="replace")
+    if isinstance(value, str) and value:
+        try:
+            parsed = json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+        return dict(parsed) if isinstance(parsed, Mapping) else {}
+    return {}
+
+
+def _json_scalar_string(value: Any) -> str:
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if value is None:
+        return ""
+    return str(value)
+
+
 def _list(value: Any) -> list[Any]:
     if isinstance(value, list):
         return value

apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json
+
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
@@ -302,6 +304,88 @@ async def test_log_controlled_writeback_consumer_loader_uses_direct_fallback(
     assert payload["operation_boundaries"]["metadata_ledger_read_performed"] is True
 
 
+def test_log_controlled_writeback_consumer_direct_rows_parse_json_text():
+    dispatch_row = _ledger_rows()[0]
+    consumer_row = _consumer_receipt_rows()[0]
+
+    direct_dispatch_rows = consumer_module._direct_result_rows(
+        [
+            {
+                "op_id": dispatch_row["op_id"],
+                "operation_type": dispatch_row["operation_type"],
+                "actor": dispatch_row["actor"],
+                "status": dispatch_row["status"],
+                "created_at": dispatch_row["created_at"],
+                "input_json": json.dumps({
+                    "semantic_operation_type": dispatch_row[
+                        "semantic_operation_type"
+                    ],
+                    "ledger_operation_type": dispatch_row["ledger_operation_type"],
+                    "dispatch_receipt_id": dispatch_row["dispatch_receipt_id"],
+                    "batch_id": dispatch_row["batch_id"],
+                    "target": dispatch_row["target"],
+                    "target_surface": dispatch_row["target_surface"],
+                    "risk_tier": dispatch_row["risk_tier"],
+                    "project_id": dispatch_row["project_id"],
+                    "raw_payload_included": False,
+                }),
+                "output_json": json.dumps({
+                    "next_action": dispatch_row["next_action"],
+                    "post_apply_verifier_refs": dispatch_row[
+                        "post_apply_verifier_refs"
+                    ],
+                }),
+            }
+        ],
+        row_kind="dispatch",
+    )
+    direct_consumer_rows = consumer_module._direct_result_rows(
+        [
+            {
+                "op_id": consumer_row["op_id"],
+                "operation_type": consumer_row["operation_type"],
+                "actor": consumer_row["actor"],
+                "status": consumer_row["status"],
+                "created_at": consumer_row["created_at"],
+                "input_json": json.dumps({
+                    "semantic_operation_type": consumer_row[
+                        "semantic_operation_type"
+                    ],
+                    "ledger_operation_type": consumer_row["ledger_operation_type"],
+                    "consumer_receipt_id": consumer_row["consumer_receipt_id"],
+                    "source_dispatch_receipt_id": consumer_row[
+                        "source_dispatch_receipt_id"
+                    ],
+                    "target": consumer_row["target"],
+                    "target_surface": consumer_row["target_surface"],
+                    "consumer_surface": consumer_row["consumer_surface"],
+                    "runtime_target_write_performed": True,
+                    "raw_payload_included": False,
+                }),
+                "output_json": json.dumps({
+                    "target_context_receipt_write_performed": True,
+                    "post_apply_verifier_refs": consumer_row[
+                        "post_apply_verifier_refs"
+                    ],
+                }),
+            }
+        ],
+        row_kind="consumer",
+    )
+
+    assert direct_dispatch_rows[0]["dispatch_receipt_id"] == (
+        dispatch_row["dispatch_receipt_id"]
+    )
+    assert direct_dispatch_rows[0]["raw_payload_included"] == "false"
+    assert direct_dispatch_rows[0]["post_apply_verifier_refs"] == (
+        dispatch_row["post_apply_verifier_refs"]
+    )
+    assert direct_consumer_rows[0]["source_dispatch_receipt_id"] == (
+        dispatch_row["dispatch_receipt_id"]
+    )
+    assert direct_consumer_rows[0]["runtime_target_write_performed"] == "true"
+
+
 def test_log_controlled_writeback_consumer_endpoint_returns_readback(monkeypatch):
     async def fake_loader():
         fake_db = _FakeDb(_ledger_rows())