feat(api): attach log intelligence runtime sample

apps/api/src/services/ai_agent_log_intelligence_integration_readback.py

@@ -8,12 +8,15 @@ call MCP tools, trigger workflows, or execute runtime repairs.
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Any
 
 from src.services.snapshot_paths import resolve_repo_root
 
 _SCHEMA_VERSION = "ai_agent_log_intelligence_integration_readback_v1"
+_RUNTIME_SAMPLE_SCHEMA_VERSION = "ai_agent_log_intelligence_runtime_sample_readback_v1"
+_RUNTIME_SAMPLE_FILE = "ai-agent-log-intelligence-runtime-sample-readback.snapshot.json"
 _DEFAULT_REPO_ROOT = resolve_repo_root(Path(__file__))
 
 _LANES: tuple[dict[str, Any], ...] = (
@@ -141,33 +144,48 @@ def load_latest_ai_agent_log_intelligence_integration_readback(
     lanes = [_build_lane(root, lane) for lane in _LANES]
     ready_lane_count = sum(1 for lane in lanes if lane["status"] == "source_refs_present")
     missing_lane_count = len(lanes) - ready_lane_count
+    runtime_sample = _load_runtime_sample(root)
+    runtime_sample_present = runtime_sample is not None
     label_field_count = len(
         {field for group in _LABEL_TAXONOMY for field in group["required_fields"]}
     )
     active_blockers = []
     if missing_lane_count:
         active_blockers.append("committed_source_refs_missing")
-    active_blockers.append("runtime_e2e_log_sample_readback_missing")
+    if not runtime_sample_present:
+        active_blockers.append("runtime_e2e_log_sample_readback_missing")
 
     return {
         "schema_version": _SCHEMA_VERSION,
         "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
         "scope": "ai_agent_log_intelligence_integration",
         "status": (
-            "controlled_apply_ready_missing_runtime_e2e_log_sample"
-            if not missing_lane_count
-            else "blocked_missing_committed_source_refs"
+            "blocked_missing_committed_source_refs"
+            if missing_lane_count
+            else (
+                "controlled_apply_ready_for_trusted_feedback_receipt"
+                if runtime_sample_present
+                else "controlled_apply_ready_missing_runtime_e2e_log_sample"
+            )
         ),
         "readback": {
             "workplan_id": "P1-LOG-INTELLIGENCE",
             "workplan_title": "所有服務日誌貼標並串接 KM / RAG / MCP / PlayBook / AI Agent",
             "repo_root_ref": str(root),
             "safe_next_step": (
-                "add_runtime_log_sample_verifier_then_write_trusted_KM_RAG_PlayBook_feedback_receipt"
+                "run_trusted_feedback_receipt_dry_run_then_enable_km_rag_playbook_verifier"
+                if runtime_sample_present
+                else "add_runtime_log_sample_verifier_then_write_trusted_KM_RAG_PlayBook_feedback_receipt"
             ),
         },
         "integration_lanes": lanes,
         "label_taxonomy": list(_LABEL_TAXONOMY),
+        "runtime_e2e_verifier": runtime_sample
+        or {
+            "schema_version": _RUNTIME_SAMPLE_SCHEMA_VERSION,
+            "status": "missing_runtime_sample_readback",
+            "expected_ref": f"docs/operations/{_RUNTIME_SAMPLE_FILE}",
+        },
         "rollups": {
             "lane_count": len(lanes),
             "ready_lane_count": ready_lane_count,
@@ -177,7 +195,22 @@ def load_latest_ai_agent_log_intelligence_integration_readback(
             "readiness_percent": _percent(ready_lane_count / max(len(lanes), 1) * 100),
             "label_group_count": len(_LABEL_TAXONOMY),
             "required_label_field_count": label_field_count,
-            "runtime_e2e_log_sample_readback_present": False,
+            "runtime_e2e_log_sample_readback_present": runtime_sample_present,
+            "runtime_sample_count": len(runtime_sample.get("samples", []))
+            if runtime_sample
+            else 0,
+            "km_rag_feedback_receipt_ready": _feedback_ready(
+                runtime_sample,
+                "km_rag_feedback_receipt_ready",
+            ),
+            "playbook_feedback_receipt_ready": _feedback_ready(
+                runtime_sample,
+                "playbook_feedback_receipt_ready",
+            ),
+            "mcp_audit_feedback_receipt_ready": _feedback_ready(
+                runtime_sample,
+                "mcp_audit_feedback_receipt_ready",
+            ),
             "km_rag_playbook_trust_writeback_authorized": False,
             "mcp_tool_execution_authorized": False,
         },
@@ -198,6 +231,73 @@ def load_latest_ai_agent_log_intelligence_integration_readback(
     }
 
 
+def _load_runtime_sample(root: Path) -> dict[str, Any] | None:
+    path = root / "docs" / "operations" / _RUNTIME_SAMPLE_FILE
+    if not path.exists():
+        return None
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    _require_runtime_sample(payload, str(path))
+    return payload
+
+
+def _require_runtime_sample(payload: dict[str, Any], label: str) -> None:
+    if payload.get("schema_version") != _RUNTIME_SAMPLE_SCHEMA_VERSION:
+        raise ValueError(f"{label}: unsupported schema_version={payload.get('schema_version')!r}")
+    samples = payload.get("samples")
+    if not isinstance(samples, list) or not samples:
+        raise ValueError(f"{label}: samples must be a non-empty list")
+    for sample in samples:
+        if not isinstance(sample, dict):
+            raise ValueError(f"{label}: sample must be an object")
+        for field in ("sample_id", "service", "source_system", "environment", "project_id"):
+            if not sample.get(field):
+                raise ValueError(f"{label}: sample missing {field}")
+        if sample.get("raw_log_payload_persisted") is not False:
+            raise ValueError(f"{label}: raw log payload must not be persisted")
+
+    receipt = payload.get("feedback_receipt_readiness") or {}
+    required_ready = {
+        "km_rag_feedback_receipt_ready",
+        "playbook_feedback_receipt_ready",
+        "mcp_audit_feedback_receipt_ready",
+        "post_apply_verifier_feedback_receipt_ready",
+        "trusted_writeback_still_requires_dry_run",
+    }
+    missing_ready = sorted(field for field in required_ready if receipt.get(field) is not True)
+    if missing_ready:
+        raise ValueError(f"{label}: feedback receipt readiness missing: {missing_ready}")
+
+    boundaries = payload.get("operation_boundaries") or {}
+    if boundaries.get("runtime_log_sample_read_performed") is not True:
+        raise ValueError(f"{label}: runtime log sample readback must be true")
+    forbidden_true = sorted(
+        field
+        for field in (
+            "raw_log_payload_persisted",
+            "secret_value_collection_allowed",
+            "km_write_performed",
+            "rag_index_write_performed",
+            "playbook_trust_write_performed",
+            "mcp_tool_call_performed",
+            "runtime_repair_performed",
+            "workflow_trigger_performed",
+            "github_api_used",
+        )
+        if boundaries.get(field) is not False
+    )
+    if forbidden_true:
+        raise ValueError(f"{label}: forbidden operation boundary true: {forbidden_true}")
+
+
+def _feedback_ready(runtime_sample: dict[str, Any] | None, field: str) -> bool:
+    if not runtime_sample:
+        return False
+    return (runtime_sample.get("feedback_receipt_readiness") or {}).get(field) is True
+
+
 def _build_lane(root: Path, lane: dict[str, Any]) -> dict[str, Any]:
     evidence_refs = [str(ref) for ref in lane["required_refs"]]
     missing_refs = [ref for ref in evidence_refs if not _ref_exists(root, ref)]