feat(api): plan controlled log writeback

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],

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

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