fix(agent): consume log writeback receipts

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -3535,6 +3535,9 @@ def _attach_runtime_receipt_readback(
         "live_log_controlled_writeback_consumer_dispatch_ledger_count": (
             log_consumer_dispatch_ledger_count
         ),
+        "live_log_controlled_writeback_consumer_apply_receipt_count": _int_value(
+            log_consumer_rollups.get("consumer_apply_receipt_row_count")
+        ),
         "live_log_controlled_writeback_consumer_binding_count": _int_value(
             log_consumer_rollups.get("consumer_binding_count")
         ),
@@ -3559,6 +3562,14 @@ def _attach_runtime_receipt_readback(
         "live_log_controlled_writeback_consumer_verifier_ref_count": _int_value(
             log_consumer_rollups.get("post_apply_verifier_ref_count")
         ),
+        "live_log_controlled_writeback_target_context_receipt_write_count": (
+            _int_value(log_consumer_rollups.get("target_context_receipt_write_count"))
+        ),
+        "live_log_controlled_writeback_runtime_target_write_count": (
+            1
+            if log_consumer_rollups.get("runtime_target_write_performed") is True
+            else 0
+        ),
         "live_log_controlled_writeback_km_consumer_binding_count": _int_value(
             log_consumer_rollups.get("km_consumer_binding_count")
         ),

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py

@@ -0,0 +1,423 @@
+"""AI Agent LOG controlled writeback consumer apply.
+
+Consumes ready metadata-only dispatch receipts into a consumer-context receipt
+ledger. This is an idempotent controlled apply step: it writes receipt metadata
+to automation_operation_log so KM / RAG / PlayBook / MCP / verifier / AI Agent
+loops can read a stable handoff, without persisting raw logs, calling tools,
+triggering workflows, sending Telegram, or reading secrets.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from sqlalchemy import text
+
+from src.db.base import get_db_context
+from src.services.ai_agent_log_controlled_writeback_consumer_readback import (
+    CONSUMER_EXECUTOR_ROUTE,
+    CONSUMER_OPERATION_TYPE,
+    DEFAULT_PROJECT_ID,
+    load_latest_ai_agent_log_controlled_writeback_consumer_readback,
+)
+from src.services.ai_agent_log_controlled_writeback_dispatch import (
+    FALLBACK_LEDGER_OPERATION_TYPE,
+)
+
+SCHEMA_VERSION = "ai_agent_log_controlled_writeback_consumer_apply_receipt_v1"
+
+
+async def consume_latest_ai_agent_log_controlled_writeback(
+    *,
+    project_id: str = DEFAULT_PROJECT_ID,
+) -> dict[str, Any]:
+    """Persist idempotent consumer-context receipts for ready LOG bindings."""
+
+    readback = await load_latest_ai_agent_log_controlled_writeback_consumer_readback(
+        project_id=project_id,
+    )
+    receipt = build_ai_agent_log_controlled_writeback_consumer_apply_receipt(readback)
+    if receipt["active_blockers"]:
+        return receipt
+
+    rows = []
+    async with get_db_context(project_id) as db:
+        ledger_operation_type = await _resolve_ledger_operation_type(db)
+        for item in receipt["consumer_apply_receipts"]:
+            rows.append(
+                await _insert_consumer_receipt_row(
+                    db,
+                    project_id=project_id,
+                    item=item,
+                    ledger_operation_type=ledger_operation_type,
+                )
+            )
+
+    inserted_count = sum(1 for row in rows if row["created"] is True)
+    existing_count = sum(1 for row in rows if row["created"] is False)
+    receipt["apply_result"] = {
+        "operation_type": CONSUMER_OPERATION_TYPE,
+        "ledger_operation_type": ledger_operation_type,
+        "semantic_operation_type": CONSUMER_OPERATION_TYPE,
+        "inserted_count": inserted_count,
+        "existing_count": existing_count,
+        "consumer_apply_receipt_row_count": len(rows),
+        "rows": rows,
+    }
+    receipt["rollups"]["runtime_target_write_performed"] = True
+    receipt["rollups"]["consumer_apply_receipt_row_count"] = len(rows)
+    receipt["rollups"]["inserted_consumer_apply_receipt_row_count"] = inserted_count
+    receipt["rollups"]["existing_consumer_apply_receipt_row_count"] = existing_count
+    receipt["operation_boundaries"]["consumer_context_receipt_write_performed"] = True
+    receipt["operation_boundaries"]["runtime_target_write_performed"] = True
+    return receipt
+
+
+def build_ai_agent_log_controlled_writeback_consumer_apply_receipt(
+    readback: dict[str, Any],
+) -> dict[str, Any]:
+    """Build a deterministic controlled-apply envelope from consumer readback."""
+
+    bindings = [
+        binding
+        for binding in readback.get("consumer_bindings", [])
+        if isinstance(binding, dict)
+    ]
+    apply_receipts = [_consumer_apply_receipt(binding) for binding in bindings]
+    active_blockers = _active_blockers(
+        readback=readback,
+        bindings=bindings,
+        apply_receipts=apply_receipts,
+    )
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
+        "scope": "ai_agent_log_controlled_writeback_consumer_apply",
+        "status": (
+            "controlled_writeback_consumer_apply_ready"
+            if not active_blockers
+            else "blocked_waiting_controlled_writeback_consumer_apply_inputs"
+        ),
+        "readback": {
+            "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-CONSUMER-APPLY",
+            "workplan_title": (
+                "LOG metadata dispatch receipts consumed into KM / RAG / PlayBook / "
+                "MCP / verifier / AI Agent context receipt ledger"
+            ),
+            "source_schema_version": readback.get("schema_version"),
+            "source_status": readback.get("status"),
+            "operation_type": CONSUMER_OPERATION_TYPE,
+            "executor_route": CONSUMER_EXECUTOR_ROUTE,
+            "safe_next_step": (
+                "post_apply_verify_consumer_context_receipts_then_read_runtime_control"
+            ),
+        },
+        "consumer_apply_receipts": apply_receipts,
+        "rollups": {
+            "source_consumer_binding_count": len(bindings),
+            "ready_source_consumer_binding_count": sum(
+                1
+                for binding in bindings
+                if binding.get("status") == "ready_for_consumer_context"
+            ),
+            "target_count": 6,
+            "consumer_apply_receipt_count": len(apply_receipts),
+            "ready_consumer_apply_receipt_count": sum(
+                1 for item in apply_receipts if item["status"] == "ready_for_context_receipt_write"
+            ),
+            "metadata_only_receipt_count": sum(
+                1 for item in apply_receipts if item["raw_payload_included"] is False
+            ),
+            "post_apply_verifier_ref_count": sum(
+                len(item["post_apply_verifier_refs"]) for item in apply_receipts
+            ),
+            "runtime_target_write_performed": False,
+            "consumer_apply_receipt_row_count": 0,
+            "inserted_consumer_apply_receipt_row_count": 0,
+            "existing_consumer_apply_receipt_row_count": 0,
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "metadata_ledger_write_only": True,
+            "consumer_context_receipt_write_performed": False,
+            "runtime_target_write_performed": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "workflow_trigger_performed": False,
+            "raw_log_payload_persisted": False,
+            "secret_value_collection_allowed": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def _consumer_apply_receipt(binding: dict[str, Any]) -> dict[str, Any]:
+    target = str(binding.get("target") or "")
+    dispatch_receipt_id = str(binding.get("dispatch_receipt_id") or "")
+    verifier_refs = _strings(
+        (binding.get("post_apply_verifier") or {}).get("verifier_refs")
+        or binding.get("post_apply_verifier_refs")
+    )
+    return {
+        "consumer_receipt_id": f"{CONSUMER_OPERATION_TYPE}::{target}",
+        "source_dispatch_receipt_id": dispatch_receipt_id,
+        "target": target,
+        "target_surface": str(binding.get("target_surface") or ""),
+        "consumer_surface": str(binding.get("consumer_surface") or ""),
+        "risk_tier": str(binding.get("risk_tier") or ""),
+        "executor_route": CONSUMER_EXECUTOR_ROUTE,
+        "status": (
+            "ready_for_context_receipt_write"
+            if _binding_ready_for_apply(binding, verifier_refs)
+            else "blocked_waiting_context_receipt_controls"
+        ),
+        "source_ledger_op_id": str(binding.get("ledger_op_id") or ""),
+        "post_apply_verifier_refs": verifier_refs,
+        "raw_payload_included": bool(binding.get("raw_payload_included") is not False),
+        "target_selector": {
+            "dispatch_receipt_id": dispatch_receipt_id,
+            "target": target,
+            "target_surface": str(binding.get("target_surface") or ""),
+            "consumer_surface": str(binding.get("consumer_surface") or ""),
+        },
+        "source_of_truth_diff": {
+            "current_state": "consumer_binding_ready",
+            "desired_state": "consumer_context_receipt_recorded",
+            "delta_kind": f"{target}_consumer_context_receipt",
+            "raw_payload_included": False,
+        },
+        "check_mode": {
+            "enabled": True,
+            "checks": [
+                "source_dispatch_receipt_present",
+                "consumer_surface_present",
+                "metadata_only_raw_payload_absent",
+                "post_apply_verifier_refs_present",
+                "source_binding_ready",
+            ],
+        },
+        "rollback": {
+            "required": True,
+            "rollback_ref": (
+                "rollback://ai-agent-log-controlled-writeback-consumer-apply/"
+                f"{CONSUMER_OPERATION_TYPE}::{target}"
+            ),
+            "strategy": "mark_consumer_context_receipt_superseded_and_ignore",
+        },
+    }
+
+
+def _binding_ready_for_apply(
+    binding: dict[str, Any],
+    verifier_refs: list[str],
+) -> bool:
+    if binding.get("status") != "ready_for_consumer_context":
+        return False
+    if str(binding.get("dispatch_receipt_id") or "") == "":
+        return False
+    if str(binding.get("target") or "") == "":
+        return False
+    if str(binding.get("target_surface") or "") == "":
+        return False
+    if str(binding.get("consumer_surface") or "") == "":
+        return False
+    if binding.get("raw_payload_included") is not False:
+        return False
+    return bool(verifier_refs)
+
+
+def _active_blockers(
+    *,
+    readback: dict[str, Any],
+    bindings: list[dict[str, Any]],
+    apply_receipts: list[dict[str, Any]],
+) -> list[str]:
+    blockers: list[str] = []
+    if readback.get("status") != "controlled_writeback_consumer_readback_ready":
+        blockers.append("controlled_writeback_consumer_readback_not_ready")
+    if (readback.get("controlled_consume") or {}).get("controlled_consume_allowed") is not True:
+        blockers.append("controlled_consume_not_allowed")
+    if readback.get("active_blockers"):
+        blockers.extend(str(item) for item in readback.get("active_blockers") or [])
+    if not bindings:
+        blockers.append("consumer_bindings_missing")
+    for item in apply_receipts:
+        target = item["target"] or "unknown"
+        if item["status"] != "ready_for_context_receipt_write":
+            blockers.append(f"{target}_consumer_apply_receipt_not_ready")
+        if item["raw_payload_included"] is not False:
+            blockers.append(f"{target}_raw_payload_included")
+    return _unique(blockers)
+
+
+async def _insert_consumer_receipt_row(
+    db: Any,
+    *,
+    project_id: str,
+    item: dict[str, Any],
+    ledger_operation_type: str,
+) -> dict[str, Any]:
+    result = await db.execute(
+        text("""
+            INSERT INTO automation_operation_log (
+                operation_type, actor, status,
+                input, output, dry_run_result, tags
+            )
+            SELECT
+                :ledger_operation_type,
+                :actor,
+                'success',
+                CAST(:input AS jsonb),
+                CAST(:output AS jsonb),
+                CAST(:dry_run_result AS jsonb),
+                :tags
+            WHERE NOT EXISTS (
+                SELECT 1
+                FROM automation_operation_log existing
+                WHERE coalesce(
+                    existing.input ->> 'semantic_operation_type',
+                    existing.operation_type
+                ) = :operation_type
+                  AND existing.input ->> 'consumer_receipt_id' = :consumer_receipt_id
+            )
+            RETURNING op_id::text
+        """),
+        {
+            "operation_type": CONSUMER_OPERATION_TYPE,
+            "ledger_operation_type": ledger_operation_type,
+            "actor": CONSUMER_EXECUTOR_ROUTE,
+            "consumer_receipt_id": item["consumer_receipt_id"],
+            "input": json.dumps(
+                {
+                    "schema_version": SCHEMA_VERSION,
+                    "project_id": project_id,
+                    "semantic_operation_type": CONSUMER_OPERATION_TYPE,
+                    "ledger_operation_type": ledger_operation_type,
+                    "consumer_receipt_id": item["consumer_receipt_id"],
+                    "source_dispatch_receipt_id": item["source_dispatch_receipt_id"],
+                    "target": item["target"],
+                    "target_surface": item["target_surface"],
+                    "consumer_surface": item["consumer_surface"],
+                    "risk_tier": item["risk_tier"],
+                    "source_ledger_op_id": item["source_ledger_op_id"],
+                    "runtime_target_write_performed": True,
+                    "raw_payload_included": False,
+                },
+                ensure_ascii=False,
+            ),
+            "output": json.dumps(
+                {
+                    "executor_route": CONSUMER_EXECUTOR_ROUTE,
+                    "semantic_operation_type": CONSUMER_OPERATION_TYPE,
+                    "ledger_operation_type": ledger_operation_type,
+                    "consumer_context_receipt_recorded": True,
+                    "target_context_receipt_write_performed": True,
+                    "runtime_target_write_performed": True,
+                    "post_apply_verifier_refs": item["post_apply_verifier_refs"],
+                    "next_action": "readback_consumer_context_receipts",
+                    "km_write_performed": False,
+                    "rag_index_write_performed": False,
+                    "playbook_trust_write_performed": False,
+                    "mcp_tool_call_performed": False,
+                    "telegram_send_performed": False,
+                },
+                ensure_ascii=False,
+            ),
+            "dry_run_result": json.dumps(
+                {
+                    "target_selector": item["target_selector"],
+                    "source_of_truth_diff": item["source_of_truth_diff"],
+                    "check_mode": item["check_mode"],
+                    "rollback": item["rollback"],
+                    "post_apply_verifier_ref_count": len(item["post_apply_verifier_refs"]),
+                    "raw_payload_included": False,
+                },
+                ensure_ascii=False,
+            ),
+            "tags": [
+                "ai_agent",
+                "log_feedback",
+                "controlled_writeback_consumer",
+                item["target"],
+            ],
+        },
+    )
+    op_id = result.scalar()
+    if op_id:
+        return {
+            "consumer_receipt_id": item["consumer_receipt_id"],
+            "target": item["target"],
+            "created": True,
+            "op_id": str(op_id),
+        }
+
+    existing = await db.execute(
+        text("""
+            SELECT op_id::text
+            FROM automation_operation_log
+            WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                = :operation_type
+              AND input ->> 'consumer_receipt_id' = :consumer_receipt_id
+            ORDER BY created_at DESC
+            LIMIT 1
+        """),
+        {
+            "operation_type": CONSUMER_OPERATION_TYPE,
+            "consumer_receipt_id": item["consumer_receipt_id"],
+        },
+    )
+    return {
+        "consumer_receipt_id": item["consumer_receipt_id"],
+        "target": item["target"],
+        "created": False,
+        "op_id": str(existing.scalar() or ""),
+    }
+
+
+async def _resolve_ledger_operation_type(db: Any) -> str:
+    result = await db.execute(
+        text("""
+            SELECT CASE
+                WHEN EXISTS (
+                    SELECT 1
+                    FROM pg_constraint
+                    WHERE conname = 'automation_operation_log_type_valid'
+                      AND pg_get_constraintdef(oid) LIKE '%' || :operation_type || '%'
+                )
+                THEN :operation_type
+                ELSE :fallback_operation_type
+            END
+        """),
+        {
+            "operation_type": CONSUMER_OPERATION_TYPE,
+            "fallback_operation_type": FALLBACK_LEDGER_OPERATION_TYPE,
+        },
+    )
+    resolved = str(result.scalar() or "")
+    if resolved == CONSUMER_OPERATION_TYPE:
+        return CONSUMER_OPERATION_TYPE
+    return FALLBACK_LEDGER_OPERATION_TYPE
+
+
+def _strings(value: Any) -> list[str]:
+    if isinstance(value, list):
+        return [str(item) for item in value]
+    if isinstance(value, tuple):
+        return [str(item) for item in value]
+    return []
+
+
+def _unique(values: list[str]) -> list[str]:
+    seen = set()
+    result = []
+    for value in values:
+        if value in seen:
+            continue
+        seen.add(value)
+        result.append(value)
+    return result

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py

@@ -23,6 +23,8 @@ from src.services.ai_agent_log_controlled_writeback_dispatch import (
 
 SCHEMA_VERSION = "ai_agent_log_controlled_writeback_consumer_readback_v1"
 DEFAULT_PROJECT_ID = "awoooi"
+CONSUMER_OPERATION_TYPE = "log_controlled_writeback_consumed"
+CONSUMER_EXECUTOR_ROUTE = "ai_agent_metadata_writeback_consumer"
 _TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
 _CONSUMER_SURFACES = {
     "km": "knowledge_memory_context",
@@ -73,9 +75,48 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
             """),
             {"operation_type": OPERATION_TYPE},
         )
