feat(api): dry run log feedback receipts

apps/api/src/api/v1/agents.py

@@ -97,6 +97,9 @@ from src.services.ai_agent_interaction_learning_proof import (
 from src.services.ai_agent_learning_writeback_approval_package import (
     load_latest_ai_agent_learning_writeback_approval_package,
 )
+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_live_read_model_gate import (
     load_latest_ai_agent_live_read_model_gate,
 )
@@ -1835,6 +1838,37 @@ async def get_agent_log_intelligence_integration_readback() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-log-feedback-receipt-dry-run",
+    response_model=dict[str, Any],
+    summary="取得 AI Agent LOG feedback receipt dry-run",
+    description=(
+        "把 LOG intelligence runtime sample 轉成 KM、RAG、PlayBook、MCP audit、"
+        "post-apply verifier 與 AI Agent 決策上下文的 dry-run receipt 候選。"
+        "此端點不寫 KM、不寫 RAG index、不更新 PlayBook trust、不呼叫 MCP tool、"
+        "不執行 verifier、不觸發 runtime action、不保存 raw log payload、不讀 secret、不呼叫 GitHub。"
+    ),
+)
+async def get_agent_log_feedback_receipt_dry_run() -> dict[str, Any]:
+    """Return LOG feedback receipt dry-run candidates for AI automation."""
+    try:
+        payload = await asyncio.to_thread(
+            load_latest_ai_agent_log_feedback_receipt_dry_run
+        )
+        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_feedback_receipt_dry_run_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI Agent LOG feedback receipt dry-run 無效",
+        ) from exc
+
+
 @router.get(
     "/agent-telegram-receipt-approval-package",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_log_feedback_receipt_dry_run.py

@@ -0,0 +1,214 @@
+"""AI Agent LOG feedback receipt dry-run.
+
+Transforms the LOG intelligence runtime sample readback into structured
+candidate receipts for KM, RAG, PlayBook, MCP audit, verifier feedback, and AI
+Agent decision context. This module is deliberately dry-run only: no KM writes,
+no vector indexing, no PlayBook trust update, no MCP tool call, and no runtime
+repair.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from src.services.ai_agent_log_intelligence_integration_readback import (
+    load_latest_ai_agent_log_intelligence_integration_readback,
+)
+
+_SCHEMA_VERSION = "ai_agent_log_feedback_receipt_dry_run_v1"
+
+_PIPELINE_NODES: tuple[dict[str, Any], ...] = (
+    {
+        "node_id": "collect_labeled_runtime_logs",
+        "stage": "ingest",
+        "target": "runtime_log_sample",
+        "produces": "metadata_only_log_receipt",
+    },
+    {
+        "node_id": "normalize_service_package_tool_labels",
+        "stage": "label",
+        "target": "label_taxonomy",
+        "produces": "project_product_service_package_tool_labels",
+    },
+    {
+        "node_id": "classify_signal_for_agent_context",
+        "stage": "classify",
+        "target": "agent_context",
+        "produces": "signal_kind_and_risk_context",
+    },
+    {
+        "node_id": "build_km_memory_candidates",
+        "stage": "candidate",
+        "target": "km",
+        "produces": "km_memory_receipt_candidate",
+    },
+    {
+        "node_id": "build_rag_chunk_candidates",
+        "stage": "candidate",
+        "target": "rag",
+        "produces": "rag_chunk_receipt_candidate",
+    },
+    {
+        "node_id": "build_playbook_trust_candidates",
+        "stage": "candidate",
+        "target": "playbook",
+        "produces": "playbook_trust_receipt_candidate",
+    },
+    {
+        "node_id": "build_mcp_audit_feedback_candidates",
+        "stage": "candidate",
+        "target": "mcp",
+        "produces": "mcp_audit_feedback_candidate",
+    },
+    {
+        "node_id": "bind_post_apply_verifier_feedback",
+        "stage": "verify",
+        "target": "verifier",
+        "produces": "post_apply_verifier_feedback_receipt",
+    },
+    {
+        "node_id": "publish_agent_decision_context_dry_run",
+        "stage": "agent_context",
+        "target": "ai_agent",
+        "produces": "agent_decision_context_dry_run",
+    },
+)
+
+_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
+
+
+def load_latest_ai_agent_log_feedback_receipt_dry_run() -> dict[str, Any]:
+    """Return a dry-run receipt plan derived from runtime LOG samples."""
+    source = load_latest_ai_agent_log_intelligence_integration_readback()
+    rollups = source.get("rollups") or {}
+    runtime_sample = source.get("runtime_e2e_verifier") or {}
+    samples = _sample_rows(runtime_sample)
+    runtime_ready = rollups.get("runtime_e2e_log_sample_readback_present") is True
+    source_ready = (
+        source.get("status") == "controlled_apply_ready_for_trusted_feedback_receipt"
+        and rollups.get("ready_lane_count") == rollups.get("lane_count")
+    )
+    candidate_receipts = _candidate_receipts(samples)
+    active_blockers = []
+    if not source_ready:
+        active_blockers.append("log_intelligence_source_readback_not_ready")
+    if not runtime_ready:
+        active_blockers.append("runtime_sample_receipt_missing")
+
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
+        "scope": "ai_agent_log_feedback_receipt_dry_run",
+        "status": (
+            "trusted_feedback_receipt_dry_run_ready"
+            if not active_blockers
+            else "blocked_waiting_log_intelligence_runtime_receipt"
+        ),
+        "readback": {
+            "workplan_id": "P1-LOG-FEEDBACK-DRY-RUN",
+            "workplan_title": "LOG 貼標後產出 KM / RAG / MCP / PlayBook / AI Agent dry-run receipt",
+            "source_schema_version": source.get("schema_version"),
+            "source_status": source.get("status"),
+            "safe_next_step": "enable_post_write_verifier_for_km_rag_playbook_trust_receipts",
+        },
+        "pipeline_nodes": _pipeline_nodes(source_ready=source_ready, runtime_ready=runtime_ready),
+        "candidate_receipts": candidate_receipts,
+        "rollups": {
+            "pipeline_node_count": len(_PIPELINE_NODES),
+            "completed_node_count": len(_PIPELINE_NODES) if not active_blockers else 0,
+            "target_count": len(_TARGETS),
+            "candidate_receipt_count": len(candidate_receipts),
+            "runtime_sample_count": len(samples),
+            "source_lane_count": rollups.get("lane_count", 0),
+            "source_ready_lane_count": rollups.get("ready_lane_count", 0),
+            "dry_run_ready": not active_blockers,
+            "km_candidate_count": _count_target(candidate_receipts, "km"),
+            "rag_candidate_count": _count_target(candidate_receipts, "rag"),
+            "playbook_candidate_count": _count_target(candidate_receipts, "playbook"),
+            "mcp_candidate_count": _count_target(candidate_receipts, "mcp"),
+            "verifier_candidate_count": _count_target(candidate_receipts, "verifier"),
+            "ai_agent_context_candidate_count": _count_target(candidate_receipts, "ai_agent"),
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "dry_run_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,
+            "post_apply_verifier_executed": False,
+            "raw_log_payload_persisted": False,
+            "secret_value_collection_allowed": False,
+            "workflow_trigger_performed": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def _pipeline_nodes(*, source_ready: bool, runtime_ready: bool) -> list[dict[str, Any]]:
+    status = "dry_run_ready" if source_ready and runtime_ready else "blocked_waiting_source_receipt"
+    return [
+        {
+            **node,
+            "status": status,
+            "write_enabled": False,
+            "requires_post_write_verifier": node["target"] in {"km", "rag", "playbook", "verifier"},
+        }
+        for node in _PIPELINE_NODES
+    ]
+
+
+def _sample_rows(runtime_sample: dict[str, Any]) -> list[dict[str, Any]]:
+    samples = runtime_sample.get("samples")
+    if not isinstance(samples, list):
+        return []
+    rows = []
+    for sample in samples:
+        if not isinstance(sample, dict):
+            continue
+        rows.append(
+            {
+                "sample_id": str(sample.get("sample_id") or ""),
+                "service": str(sample.get("service") or ""),
+                "source_system": str(sample.get("source_system") or ""),
+                "project_id": str(sample.get("project_id") or ""),
+                "product": str(sample.get("product") or ""),
+                "package": str(sample.get("package") or ""),
+                "tool": str(sample.get("tool") or ""),
+                "redaction_state": str(sample.get("redaction_state") or ""),
+                "observed_events": [str(item) for item in sample.get("observed_events") or []],
+                "observed_fields": [str(item) for item in sample.get("observed_fields") or []],
+            }
+        )
+    return rows
+
+
+def _candidate_receipts(samples: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    receipts: list[dict[str, Any]] = []
+    for sample in samples:
+        for target in _TARGETS:
+            receipts.append(
+                {
+                    "receipt_id": f"{sample['sample_id']}::{target}",
+                    "target": target,
+                    "source_sample_id": sample["sample_id"],
+                    "project_id": sample["project_id"],
+                    "product": sample["product"],
+                    "service": sample["service"],
+                    "package": sample["package"],
+                    "tool": sample["tool"],
+                    "source_system": sample["source_system"],
+                    "redaction_state": sample["redaction_state"],
+                    "observed_event_count": len(sample["observed_events"]),
+                    "observed_field_count": len(sample["observed_fields"]),
+                    "dry_run_status": "candidate_ready",
+                    "write_enabled": False,
+                    "raw_log_payload_persisted": False,
+                }
+            )
+    return receipts
+
+
+def _count_target(receipts: list[dict[str, Any]], target: str) -> int:
+    return sum(1 for receipt in receipts if receipt.get("target") == target)