diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index cfe08dc3e..a0f9e1949 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -100,6 +100,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_plan_readback import ( + load_latest_ai_agent_log_controlled_writeback_plan_readback, +) from src.services.ai_agent_log_feedback_receipt_dry_run import ( load_latest_ai_agent_log_feedback_receipt_dry_run, ) @@ -1903,6 +1906,38 @@ async def get_agent_log_post_write_verifier_dry_run() -> dict[str, Any]: ) from exc +@router.get( + "/agent-log-controlled-writeback-plan-readback", + response_model=dict[str, Any], + summary="取得 AI Agent LOG controlled writeback plan readback", + description=( + "把已驗證的 LOG feedback receipt 轉成 KM、RAG、PlayBook、MCP audit、" + "post-apply verifier 與 AI Agent 決策上下文的受控回寫計畫，包含 target selector、" + "source-of-truth diff、check-mode、rollback 與 post-apply verifier。" + "此端點只輸出 plan readback，不寫 KM、不寫 RAG index、不更新 PlayBook trust、" + "不呼叫 MCP tool、不執行 runtime action、不保存 raw log payload、不讀 secret、不呼叫 GitHub。" + ), +) +async def get_agent_log_controlled_writeback_plan_readback() -> dict[str, Any]: + """Return LOG feedback controlled writeback plan readback.""" + try: + payload = await asyncio.to_thread( + load_latest_ai_agent_log_controlled_writeback_plan_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_plan_readback_invalid", error=str(exc)) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="AI Agent LOG controlled writeback plan 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_plan_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py new file mode 100644 index 000000000..f75f666d1 --- /dev/null +++ b/apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py @@ -0,0 +1,273 @@ +"""AI Agent LOG controlled writeback plan readback. + +Builds the machine-readable target selectors, source-of-truth diffs, +check-mode gates, rollback refs, and post-apply verifier refs needed to turn +verified LOG feedback receipts into a controlled KM / RAG / PlayBook / MCP / +AI Agent writeback lane. This module prepares the apply plan only; it does not +perform writes, call MCP tools, trigger workflows, or persist raw log payloads. +""" + +from __future__ import annotations + +from typing import Any + +from src.services.ai_agent_log_feedback_receipt_dry_run import ( + load_latest_ai_agent_log_feedback_receipt_dry_run, +) +from src.services.ai_agent_log_post_write_verifier_dry_run import ( + load_latest_ai_agent_log_post_write_verifier_dry_run, +) + +_SCHEMA_VERSION = "ai_agent_log_controlled_writeback_plan_readback_v1" +_VERIFIER_READY_STATUS = "post_write_verifier_dry_run_ready" +_FEEDBACK_READY_STATUS = "trusted_feedback_receipt_dry_run_ready" +_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent") + + +def load_latest_ai_agent_log_controlled_writeback_plan_readback() -> dict[str, Any]: + """Return a controlled apply plan for verified LOG feedback receipts.""" + verifier = load_latest_ai_agent_log_post_write_verifier_dry_run() + feedback = load_latest_ai_agent_log_feedback_receipt_dry_run() + receipts = _candidate_receipts(feedback) + verifier_ready = ( + verifier.get("status") == _VERIFIER_READY_STATUS + and verifier.get("active_blockers") == [] + and (verifier.get("rollups") or {}).get("controlled_writeback_next_step_ready") + is True + ) + feedback_ready = ( + feedback.get("status") == _FEEDBACK_READY_STATUS + and feedback.get("active_blockers") == [] + and (feedback.get("rollups") or {}).get("dry_run_ready") is True + ) + writeback_plans = _writeback_plans(receipts) + active_blockers = _active_blockers( + verifier_ready=verifier_ready, + feedback_ready=feedback_ready, + receipts=receipts, + writeback_plans=writeback_plans, + ) + + return { + "schema_version": _SCHEMA_VERSION, + "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK", + "scope": "ai_agent_log_controlled_writeback_plan", + "status": ( + "controlled_writeback_plan_ready" + if not active_blockers + else "blocked_waiting_controlled_writeback_plan_inputs" + ), + "readback": { + "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-PLAN", + "workplan_title": "LOG feedback controlled KM / RAG / MCP / PlayBook / AI Agent writeback plan", + "feedback_schema_version": feedback.get("schema_version"), + "feedback_status": feedback.get("status"), + "verifier_schema_version": verifier.get("schema_version"), + "verifier_status": verifier.get("status"), + "safe_next_step": "execute_controlled_writeback_apply_with_post_apply_verifier", + }, + "controlled_apply": { + "mode": "controlled_apply_prepared", + "target_selector_required": True, + "source_of_truth_diff_required": True, + "check_mode_required": True, + "rollback_required": True, + "post_apply_verifier_required": True, + "owner_review_required_for_low_medium_high": False, + "critical_break_glass_required": True, + }, + "writeback_plans": writeback_plans, + "target_rollups": _target_rollups(writeback_plans), + "rollups": { + "target_count": len(_TARGETS), + "candidate_receipt_count": len(receipts), + "writeback_plan_count": len(writeback_plans), + "ready_writeback_plan_count": sum( + 1 for plan in writeback_plans if plan["status"] == "controlled_apply_ready" + ), + "check_mode_plan_count": sum( + 1 for plan in writeback_plans if plan["check_mode"]["enabled"] is True + ), + "rollback_plan_count": sum( + 1 for plan in writeback_plans if plan["rollback"]["required"] is True + ), + "post_apply_verifier_plan_count": sum( + 1 + for plan in writeback_plans + if plan["post_apply_verifier"]["required"] is True + ), + "controlled_writeback_plan_ready": not active_blockers, + "runtime_write_performed": False, + }, + "active_blockers": active_blockers, + "operation_boundaries": { + "plan_readback_only": True, + "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 _candidate_receipts(feedback: dict[str, Any]) -> list[dict[str, Any]]: + receipts = feedback.get("candidate_receipts") + if not isinstance(receipts, list): + return [] + return [receipt for receipt in receipts if isinstance(receipt, dict)] + + +def _writeback_plans(receipts: list[dict[str, Any]]) -> list[dict[str, Any]]: + plans = [] + for receipt in receipts: + target = str(receipt.get("target") or "") + source_sample_id = str(receipt.get("source_sample_id") or "") + receipt_id = str(receipt.get("receipt_id") or "") + plan_id = f"log-feedback-writeback::{target}::{source_sample_id}" + plans.append( + { + "plan_id": plan_id, + "receipt_id": receipt_id, + "target": target, + "status": "controlled_apply_ready", + "risk_tier": "medium" if target in {"km", "rag", "playbook"} else "low", + "target_selector": { + "project_id": receipt.get("project_id"), + "product": receipt.get("product"), + "service": receipt.get("service"), + "package": receipt.get("package"), + "tool": receipt.get("tool"), + "source_system": receipt.get("source_system"), + "source_sample_id": source_sample_id, + "target_surface": _target_surface(target), + }, + "source_of_truth_diff": { + "current_state": "metadata_only_feedback_not_written", + "desired_state": "metadata_only_feedback_receipt_bound_to_target", + "delta_kind": _delta_kind(target), + "raw_payload_included": False, + "redaction_state": receipt.get("redaction_state"), + "observed_event_count": receipt.get("observed_event_count"), + "observed_field_count": receipt.get("observed_field_count"), + }, + "check_mode": { + "enabled": True, + "checks": [ + "target_selector_resolves_single_surface", + "metadata_only_redaction_state", + "raw_log_payload_absent", + "post_apply_verifier_ref_present", + ], + }, + "rollback": { + "required": True, + "rollback_ref": f"rollback://ai-agent-log-feedback/{target}/{source_sample_id}", + "strategy": "mark_receipt_superseded_and_remove_target_binding", + }, + "post_apply_verifier": { + "required": True, + "verifier_ref": f"post-write-verifier://ai-agent-log-feedback/{target}/{source_sample_id}", + "canonical_readback": ( + "/api/v1/agents/agent-log-controlled-writeback-plan-readback" + ), + }, + "write_enabled_by_plan": False, + } + ) + return plans + + +def _target_rollups(plans: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [ + { + "target": target, + "plan_count": sum(1 for plan in plans if plan.get("target") == target), + "ready_plan_count": sum( + 1 + for plan in plans + if plan.get("target") == target + and plan.get("status") == "controlled_apply_ready" + ), + "check_mode_ready": all( + plan.get("check_mode", {}).get("enabled") is True + for plan in plans + if plan.get("target") == target + ), + "rollback_ready": all( + plan.get("rollback", {}).get("required") is True + for plan in plans + if plan.get("target") == target + ), + "post_apply_verifier_ready": all( + plan.get("post_apply_verifier", {}).get("required") is True + for plan in plans + if plan.get("target") == target + ), + } + for target in _TARGETS + ] + + +def _active_blockers( + *, + verifier_ready: bool, + feedback_ready: bool, + receipts: list[dict[str, Any]], + writeback_plans: list[dict[str, Any]], +) -> list[str]: + blockers = [] + if not verifier_ready: + blockers.append("post_write_verifier_dry_run_not_ready") + if not feedback_ready: + blockers.append("feedback_receipt_dry_run_not_ready") + if not receipts: + blockers.append("candidate_receipts_missing") + if len(writeback_plans) != len(receipts): + blockers.append("writeback_plan_count_mismatch") + for target in _TARGETS: + if not any(plan.get("target") == target for plan in writeback_plans): + blockers.append(f"{target}_writeback_plan_missing") + for plan in writeback_plans: + if plan.get("write_enabled_by_plan") is not False: + blockers.append(f"{plan.get('plan_id')}:write_enabled_by_plan_not_false") + if plan.get("source_of_truth_diff", {}).get("raw_payload_included") is not False: + blockers.append(f"{plan.get('plan_id')}:raw_payload_included") + return _unique(blockers) + + +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 _delta_kind(target: str) -> str: + return { + "km": "append_km_memory_receipt", + "rag": "index_metadata_only_rag_chunk", + "playbook": "append_playbook_trust_signal", + "mcp": "append_mcp_audit_feedback", + "verifier": "append_post_apply_verifier_feedback", + "ai_agent": "publish_agent_context_receipt", + }.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_plan_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py new file mode 100644 index 000000000..a80d8ec0a --- /dev/null +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py @@ -0,0 +1,114 @@ +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_plan_readback import ( + load_latest_ai_agent_log_controlled_writeback_plan_readback, +) + + +def test_log_controlled_writeback_plan_loader_builds_apply_plan(): + payload = load_latest_ai_agent_log_controlled_writeback_plan_readback() + + _assert_controlled_writeback_plan(payload) + + +def test_log_controlled_writeback_plan_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-plan-readback") + + assert response.status_code == 200 + _assert_controlled_writeback_plan(response.json()) + + +def _assert_controlled_writeback_plan(payload: dict): + assert payload["schema_version"] == "ai_agent_log_controlled_writeback_plan_readback_v1" + assert payload["priority"] == "P1-LOG-KM-RAG-MCP-PLAYBOOK" + assert payload["status"] == "controlled_writeback_plan_ready" + assert payload["active_blockers"] == [] + assert payload["readback"]["feedback_status"] == "trusted_feedback_receipt_dry_run_ready" + assert payload["readback"]["verifier_status"] == "post_write_verifier_dry_run_ready" + assert payload["readback"]["safe_next_step"] == ( + "execute_controlled_writeback_apply_with_post_apply_verifier" + ) + + controlled_apply = payload["controlled_apply"] + assert controlled_apply["mode"] == "controlled_apply_prepared" + assert controlled_apply["target_selector_required"] is True + assert controlled_apply["source_of_truth_diff_required"] is True + assert controlled_apply["check_mode_required"] is True + assert controlled_apply["rollback_required"] is True + assert controlled_apply["post_apply_verifier_required"] is True + assert controlled_apply["owner_review_required_for_low_medium_high"] is False + assert controlled_apply["critical_break_glass_required"] is True + + assert payload["rollups"]["target_count"] == 6 + assert payload["rollups"]["candidate_receipt_count"] == 12 + assert payload["rollups"]["writeback_plan_count"] == 12 + assert payload["rollups"]["ready_writeback_plan_count"] == 12 + assert payload["rollups"]["check_mode_plan_count"] == 12 + assert payload["rollups"]["rollback_plan_count"] == 12 + assert payload["rollups"]["post_apply_verifier_plan_count"] == 12 + assert payload["rollups"]["controlled_writeback_plan_ready"] is True + assert payload["rollups"]["runtime_write_performed"] is False + + target_rollups = {item["target"]: item for item in payload["target_rollups"]} + assert set(target_rollups) == {"km", "rag", "playbook", "mcp", "verifier", "ai_agent"} + for rollup in target_rollups.values(): + assert rollup["plan_count"] == 2 + assert rollup["ready_plan_count"] == 2 + assert rollup["check_mode_ready"] is True + assert rollup["rollback_ready"] is True + assert rollup["post_apply_verifier_ready"] is True + + plans = payload["writeback_plans"] + assert {plan["target"] for plan in plans} == set(target_rollups) + assert all(plan["status"] == "controlled_apply_ready" for plan in plans) + assert all(plan["write_enabled_by_plan"] is False for plan in plans) + for plan in plans: + selector = plan["target_selector"] + assert selector["project_id"] == "awoooi" + assert selector["product"] == "awoooi" + assert selector["service"] in {"awoooi-api", "awoooi-worker"} + assert selector["package"] == "apps/api" + assert selector["tool"] == "kubectl_logs_readback" + assert selector["source_system"] == "k8s_logs" + assert selector["target_surface"] + assert plan["source_of_truth_diff"]["current_state"] == ( + "metadata_only_feedback_not_written" + ) + assert plan["source_of_truth_diff"]["desired_state"] == ( + "metadata_only_feedback_receipt_bound_to_target" + ) + assert plan["source_of_truth_diff"]["raw_payload_included"] is False + assert plan["source_of_truth_diff"]["redaction_state"] in { + "metadata_only_no_raw_payload", + "metadata_only_no_redacted_payload", + } + assert plan["check_mode"]["enabled"] is True + assert "target_selector_resolves_single_surface" in plan["check_mode"]["checks"] + assert plan["rollback"]["required"] is True + assert plan["rollback"]["rollback_ref"].startswith( + "rollback://ai-agent-log-feedback/" + ) + assert plan["post_apply_verifier"]["required"] is True + assert plan["post_apply_verifier"]["verifier_ref"].startswith( + "post-write-verifier://ai-agent-log-feedback/" + ) + + boundaries = payload["operation_boundaries"] + assert boundaries["plan_readback_only"] is True + 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