+        consumer_result = await db.execute(
+            text("""
+                SELECT
+                    op_id::text AS op_id,
+                    operation_type,
+                    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,
+                    output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs
+                FROM automation_operation_log
+                WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
+                    = :operation_type
+                ORDER BY created_at DESC, op_id DESC
+                LIMIT 50
+            """),
+            {"operation_type": CONSUMER_OPERATION_TYPE},
+        )
     rows = _result_rows(result)
-    bindings = [_consumer_binding(row) for row in rows]
+    consumer_rows = _result_rows(consumer_result)
+    consumer_receipts = _consumer_receipts_by_dispatch_id(consumer_rows)
+    bindings = [_consumer_binding(row, consumer_receipts) for row in rows]
     active_blockers = _active_blockers(bindings)
+    target_rollups = _target_rollups(bindings)
+    runtime_target_write_performed = (
+        not active_blockers
+        and bool(bindings)
+        and all(item["target_write_performed"] is True for item in target_rollups)
+    )
 
     return {
         "schema_version": SCHEMA_VERSION,
@@ -110,13 +151,17 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
             "check_mode_required": True,
             "rollback_required": True,
             "post_apply_verifier_required": True,
-            "runtime_target_write_performed": False,
+            "runtime_target_write_performed": runtime_target_write_performed,
+            "consumer_apply_route": (
+                "/api/v1/agents/agent-log-controlled-writeback-consumer-apply"
+            ),
         },
         "consumer_bindings": bindings,
