From 7ee0a17ab7b6842dc8eb2104350cbab300d05243 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:24:49 +0800 Subject: [PATCH] fix(agents): parse direct log consumer receipts --- ..._controlled_writeback_consumer_readback.py | 164 ++++++++++++------ ...trolled_writeback_consumer_readback_api.py | 84 +++++++++ 2 files changed, 194 insertions(+), 54 deletions(-) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py index b90d68287..fa60a3509 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py +++ b/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 diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py index ffa48621b..7e2e98314 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py +++ b/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())