From 6580bbd641834bd49bed824ca33a1cc13c950e1f Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 1 Jul 2026 20:41:42 +0800 Subject: [PATCH] fix(agent): consume log writeback receipts --- apps/api/src/api/v1/agents.py | 31 ++ .../ai_agent_autonomous_runtime_control.py | 11 + ...log_controlled_writeback_consumer_apply.py | 423 ++++++++++++++++++ ..._controlled_writeback_consumer_readback.py | 150 ++++++- ...controlled_writeback_consumer_apply_api.py | 216 +++++++++ ...trolled_writeback_consumer_readback_api.py | 57 ++- 6 files changed, 878 insertions(+), 10 deletions(-) create mode 100644 apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py create mode 100644 apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 397b57bac..41de77a3f 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -101,6 +101,9 @@ from src.services.ai_agent_learning_writeback_approval_package import ( from src.services.ai_agent_live_read_model_gate import ( load_latest_ai_agent_live_read_model_gate, ) +from src.services.ai_agent_log_controlled_writeback_consumer_apply import ( + consume_latest_ai_agent_log_controlled_writeback, +) from src.services.ai_agent_log_controlled_writeback_consumer_readback import ( load_latest_ai_agent_log_controlled_writeback_consumer_readback, ) @@ -2425,6 +2428,34 @@ async def get_agent_log_controlled_writeback_consumer_readback() -> dict[str, An ) from exc +@router.post( + "/agent-log-controlled-writeback-consumer-apply", + response_model=dict[str, Any], + summary="執行 AI Agent LOG controlled writeback consumer context receipt apply", + description=( + "把 ready 的 LOG metadata-only dispatch receipts 寫成 consumer context " + "receipt ledger，供 KM / RAG / PlayBook / MCP / verifier / AI Agent loop " + "讀取。此端點只寫 metadata receipt，不保存 raw log payload、不讀 secret、" + "不呼叫 MCP tool、不發 Telegram、不觸發 workflow、不呼叫 GitHub；" + "critical break-glass 仍維持硬阻擋。" + ), +) +async def post_agent_log_controlled_writeback_consumer_apply() -> dict[str, Any]: + """Persist metadata-only LOG controlled writeback consumer receipts.""" + try: + payload = await consume_latest_ai_agent_log_controlled_writeback() + return redact_public_lan_topology(payload) + except (json.JSONDecodeError, ValueError) as exc: + logger.error( + "ai_agent_log_controlled_writeback_consumer_apply_invalid", + error=str(exc), + ) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="AI Agent LOG controlled writeback consumer apply 無效", + ) from exc + + @router.get( "/agent-telegram-receipt-approval-package", response_model=dict[str, Any], diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 2ea6d56f3..29f103f54 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/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") ), diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py new file mode 100644 index 000000000..402929ae5 --- /dev/null +++ b/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 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 26bb535ad..5400409f7 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 @@ -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): diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py new file mode 100644 index 000000000..36730e771 --- /dev/null +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.api.v1 import agents +from src.api.v1.agents import router +from src.services import ( + ai_agent_log_controlled_writeback_consumer_apply as apply_module, +) +from src.services.ai_agent_log_controlled_writeback_consumer_apply import ( + build_ai_agent_log_controlled_writeback_consumer_apply_receipt, + consume_latest_ai_agent_log_controlled_writeback, +) + + +class _FakeResult: + def __init__(self, value: str | None): + self.value = value + + def scalar(self) -> str | None: + return self.value + + +class _FakeDb: + def __init__(self): + self.params: list[dict] = [] + + async def execute(self, statement, params: dict): + self.params.append(params) + if "fallback_operation_type" in params: + return _FakeResult("km_linked") + if "input" in params: + return _FakeResult( + f"consumer-op-{params['consumer_receipt_id'].rsplit('::', 1)[-1]}" + ) + assert "pg_get_constraintdef" not in str(statement) + return _FakeResult("existing-consumer-op") + + +class _FakeContext: + def __init__(self, db: _FakeDb): + self.db = db + + async def __aenter__(self) -> _FakeDb: + return self.db + + async def __aexit__(self, _exc_type, _exc, _tb) -> bool: + return False + + +def _consumer_readback() -> dict: + targets = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent") + bindings = [ + { + "consumer_binding_id": f"consumer::log_controlled_writeback_dispatched::{target}", + "dispatch_receipt_id": f"log_controlled_writeback_dispatched::{target}", + "ledger_op_id": f"op-{target}", + "target": target, + "target_surface": f"{target}_metadata_feedback_binding", + "consumer_surface": f"{target}_consumer_context", + "risk_tier": "medium" if target in {"km", "rag", "playbook"} else "low", + "status": "ready_for_consumer_context", + "raw_payload_included": False, + "post_apply_verifier_refs": [f"post-write-verifier://{target}/sample"], + "post_apply_verifier": { + "required": True, + "verifier_refs": [f"post-write-verifier://{target}/sample"], + }, + "target_write_performed": False, + } + for target in targets + ] + return { + "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1", + "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK", + "scope": "ai_agent_log_controlled_writeback_consumer_readback", + "status": "controlled_writeback_consumer_readback_ready", + "readback": { + "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-CONSUMER-READBACK", + "source_operation_type": "log_controlled_writeback_dispatched", + "source_executor_route": "ai_agent_metadata_writeback_executor", + }, + "controlled_consume": { + "controlled_consume_allowed": True, + "runtime_target_write_performed": False, + }, + "consumer_bindings": bindings, + "rollups": { + "controlled_consumer_readback_ready": True, + "consumer_binding_count": 6, + }, + "active_blockers": [], + "operation_boundaries": { + "consumer_readback_only": True, + "metadata_ledger_read_performed": True, + "raw_log_payload_persisted": False, + "secret_value_collection_allowed": False, + "github_api_used": False, + }, + } + + +def test_log_controlled_writeback_consumer_apply_builder_creates_receipts(): + payload = build_ai_agent_log_controlled_writeback_consumer_apply_receipt( + _consumer_readback() + ) + + assert payload["schema_version"] == ( + "ai_agent_log_controlled_writeback_consumer_apply_receipt_v1" + ) + assert payload["status"] == "controlled_writeback_consumer_apply_ready" + assert payload["active_blockers"] == [] + assert payload["readback"]["operation_type"] == "log_controlled_writeback_consumed" + assert payload["rollups"]["consumer_apply_receipt_count"] == 6 + assert payload["rollups"]["ready_consumer_apply_receipt_count"] == 6 + assert payload["rollups"]["runtime_target_write_performed"] is False + assert {item["target"] for item in payload["consumer_apply_receipts"]} == { + "km", + "rag", + "playbook", + "mcp", + "verifier", + "ai_agent", + } + for item in payload["consumer_apply_receipts"]: + assert item["status"] == "ready_for_context_receipt_write" + assert item["raw_payload_included"] is False + assert item["check_mode"]["enabled"] is True + assert item["rollback"]["required"] is True + + boundaries = payload["operation_boundaries"] + assert boundaries["metadata_ledger_write_only"] is True + assert boundaries["consumer_context_receipt_write_performed"] is False + assert boundaries["km_write_performed"] is False + assert boundaries["rag_index_write_performed"] is False + assert boundaries["playbook_trust_write_performed"] is False + assert boundaries["mcp_tool_call_performed"] is False + assert boundaries["raw_log_payload_persisted"] is False + assert boundaries["secret_value_collection_allowed"] is False + assert boundaries["github_api_used"] is False + + +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_apply_writes_idempotent_rows(monkeypatch): + fake_db = _FakeDb() + monkeypatch.setattr( + apply_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + async def fake_readback(*, project_id: str): + assert project_id == "awoooi" + return _consumer_readback() + + monkeypatch.setattr( + apply_module, + "load_latest_ai_agent_log_controlled_writeback_consumer_readback", + fake_readback, + ) + + payload = await consume_latest_ai_agent_log_controlled_writeback(project_id="awoooi") + + assert payload["status"] == "controlled_writeback_consumer_apply_ready" + assert payload["rollups"]["runtime_target_write_performed"] is True + assert payload["rollups"]["consumer_apply_receipt_row_count"] == 6 + assert payload["rollups"]["inserted_consumer_apply_receipt_row_count"] == 6 + assert payload["operation_boundaries"]["consumer_context_receipt_write_performed"] is True + assert payload["operation_boundaries"]["runtime_target_write_performed"] is True + assert payload["apply_result"]["operation_type"] == "log_controlled_writeback_consumed" + assert payload["apply_result"]["ledger_operation_type"] == "km_linked" + + write_params = [params for params in fake_db.params if "input" in params] + assert len(write_params) == 6 + for params in write_params: + assert params["operation_type"] == "log_controlled_writeback_consumed" + assert params["ledger_operation_type"] == "km_linked" + assert params["actor"] == "ai_agent_metadata_writeback_consumer" + assert '"semantic_operation_type": "log_controlled_writeback_consumed"' in params["input"] + assert '"runtime_target_write_performed": true' in params["input"] + assert '"raw_payload_included": false' in params["input"] + assert '"target_context_receipt_write_performed": true' in params["output"] + assert '"km_write_performed": false' in params["output"] + + +def test_log_controlled_writeback_consumer_apply_endpoint_returns_receipt(monkeypatch): + async def fake_apply(): + payload = build_ai_agent_log_controlled_writeback_consumer_apply_receipt( + _consumer_readback() + ) + payload["rollups"]["runtime_target_write_performed"] = True + payload["operation_boundaries"]["runtime_target_write_performed"] = True + return payload + + monkeypatch.setattr( + agents, + "consume_latest_ai_agent_log_controlled_writeback", + fake_apply, + ) + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.post( + "/api/v1/agents/agent-log-controlled-writeback-consumer-apply" + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["schema_version"] == ( + "ai_agent_log_controlled_writeback_consumer_apply_receipt_v1" + ) + assert payload["status"] == "controlled_writeback_consumer_apply_ready" + assert payload["rollups"]["runtime_target_write_performed"] is True 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 979f0b005..7b96589ce 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 @@ -26,12 +26,15 @@ class _FakeMappingResult: class _FakeDb: - def __init__(self, rows: list[dict]): + def __init__(self, rows: list[dict], consumer_rows: list[dict] | None = None): self.rows = rows + self.consumer_rows = consumer_rows or [] self.params: list[dict] = [] async def execute(self, _statement, params: dict): self.params.append(params) + if params.get("operation_type") == "log_controlled_writeback_consumed": + return _FakeMappingResult(self.consumer_rows) return _FakeMappingResult(self.rows) @@ -79,6 +82,32 @@ def _ledger_rows() -> list[dict]: ] +def _consumer_receipt_rows() -> list[dict]: + return [ + { + "op_id": f"consumer-op-{target}", + "operation_type": "km_linked", + "actor": "ai_agent_metadata_writeback_consumer", + "status": "success", + "created_at": "2026-06-30T01:45:00+08:00", + "semantic_operation_type": "log_controlled_writeback_consumed", + "ledger_operation_type": "km_linked", + "consumer_receipt_id": f"log_controlled_writeback_consumed::{target}", + "source_dispatch_receipt_id": f"log_controlled_writeback_dispatched::{target}", + "target": target, + "target_surface": f"{target}_metadata_feedback_binding", + "consumer_surface": f"{target}_consumer_context", + "runtime_target_write_performed": "true", + "raw_payload_included": "false", + "consumer_context_receipt_recorded": "true", + "target_context_receipt_write_performed": "true", + "next_action": "readback_consumer_context_receipts", + "post_apply_verifier_refs": [f"post-write-verifier://{target}"], + } + for target in ("km", "rag", "playbook", "mcp", "verifier", "ai_agent") + ] + + @pytest.mark.asyncio async def test_log_controlled_writeback_consumer_loader_reads_live_ledger(monkeypatch): fake_db = _FakeDb(_ledger_rows()) @@ -91,7 +120,31 @@ async def test_log_controlled_writeback_consumer_loader_reads_live_ledger(monkey payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() _assert_consumer_readback(payload) - assert fake_db.params == [{"operation_type": "log_controlled_writeback_dispatched"}] + assert fake_db.params == [ + {"operation_type": "log_controlled_writeback_dispatched"}, + {"operation_type": "log_controlled_writeback_consumed"}, + ] + + +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_loader_reads_apply_receipts(monkeypatch): + fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows()) + monkeypatch.setattr( + consumer_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + + assert payload["controlled_consume"]["runtime_target_write_performed"] is True + assert payload["rollups"]["consumer_apply_receipt_row_count"] == 6 + assert payload["rollups"]["target_context_receipt_write_count"] == 6 + assert payload["rollups"]["runtime_target_write_performed"] is True + assert payload["operation_boundaries"]["runtime_target_write_performed"] is True + for binding in payload["consumer_bindings"]: + assert binding["target_write_performed"] is True + assert binding["target_write_receipt"]["runtime_target_write_performed"] is True def test_log_controlled_writeback_consumer_endpoint_returns_readback(monkeypatch):