-        "target_rollups": _target_rollups(bindings),
+        "target_rollups": target_rollups,
         "rollups": {
             "target_count": len(_TARGETS),
             "dispatch_ledger_row_count": len(rows),
+            "consumer_apply_receipt_row_count": len(consumer_rows),
             "consumer_binding_count": len(bindings),
             "ready_consumer_binding_count": sum(
                 1 for binding in bindings if binding["status"] == "ready_for_consumer_context"
@@ -137,12 +182,29 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
             "mcp_consumer_binding_count": _target_count(bindings, "mcp"),
             "verifier_consumer_binding_count": _target_count(bindings, "verifier"),
             "ai_agent_consumer_binding_count": _target_count(bindings, "ai_agent"),
-            "runtime_target_write_performed": False,
+            "target_context_receipt_write_count": sum(
+                1 for binding in bindings if binding["target_write_performed"] is True
+            ),
+            "km_context_receipt_write_count": _target_write_count(bindings, "km"),
+            "rag_context_receipt_write_count": _target_write_count(bindings, "rag"),
+            "playbook_context_receipt_write_count": _target_write_count(
+                bindings, "playbook"
+            ),
+            "mcp_context_receipt_write_count": _target_write_count(bindings, "mcp"),
+            "verifier_context_receipt_write_count": _target_write_count(
+                bindings, "verifier"
+            ),
+            "ai_agent_context_receipt_write_count": _target_write_count(
+                bindings, "ai_agent"
+            ),
+            "runtime_target_write_performed": runtime_target_write_performed,
         },
         "active_blockers": active_blockers,
         "operation_boundaries": {
             "consumer_readback_only": True,
             "metadata_ledger_read_performed": True,
+            "consumer_apply_receipt_read_performed": True,
+            "runtime_target_write_performed": runtime_target_write_performed,
             "km_write_performed": False,
             "rag_index_write_performed": False,
             "playbook_trust_write_performed": False,
@@ -157,14 +219,27 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
     }
 
 
-def _consumer_binding(row: Mapping[str, Any]) -> dict[str, Any]:
+def _consumer_binding(
+    row: Mapping[str, Any],
+    consumer_receipts: Mapping[str, Mapping[str, Any]],
+) -> dict[str, Any]:
     target = str(row.get("target") or "")
+    dispatch_receipt_id = str(row.get("dispatch_receipt_id") or "")
+    consumer_receipt = consumer_receipts.get(dispatch_receipt_id)
     verifier_refs = _strings(row.get("post_apply_verifier_refs"))
     raw_payload_included = str(row.get("raw_payload_included") or "").lower() == "true"
     return {
-        "consumer_binding_id": f"consumer::{row.get('dispatch_receipt_id') or row.get('op_id')}",
-        "dispatch_receipt_id": str(row.get("dispatch_receipt_id") or ""),
+        "consumer_binding_id": f"consumer::{dispatch_receipt_id or row.get('op_id')}",
+        "dispatch_receipt_id": dispatch_receipt_id,
         "ledger_op_id": str(row.get("op_id") or ""),
+        "consumer_receipt_id": (
+            str(consumer_receipt.get("consumer_receipt_id") or "")
+            if consumer_receipt
+            else ""
+        ),
+        "consumer_receipt_op_id": (
+            str(consumer_receipt.get("op_id") or "") if consumer_receipt else ""
+        ),
         "target": target,
         "target_surface": str(row.get("target_surface") or ""),
         "consumer_surface": _CONSUMER_SURFACES.get(target, "unknown_consumer_surface"),
@@ -216,7 +291,33 @@ def _consumer_binding(row: Mapping[str, Any]) -> dict[str, Any]:
             "required": True,
             "verifier_refs": verifier_refs,
         },
-        "target_write_performed": False,
+        "target_write_performed": consumer_receipt is not None,
+        "target_write_receipt": (
+            {
+                "consumer_receipt_id": str(consumer_receipt.get("consumer_receipt_id") or ""),
+                "ledger_op_id": str(consumer_receipt.get("op_id") or ""),
+                "status": str(consumer_receipt.get("status") or ""),
+                "actor": str(consumer_receipt.get("actor") or ""),
+                "semantic_operation_type": str(
+                    consumer_receipt.get("semantic_operation_type") or ""
+                ),
+                "runtime_target_write_performed": (
+                    str(
+                        consumer_receipt.get("runtime_target_write_performed") or ""
+                    ).lower()
+                    == "true"
+                ),
+                "target_context_receipt_write_performed": (
+                    str(
+                        consumer_receipt.get("target_context_receipt_write_performed")
+                        or ""
+                    ).lower()
+                    == "true"
+                ),
+            }
+            if consumer_receipt
+            else None
+        ),
     }
 
 
