feat(agents): 新增 P2-411 owner acceptance event bus

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

@@ -49,6 +49,9 @@ from src.services.ai_agent_12_agent_war_room import (
 from src.services.ai_agent_action_audit_ledger import (
     load_latest_ai_agent_action_audit_ledger,
 )
+from src.services.ai_agent_action_owner_acceptance_event_bus import (
+    load_latest_ai_agent_action_owner_acceptance_event_bus,
+)
 from src.services.ai_agent_automation_backlog_snapshot import (
     load_latest_ai_agent_automation_backlog_snapshot,
 )
@@ -972,6 +975,37 @@ async def get_agent_action_audit_ledger() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-action-owner-acceptance-event-bus",
+    response_model=dict[str, Any],
+    summary="取得 P2-411 AI Agent Owner Acceptance / Handoff Event Bus",
+    description=(
+        "讀取最新已提交的 P2-411 AI Agent owner acceptance / handoff event bus "
+        "no-write 快照；此端點只呈現 owner acceptance lane、handoff event template、"
+        "RAG memory proposal、verifier gate 與 no-write activation boundary。它不 publish event bus、"
+        "不寫 audit DB、不寫 timeline、不寫 KM、不更新 PlayBook trust、不寫 Gateway queue、"
+        "不送 Telegram、不呼叫 Bot API、不 dispatch worker、不寫 production、不讀 secret、"
+        "不呼叫付費 API、不改主機、不執行 kubectl 或不可逆操作。"
+    ),
+)
+async def get_agent_action_owner_acceptance_event_bus() -> dict[str, Any]:
+    """回傳最新 P2-411 owner acceptance / handoff event bus 只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_ai_agent_action_owner_acceptance_event_bus)
+        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_action_owner_acceptance_event_bus_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="P2-411 AI Agent Owner Acceptance / Handoff Event Bus 快照無效",
+        ) from exc
+
+
 @router.get(
     "/agent-host-runaway-aiops-loop-readiness",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_action_owner_acceptance_event_bus.py

@@ -0,0 +1,430 @@
+"""
+P2-411 AI Agent action owner acceptance event bus snapshot.
+
+Loads the latest committed owner acceptance / handoff event bus baseline. This
+module validates no-write owner acceptance lanes, handoff event templates, and
+RAG memory proposals. It never publishes event bus messages, writes audit DB
+rows, timeline events, KM, PlayBook trust, Gateway queues, Telegram messages,
+secrets, hosts, Kubernetes resources, or production state.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_evaluations_dir
+
+_DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__))
+_SNAPSHOT_PATTERN = "ai_agent_action_owner_acceptance_event_bus_*.json"
+_SCHEMA_VERSION = "ai_agent_action_owner_acceptance_event_bus_v1"
+_RUNTIME_AUTHORITY = "agent_action_owner_acceptance_event_bus_no_write_committed_snapshot"
+_EXPECTED_CURRENT_TASK = "P2-411"
+_EXPECTED_NEXT_TASK = "P2-412"
+_EXPECTED_SOURCE_SCHEMAS = {
+    "ai_agent_high_risk_owner_review_queue_v1",
+    "ai_agent_action_audit_ledger_v1",
+    "ai_agent_communication_learning_contract_v1",
+    "ai_agent_12_agent_war_room_v1",
+}
+_TRUE_TRUTH_FLAGS = {
+    "p2_409_owner_queue_loaded",
+    "p2_410_audit_ledger_loaded",
+    "communication_contract_loaded",
+    "war_room_loaded",
+    "owner_acceptance_envelope_required",
+    "handoff_protocol_ready",
+    "rag_memory_proposal_ready",
+    "event_bus_no_write_mode",
+    "redacted_evidence_only",
+    "high_critical_human_gate_required",
+    "low_medium_owner_scope_required_before_worker",
+}
+_FALSE_TRUTH_FLAGS = {
+    "owner_response_received",
+    "owner_response_accepted",
+    "owner_response_rejected",
+    "external_response_ingested",
+    "event_bus_publish_enabled",
+    "audit_db_write_enabled",
+    "timeline_write_enabled",
+    "km_write_enabled",
+    "playbook_trust_write_enabled",
+    "gateway_queue_write_enabled",
+    "telegram_send_enabled",
+    "bot_api_call_enabled",
+    "worker_dispatch_enabled",
+    "receipt_production_write_enabled",
+    "production_write_enabled",
+    "secret_read_enabled",
+    "paid_api_call_enabled",
+    "host_write_enabled",
+    "kubectl_action_enabled",
+    "destructive_operation_enabled",
+}
+_ZERO_TRUTH_COUNTS = {
+    "owner_response_received_count_24h",
+    "owner_response_accepted_count_24h",
+    "owner_response_rejected_count_24h",
+    "external_response_ingested_count_24h",
+    "event_bus_publish_count_24h",
+    "audit_db_write_count_24h",
+    "timeline_write_count_24h",
+    "km_write_count_24h",
+    "playbook_trust_write_count_24h",
+    "gateway_queue_write_count_24h",
+    "telegram_send_count_24h",
+    "bot_api_call_count_24h",
+    "worker_dispatch_count_24h",
+    "receipt_production_write_count_24h",
+    "production_write_count_24h",
+    "secret_read_count_24h",
+    "paid_api_call_count_24h",
+    "host_write_count_24h",
+    "kubectl_action_count_24h",
+    "destructive_operation_count_24h",
+}
+_FALSE_LANE_FLAGS = {
+    "response_received",
+    "acceptance_passed",
+    "acceptance_rejected",
+    "runtime_write_allowed",
+    "event_bus_publish_allowed",
+    "telegram_send_allowed",
+    "rag_write_allowed",
+}
+_FALSE_EVENT_FLAGS = {
+    "event_bus_write_allowed",
+    "audit_db_write_allowed",
+    "timeline_write_allowed",
+    "km_write_allowed",
+    "playbook_trust_write_allowed",
+    "gateway_queue_write_allowed",
+    "telegram_send_allowed",
+    "production_write_allowed",
+}
+_FALSE_PROPOSAL_FLAGS = {
+    "km_write_allowed",
+    "playbook_trust_write_allowed",
+    "embedding_write_allowed",
+}
+_TRUE_BOUNDARY_FLAGS = {
+    "committed_snapshot_read_allowed",
+    "owner_acceptance_lane_preview_allowed",
+    "handoff_event_template_preview_allowed",
+    "rag_memory_proposal_preview_allowed",
+    "governance_ui_projection_allowed",
+}
+_FALSE_BOUNDARY_FLAGS = {
+    "event_bus_publish_enabled",
+    "audit_db_write_enabled",
+    "timeline_write_enabled",
+    "km_write_enabled",
+    "playbook_trust_write_enabled",
+    "gateway_queue_write_enabled",
+    "telegram_send_enabled",
+    "bot_api_call_enabled",
+    "worker_dispatch_enabled",
+    "receipt_production_write_enabled",
+    "production_write_enabled",
+    "secret_read_enabled",
+    "paid_api_call_enabled",
+    "host_write_enabled",
+    "kubectl_action_enabled",
+    "destructive_operation_enabled",
+}
+_ZERO_ROLLUP_FIELDS = {
+    "owner_response_received_count",
+    "owner_response_accepted_count",
+    "owner_response_rejected_count",
+    "external_response_ingested_count",
+    "event_bus_publish_count",
+    "audit_db_write_count",
+    "timeline_write_count",
+    "km_write_count",
+    "playbook_trust_write_count",
+    "gateway_queue_write_count",
+    "telegram_send_count",
+    "bot_api_call_count",
+    "worker_dispatch_count",
+    "receipt_production_write_count",
+    "production_write_count",
+    "secret_read_count",
+    "paid_api_call_count",
+    "host_write_count",
+    "kubectl_action_count",
+    "destructive_operation_count",
+}
+_FORBIDDEN_PUBLIC_TERMS = {
+    "批准" + "！",
+    "In app " + "browser",
+    "My request for " + "Codex",
+    "codex_" + "delegation",
+    "source_" + "thread_id",
+    "chain_of_thought",
+    "private reasoning text",
+    "authorization_header",
+    "telegram token value",
+    "raw_payload",
+    "raw prompt",
+    "internal collaboration transcript",
+    "工作視窗",
+    "對話內容",
+}
+
+
+def load_latest_ai_agent_action_owner_acceptance_event_bus(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed P2-411 no-write acceptance event bus snapshot."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(f"no AI Agent action owner acceptance event bus snapshots found in {directory}")
+
+    latest = candidates[-1]
+    with latest.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{latest}: expected JSON object")
+
+    label = str(latest)
+    _require_schema(payload, label)
+    _require_sources(payload, label)
+    _require_truth(payload, label)
+    _require_owner_acceptance_lanes(payload, label)
+    _require_handoff_event_templates(payload, label)
+    _require_rag_memory_proposals(payload, label)
+    _require_verifier_gates(payload, label)
+    _require_activation_boundaries(payload, label)
+    _require_redaction_contract(payload, label)
+    _require_rollups(payload, label)
+    _require_no_forbidden_public_terms(payload, label)
+    return payload
+
+
+def _require_schema(payload: dict[str, Any], label: str) -> None:
+    if payload.get("schema_version") != _SCHEMA_VERSION:
+        raise ValueError(f"{label}: expected schema_version={_SCHEMA_VERSION}")
+    status = payload.get("program_status") or {}
+    expected = {
+        "overall_completion_percent": 100,
+        "current_priority": "P0",
+        "current_task_id": _EXPECTED_CURRENT_TASK,
+        "next_task_id": _EXPECTED_NEXT_TASK,
+        "read_only_mode": True,
+        "runtime_authority": _RUNTIME_AUTHORITY,
+    }
+    mismatches = _mismatches(status, expected)
+    if mismatches:
+        raise ValueError(f"{label}: program_status mismatch: {mismatches}")
+    if not status.get("status_note"):
+        raise ValueError(f"{label}: program_status.status_note is required")
+
+
+def _require_sources(payload: dict[str, Any], label: str) -> None:
+    if not payload.get("source_refs"):
+        raise ValueError(f"{label}: source_refs must not be empty")
+    sources = payload.get("source_readbacks") or []
+    schemas = {item.get("source_schema_version") for item in sources}
+    missing = sorted(_EXPECTED_SOURCE_SCHEMAS - schemas)
+    if missing:
+        raise ValueError(f"{label}: missing source schemas: {missing}")
+    for item in sources:
+        readback_id = item.get("readback_id") or "<missing>"
+        for field in ("source_ref", "endpoint", "owner_agent", "status", "key_readback", "next_action"):
+            if not item.get(field):
+                raise ValueError(f"{label}: source readback {readback_id} missing {field}")
+
+
+def _require_truth(payload: dict[str, Any], label: str) -> None:
+    truth = payload.get("event_bus_truth") or {}
+    missing_true = sorted(flag for flag in _TRUE_TRUTH_FLAGS if truth.get(flag) is not True)
+    if missing_true:
+        raise ValueError(f"{label}: event bus truth flags must remain true: {missing_true}")
+    unsafe_false = sorted(flag for flag in _FALSE_TRUTH_FLAGS if truth.get(flag) is not False)
+    if unsafe_false:
+        raise ValueError(f"{label}: event bus truth flags must remain false: {unsafe_false}")
+    non_zero = sorted(field for field in _ZERO_TRUTH_COUNTS if truth.get(field) != 0)
+    if non_zero:
+        raise ValueError(f"{label}: event bus live counts must remain zero: {non_zero}")
+    if not truth.get("truth_note"):
+        raise ValueError(f"{label}: event_bus_truth.truth_note is required")
+
+
+def _require_owner_acceptance_lanes(payload: dict[str, Any], label: str) -> None:
+    lanes = payload.get("owner_acceptance_lanes") or []
+    if len(lanes) < 1:
+        raise ValueError(f"{label}: owner_acceptance_lanes must not be empty")
+    source_ids = {item.get("readback_id") for item in payload.get("source_readbacks") or []}
+    risk_tiers = {lane.get("risk_tier") for lane in lanes}
+    if not {"medium", "high", "critical"}.issubset(risk_tiers):
+        raise ValueError(f"{label}: acceptance lanes must cover medium, high, and critical")
+    for lane in lanes:
+        lane_id = lane.get("lane_id") or "<missing>"
+        if lane.get("acceptance_status") not in {
+            "blocked_no_external_response",
+            "blocked_missing_fields",
+            "candidate_only_no_write",
+        }:
+            raise ValueError(f"{label}: lane {lane_id}.acceptance_status is invalid")
+        if lane.get("acceptance_decision") != "not_evaluated":
+            raise ValueError(f"{label}: lane {lane_id}.acceptance_decision must remain not_evaluated")
+        unsafe = sorted(flag for flag in _FALSE_LANE_FLAGS if lane.get(flag) is not False)
+        if unsafe:
+            raise ValueError(f"{label}: lane {lane_id} live flags must remain false: {unsafe}")
+        if lane.get("side_effect_count") != 0:
+            raise ValueError(f"{label}: lane {lane_id}.side_effect_count must remain zero")
+        for field in ("source_readback_ids", "required_owner_fields", "required_evidence_refs", "next_gate"):
+            if not lane.get(field):
+                raise ValueError(f"{label}: lane {lane_id} missing {field}")
+        missing_sources = sorted(set(lane.get("source_readback_ids") or []) - source_ids)
+        if missing_sources:
+            raise ValueError(f"{label}: lane {lane_id} references missing source readbacks: {missing_sources}")
+
+
+def _require_handoff_event_templates(payload: dict[str, Any], label: str) -> None:
+    events = payload.get("handoff_event_templates") or []
+    if len(events) < 1:
+        raise ValueError(f"{label}: handoff_event_templates must not be empty")
+    lane_ids = {item.get("lane_id") for item in payload.get("owner_acceptance_lanes") or []}
+    stages = {event.get("event_stage") for event in events}
+    required_stages = {
+        "owner_response_hold",
+        "owner_response_rejection",
+        "candidate_ready_no_write",
+        "handoff_request",
+        "rag_memory_proposal",
+        "no_send_rehearsal",
+    }
+    missing_stages = sorted(required_stages - stages)
+    if missing_stages:
+        raise ValueError(f"{label}: handoff event stages missing: {missing_stages}")
+    for event in events:
+        event_id = event.get("event_id") or "<missing>"
+        unsafe = sorted(flag for flag in _FALSE_EVENT_FLAGS if event.get(flag) is not False)
+        if unsafe:
+            raise ValueError(f"{label}: event {event_id} write/send flags must remain false: {unsafe}")
+        if event.get("side_effect_count") != 0:
+            raise ValueError(f"{label}: event {event_id}.side_effect_count must remain zero")
+        for field in ("source_lane_ids", "required_event_fields", "blocked_writes", "next_gate"):
+            if not event.get(field):
+                raise ValueError(f"{label}: event {event_id} missing {field}")
+        missing_lanes = sorted(set(event.get("source_lane_ids") or []) - lane_ids)
+        if missing_lanes:
+            raise ValueError(f"{label}: event {event_id} references missing lanes: {missing_lanes}")
+
+
+def _require_rag_memory_proposals(payload: dict[str, Any], label: str) -> None:
+    proposals = payload.get("rag_memory_proposals") or []
+    if len(proposals) < 1:
+        raise ValueError(f"{label}: rag_memory_proposals must not be empty")
+    event_ids = {item.get("event_id") for item in payload.get("handoff_event_templates") or []}
+    for proposal in proposals:
+        proposal_id = proposal.get("proposal_id") or "<missing>"
+        if proposal.get("proposal_status") != "proposal_only_no_write":
+            raise ValueError(f"{label}: proposal {proposal_id}.proposal_status must remain proposal_only_no_write")
+        unsafe = sorted(flag for flag in _FALSE_PROPOSAL_FLAGS if proposal.get(flag) is not False)
+        if unsafe:
+            raise ValueError(f"{label}: proposal {proposal_id} write flags must remain false: {unsafe}")
+        if proposal.get("side_effect_count") != 0:
+            raise ValueError(f"{label}: proposal {proposal_id}.side_effect_count must remain zero")
+        for field in ("target_store", "source_event_ids", "required_redaction_checks"):
+            if not proposal.get(field):
+                raise ValueError(f"{label}: proposal {proposal_id} missing {field}")
+        missing_events = sorted(set(proposal.get("source_event_ids") or []) - event_ids)
+        if missing_events:
+            raise ValueError(f"{label}: proposal {proposal_id} references missing events: {missing_events}")
+
+
+def _require_verifier_gates(payload: dict[str, Any], label: str) -> None:
+    gates = payload.get("verifier_gates") or []
+    if len(gates) < 1:
+        raise ValueError(f"{label}: verifier_gates must not be empty")
+    for gate in gates:
+        gate_id = gate.get("gate_id") or "<missing>"
+        if not gate.get("required_checks"):
+            raise ValueError(f"{label}: verifier gate {gate_id} missing required_checks")
+        if not gate.get("failure_if_missing"):
+            raise ValueError(f"{label}: verifier gate {gate_id} missing failure_if_missing")
+        for field in ("live_verifier_allowed", "receipt_write_allowed", "runtime_action_allowed"):
+            if gate.get(field) is not False:
+                raise ValueError(f"{label}: verifier gate {gate_id}.{field} must remain false")
+
+
+def _require_activation_boundaries(payload: dict[str, Any], label: str) -> None:
+    boundaries = payload.get("activation_boundaries") or {}
+    missing = sorted(field for field in _TRUE_BOUNDARY_FLAGS if boundaries.get(field) is not True)
+    if missing:
+        raise ValueError(f"{label}: activation boundaries must remain true: {missing}")
+    unsafe = sorted(field for field in _FALSE_BOUNDARY_FLAGS if boundaries.get(field) is not False)
+    if unsafe:
+        raise ValueError(f"{label}: activation boundaries must remain false: {unsafe}")
+
+
+def _require_redaction_contract(payload: dict[str, Any], label: str) -> None:
+    contract = payload.get("display_redaction_contract") or {}
+    required_false = {
+        "unsafe_payload_display_allowed",
+        "private_reasoning_display_allowed",
+        "secret_value_display_allowed",
+        "raw_prompt_display_allowed",
+        "work_window_transcript_display_allowed",
+    }
+    if contract.get("redaction_required") is not True:
+        raise ValueError(f"{label}: redaction_required must remain true")
+    unsafe = sorted(field for field in required_false if contract.get(field) is not False)
+    if unsafe:
+        raise ValueError(f"{label}: display redaction flags must remain false: {unsafe}")
+    if not contract.get("allowed_display_fields") or not contract.get("blocked_display_fields"):
+        raise ValueError(f"{label}: display redaction contract must list allowed and blocked fields")
+
+
+def _require_rollups(payload: dict[str, Any], label: str) -> None:
+    rollups = payload.get("rollups") or {}
+    lanes = payload.get("owner_acceptance_lanes") or []
+    events = payload.get("handoff_event_templates") or []
+    proposals = payload.get("rag_memory_proposals") or []
+    gates = payload.get("verifier_gates") or []
+    sources = payload.get("source_readbacks") or []
+    expected_counts = {
+        "source_readback_count": len(sources),
+        "owner_acceptance_lane_count": len(lanes),
+        "medium_lane_count": sum(1 for lane in lanes if lane.get("risk_tier") == "medium"),
+        "high_lane_count": sum(1 for lane in lanes if lane.get("risk_tier") == "high"),
+        "critical_lane_count": sum(1 for lane in lanes if lane.get("risk_tier") == "critical"),
+        "handoff_event_template_count": len(events),
+        "rag_memory_proposal_count": len(proposals),
+        "verifier_gate_count": len(gates),
+        "required_owner_field_count": sum(len(lane.get("required_owner_fields") or []) for lane in lanes),
+        "blocked_runtime_action_count": len(
+            {
+                blocked
+                for event in events
+                for blocked in event.get("blocked_writes") or []
+            }
+        ),
+    }
+    mismatches = _mismatches(rollups, expected_counts)
+    if mismatches:
+        raise ValueError(f"{label}: rollup counts mismatch: {mismatches}")
+    non_zero = sorted(field for field in _ZERO_ROLLUP_FIELDS if rollups.get(field) != 0)
+    if non_zero:
+        raise ValueError(f"{label}: live write/send rollups must remain zero: {non_zero}")
+
+
+def _require_no_forbidden_public_terms(payload: dict[str, Any], label: str) -> None:
+    haystack = json.dumps(payload, ensure_ascii=False)
+    hits = sorted(term for term in _FORBIDDEN_PUBLIC_TERMS if term in haystack)
+    if hits:
+        raise ValueError(f"{label}: forbidden public terms detected: {hits}")
+
+
+def _mismatches(source: dict[str, Any], expected: dict[str, Any]) -> dict[str, Any]:
+    return {
+        field: {"expected": value, "actual": source.get(field)}
+        for field, value in expected.items()
+        if source.get(field) != value
+    }