From 6efb5fc97c3c46be6e1c45cfb397544b11f9c607 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 29 Jun 2026 23:45:02 +0800 Subject: [PATCH] feat(api): expose log writeback executor readback --- apps/api/src/api/v1/agents.py | 35 ++ ..._controlled_writeback_executor_readback.py | 310 ++++++++++++++++++ ...trolled_writeback_executor_readback_api.py | 113 +++++++ 3 files changed, 458 insertions(+) create mode 100644 apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py create mode 100644 apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 2d7c4a95e..b761ae621 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -103,6 +103,9 @@ from src.services.ai_agent_live_read_model_gate import ( from src.services.ai_agent_log_controlled_writeback_plan_readback import ( load_latest_ai_agent_log_controlled_writeback_plan_readback, ) +from src.services.ai_agent_log_controlled_writeback_executor_readback import ( + load_latest_ai_agent_log_controlled_writeback_executor_readback, +) from src.services.ai_agent_log_feedback_receipt_dry_run import ( load_latest_ai_agent_log_feedback_receipt_dry_run, ) @@ -1964,6 +1967,38 @@ async def get_agent_log_controlled_writeback_plan_readback() -> dict[str, Any]: ) from exc +@router.get( + "/agent-log-controlled-writeback-executor-readback", + response_model=dict[str, Any], + summary="取得 AI Agent LOG controlled writeback executor readback", + description=( + "把 LOG controlled writeback plan 轉成 AI Agent 可消費的 executor batch、" + "next action queue、check-mode、rollback 與 post-apply verifier readback。" + "低中高風險 metadata writeback 採 AI controlled apply；critical 仍需 break-glass。" + "此端點只輸出 executor readback，不 dispatch executor、不寫 KM、不寫 RAG index、" + "不更新 PlayBook trust、不呼叫 MCP tool、不保存 raw log payload、不讀 secret、不呼叫 GitHub。" + ), +) +async def get_agent_log_controlled_writeback_executor_readback() -> dict[str, Any]: + """Return LOG feedback controlled writeback executor and consumption readback.""" + try: + payload = await asyncio.to_thread( + load_latest_ai_agent_log_controlled_writeback_executor_readback + ) + return redact_public_lan_topology(payload) + except FileNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except (json.JSONDecodeError, ValueError) as exc: + logger.error("ai_agent_log_controlled_writeback_executor_readback_invalid", error=str(exc)) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="AI Agent LOG controlled writeback executor readback 無效", + ) from exc + + @router.get( "/agent-telegram-receipt-approval-package", response_model=dict[str, Any], diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py new file mode 100644 index 000000000..3e7cb98d1 --- /dev/null +++ b/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py @@ -0,0 +1,310 @@ +"""AI Agent LOG controlled writeback executor readback. + +Turns the verified LOG controlled writeback plan into executor-ready batches +and AI Agent consumption context. This endpoint-level readback opens the +controlled-apply route for low / medium / high metadata writeback decisions, +while keeping the function itself side-effect free: it does not dispatch an +executor, write KM, index RAG, update PlayBook trust, call MCP tools, trigger +workflows, or persist raw log payloads. +""" + +from __future__ import annotations + +from typing import Any + +from src.services.ai_agent_log_controlled_writeback_plan_readback import ( + load_latest_ai_agent_log_controlled_writeback_plan_readback, +) + +_SCHEMA_VERSION = "ai_agent_log_controlled_writeback_executor_readback_v1" +_PLAN_READY_STATUS = "controlled_writeback_plan_ready" +_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent") +_EXECUTOR_ROUTE = "ai_agent_metadata_writeback_executor" + + +def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[str, Any]: + """Return executor-ready LOG writeback batches for AI Agent consumption.""" + plan = load_latest_ai_agent_log_controlled_writeback_plan_readback() + writeback_plans = _writeback_plans(plan) + plan_ready = ( + plan.get("status") == _PLAN_READY_STATUS + and plan.get("active_blockers") == [] + and (plan.get("rollups") or {}).get("controlled_writeback_plan_ready") is True + ) + execution_batches = _execution_batches(writeback_plans) + active_blockers = _active_blockers( + plan_ready=plan_ready, + writeback_plans=writeback_plans, + execution_batches=execution_batches, + ) + + return { + "schema_version": _SCHEMA_VERSION, + "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK", + "scope": "ai_agent_log_controlled_writeback_executor", + "status": ( + "controlled_writeback_executor_ready" + if not active_blockers + else "blocked_waiting_controlled_writeback_executor_inputs" + ), + "readback": { + "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-EXECUTOR", + "workplan_title": "LOG feedback controlled writeback executor and AI Agent consumption readback", + "source_schema_version": plan.get("schema_version"), + "source_status": plan.get("status"), + "safe_next_step": "dispatch_controlled_metadata_writeback_batches_then_post_apply_verify", + }, + "executor_policy": { + "executor_route": _EXECUTOR_ROUTE, + "low_medium_high_controlled_apply_enabled": True, + "owner_review_required_for_low_medium_high": False, + "critical_break_glass_required": True, + "target_selector_required": True, + "source_of_truth_diff_required": True, + "check_mode_required": True, + "rollback_required": True, + "post_apply_verifier_required": True, + }, + "execution_batches": execution_batches, + "agent_consumption_context": _agent_consumption_context(execution_batches), + "rollups": { + "source_writeback_plan_count": len(writeback_plans), + "execution_batch_count": len(execution_batches), + "ready_execution_batch_count": sum( + 1 + for batch in execution_batches + if batch["status"] == "ready_for_controlled_executor_dispatch" + ), + "target_count": len(_TARGETS), + "target_selector_count": sum( + batch["target_selector_count"] for batch in execution_batches + ), + "source_of_truth_diff_count": sum( + batch["source_of_truth_diff_count"] for batch in execution_batches + ), + "check_mode_ready_count": sum( + 1 for batch in execution_batches if batch["check_mode"]["enabled"] is True + ), + "rollback_ready_count": sum( + 1 for batch in execution_batches if batch["rollback"]["required"] is True + ), + "post_apply_verifier_ready_count": sum( + 1 + for batch in execution_batches + if batch["post_apply_verifier"]["required"] is True + ), + "controlled_executor_dispatch_ready": not active_blockers, + "controlled_apply_enabled_by_policy": True, + "runtime_dispatch_performed": False, + }, + "active_blockers": active_blockers, + "operation_boundaries": { + "executor_readback_only": True, + "controlled_apply_enabled_by_policy": True, + "executor_dispatch_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, + "workflow_trigger_performed": False, + "raw_log_payload_persisted": False, + "secret_value_collection_allowed": False, + "github_api_used": False, + }, + } + + +def _writeback_plans(plan: dict[str, Any]) -> list[dict[str, Any]]: + plans = plan.get("writeback_plans") + if not isinstance(plans, list): + return [] + return [item for item in plans if isinstance(item, dict)] + + +def _execution_batches(writeback_plans: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [ + _execution_batch(target, _plans_for_target(writeback_plans, target)) + for target in _TARGETS + ] + + +def _execution_batch(target: str, plans: list[dict[str, Any]]) -> dict[str, Any]: + risk_tier = "medium" if target in {"km", "rag", "playbook"} else "low" + target_selectors = [plan.get("target_selector") or {} for plan in plans] + diffs = [plan.get("source_of_truth_diff") or {} for plan in plans] + rollback_refs = [ + str((plan.get("rollback") or {}).get("rollback_ref") or "") + for plan in plans + ] + verifier_refs = [ + str((plan.get("post_apply_verifier") or {}).get("verifier_ref") or "") + for plan in plans + ] + ready = _batch_ready(plans) + return { + "batch_id": f"log-feedback-controlled-writeback::{target}", + "target": target, + "target_surface": _target_surface(target), + "risk_tier": risk_tier, + "executor_route": _EXECUTOR_ROUTE, + "status": ( + "ready_for_controlled_executor_dispatch" + if ready + else "blocked_waiting_batch_controls" + ), + "apply_mode": "controlled_apply", + "dispatch_enabled_by_policy": True, + "plan_count": len(plans), + "plan_ids": [str(plan.get("plan_id") or "") for plan in plans], + "receipt_ids": [str(plan.get("receipt_id") or "") for plan in plans], + "target_selector_count": len(target_selectors), + "target_selectors": target_selectors, + "source_of_truth_diff_count": len(diffs), + "source_of_truth_diffs": diffs, + "check_mode": { + "enabled": True, + "required": True, + "checks": [ + "resolve_target_selectors", + "compare_source_of_truth_diffs", + "verify_metadata_only_redaction", + "verify_rollback_refs", + "verify_post_apply_verifier_refs", + ], + }, + "rollback": { + "required": True, + "rollback_refs": rollback_refs, + "strategy": "mark_receipts_superseded_and_remove_target_bindings", + }, + "post_apply_verifier": { + "required": True, + "verifier_refs": verifier_refs, + "canonical_readback": ( + "/api/v1/agents/agent-log-controlled-writeback-executor-readback" + ), + }, + "runtime_dispatch_performed": False, + } + + +def _batch_ready(plans: list[dict[str, Any]]) -> bool: + if not plans: + return False + for plan in plans: + if plan.get("status") != "controlled_apply_ready": + return False + if plan.get("write_enabled_by_plan") is not False: + return False + if not plan.get("target_selector"): + return False + if not plan.get("source_of_truth_diff"): + return False + if (plan.get("check_mode") or {}).get("enabled") is not True: + return False + if (plan.get("rollback") or {}).get("required") is not True: + return False + if (plan.get("post_apply_verifier") or {}).get("required") is not True: + return False + return True + + +def _agent_consumption_context(execution_batches: list[dict[str, Any]]) -> dict[str, Any]: + ready_batches = [ + batch + for batch in execution_batches + if batch["status"] == "ready_for_controlled_executor_dispatch" + ] + return { + "context_id": "ai-agent-log-controlled-writeback-consumption-v1", + "consumable_by": [ + "ai_agent_autonomous_runtime_control", + "awooop_work_items", + "alert_triage_loop", + "km_rag_playbook_learning_loop", + "mcp_audit_context_loop", + ], + "evidence_chain": [ + "/api/v1/agents/agent-log-intelligence-integration-readback", + "/api/v1/agents/agent-log-feedback-receipt-dry-run", + "/api/v1/agents/agent-log-post-write-verifier-dry-run", + "/api/v1/agents/agent-log-controlled-writeback-plan-readback", + ], + "next_action_queue": [ + { + "batch_id": batch["batch_id"], + "target": batch["target"], + "executor_route": batch["executor_route"], + "apply_mode": batch["apply_mode"], + "plan_count": batch["plan_count"], + "check_mode_required": batch["check_mode"]["required"], + "rollback_required": batch["rollback"]["required"], + "post_apply_verifier_required": batch["post_apply_verifier"]["required"], + } + for batch in ready_batches + ], + "learning_feedback_targets": [ + "km", + "rag", + "playbook", + "mcp", + "verifier", + "ai_agent", + ], + "raw_payload_required": False, + } + + +def _active_blockers( + *, + plan_ready: bool, + writeback_plans: list[dict[str, Any]], + execution_batches: list[dict[str, Any]], +) -> list[str]: + blockers = [] + if not plan_ready: + blockers.append("controlled_writeback_plan_not_ready") + if not writeback_plans: + blockers.append("source_writeback_plans_missing") + for target in _TARGETS: + batch = next( + (item for item in execution_batches if item.get("target") == target), + None, + ) + if not batch: + blockers.append(f"{target}_execution_batch_missing") + continue + if batch["status"] != "ready_for_controlled_executor_dispatch": + blockers.append(f"{target}_execution_batch_not_ready") + if batch["dispatch_enabled_by_policy"] is not True: + blockers.append(f"{target}_dispatch_policy_not_enabled") + if batch["runtime_dispatch_performed"] is not False: + blockers.append(f"{target}_runtime_dispatch_already_performed") + return _unique(blockers) + + +def _plans_for_target(plans: list[dict[str, Any]], target: str) -> list[dict[str, Any]]: + return [plan for plan in plans if plan.get("target") == target] + + +def _target_surface(target: str) -> str: + return { + "km": "knowledge_memory", + "rag": "rag_chunk_index", + "playbook": "playbook_trust_learning", + "mcp": "mcp_audit_context", + "verifier": "post_apply_verifier_feedback", + "ai_agent": "agent_decision_context", + }.get(target, "unknown") + + +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/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py new file mode 100644 index 000000000..506853a59 --- /dev/null +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.api.v1.agents import router +from src.services.ai_agent_log_controlled_writeback_executor_readback import ( + load_latest_ai_agent_log_controlled_writeback_executor_readback, +) + + +def test_log_controlled_writeback_executor_loader_builds_batches(): + payload = load_latest_ai_agent_log_controlled_writeback_executor_readback() + + _assert_executor_readback(payload) + + +def test_log_controlled_writeback_executor_endpoint_returns_readback(): + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.get("/api/v1/agents/agent-log-controlled-writeback-executor-readback") + + assert response.status_code == 200 + _assert_executor_readback(response.json()) + + +def _assert_executor_readback(payload: dict): + assert payload["schema_version"] == "ai_agent_log_controlled_writeback_executor_readback_v1" + assert payload["priority"] == "P1-LOG-KM-RAG-MCP-PLAYBOOK" + assert payload["status"] == "controlled_writeback_executor_ready" + assert payload["active_blockers"] == [] + assert payload["readback"]["source_status"] == "controlled_writeback_plan_ready" + assert payload["readback"]["safe_next_step"] == ( + "dispatch_controlled_metadata_writeback_batches_then_post_apply_verify" + ) + + policy = payload["executor_policy"] + assert policy["executor_route"] == "ai_agent_metadata_writeback_executor" + assert policy["low_medium_high_controlled_apply_enabled"] is True + assert policy["owner_review_required_for_low_medium_high"] is False + assert policy["critical_break_glass_required"] is True + assert policy["target_selector_required"] is True + assert policy["source_of_truth_diff_required"] is True + assert policy["check_mode_required"] is True + assert policy["rollback_required"] is True + assert policy["post_apply_verifier_required"] is True + + assert payload["rollups"]["source_writeback_plan_count"] == 12 + assert payload["rollups"]["execution_batch_count"] == 6 + assert payload["rollups"]["ready_execution_batch_count"] == 6 + assert payload["rollups"]["target_count"] == 6 + assert payload["rollups"]["target_selector_count"] == 12 + assert payload["rollups"]["source_of_truth_diff_count"] == 12 + assert payload["rollups"]["check_mode_ready_count"] == 6 + assert payload["rollups"]["rollback_ready_count"] == 6 + assert payload["rollups"]["post_apply_verifier_ready_count"] == 6 + assert payload["rollups"]["controlled_executor_dispatch_ready"] is True + assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True + assert payload["rollups"]["runtime_dispatch_performed"] is False + + batches = {batch["target"]: batch for batch in payload["execution_batches"]} + assert set(batches) == {"km", "rag", "playbook", "mcp", "verifier", "ai_agent"} + for target, batch in batches.items(): + assert batch["batch_id"] == f"log-feedback-controlled-writeback::{target}" + assert batch["executor_route"] == "ai_agent_metadata_writeback_executor" + assert batch["status"] == "ready_for_controlled_executor_dispatch" + assert batch["apply_mode"] == "controlled_apply" + assert batch["dispatch_enabled_by_policy"] is True + assert batch["plan_count"] == 2 + assert len(batch["plan_ids"]) == 2 + assert len(batch["receipt_ids"]) == 2 + assert batch["target_selector_count"] == 2 + assert batch["source_of_truth_diff_count"] == 2 + assert batch["check_mode"]["enabled"] is True + assert batch["check_mode"]["required"] is True + assert batch["rollback"]["required"] is True + assert len(batch["rollback"]["rollback_refs"]) == 2 + assert batch["post_apply_verifier"]["required"] is True + assert len(batch["post_apply_verifier"]["verifier_refs"]) == 2 + assert batch["runtime_dispatch_performed"] is False + + context = payload["agent_consumption_context"] + assert context["context_id"] == "ai-agent-log-controlled-writeback-consumption-v1" + assert "ai_agent_autonomous_runtime_control" in context["consumable_by"] + assert "km_rag_playbook_learning_loop" in context["consumable_by"] + assert len(context["evidence_chain"]) == 4 + assert len(context["next_action_queue"]) == 6 + assert set(item["target"] for item in context["next_action_queue"]) == set(batches) + assert context["learning_feedback_targets"] == [ + "km", + "rag", + "playbook", + "mcp", + "verifier", + "ai_agent", + ] + assert context["raw_payload_required"] is False + + boundaries = payload["operation_boundaries"] + assert boundaries["executor_readback_only"] is True + assert boundaries["controlled_apply_enabled_by_policy"] is True + assert boundaries["executor_dispatch_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["agent_runtime_action_performed"] is False + assert boundaries["workflow_trigger_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