@@ -270,6 +371,11 @@ def _target_rollups(bindings: list[dict[str, Any]]) -> list[dict[str, Any]]:
                 for binding in bindings
                 if binding["target"] == target
             ),
+            "target_write_performed": any(
+                binding["target_write_performed"] is True
+                for binding in bindings
+                if binding["target"] == target
+            ),
         }
         for target in _TARGETS
     ]
@@ -279,6 +385,34 @@ def _target_count(bindings: list[dict[str, Any]], target: str) -> int:
     return sum(1 for binding in bindings if binding["target"] == target)
 
 
+def _target_write_count(bindings: list[dict[str, Any]], target: str) -> int:
+    return sum(
+        1
+        for binding in bindings
+        if binding["target"] == target and binding["target_write_performed"] is True
+    )
+
+
+def _consumer_receipts_by_dispatch_id(
+    rows: list[dict[str, Any]],
+) -> dict[str, dict[str, Any]]:
+    receipts: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        if str(row.get("semantic_operation_type") or "") != CONSUMER_OPERATION_TYPE:
+            continue
+        if str(row.get("actor") or "") != CONSUMER_EXECUTOR_ROUTE:
+            continue
+        if str(row.get("status") or "") != "success":
+            continue
+        if str(row.get("raw_payload_included") or "").lower() == "true":
+            continue
+        dispatch_receipt_id = str(row.get("source_dispatch_receipt_id") or "")
+        if not dispatch_receipt_id or dispatch_receipt_id in receipts:
+            continue
+        receipts[dispatch_receipt_id] = row
+    return receipts
+
+
 def _result_rows(result: Any) -> list[dict[str, Any]]:
     mappings = getattr(result, "mappings", None)
     if callable(mappings):