feat(api): dispatch log writeback receipts

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_dispatch import (
+    dispatch_latest_ai_agent_log_controlled_writeback,
+)
 from src.services.ai_agent_log_controlled_writeback_executor_readback import (
     load_latest_ai_agent_log_controlled_writeback_executor_readback,
 )
@@ -2167,6 +2170,31 @@ async def get_agent_log_controlled_writeback_executor_readback() -> dict[str, An
         ) from exc
 
 
+@router.post(
+    "/agent-log-controlled-writeback-dispatch",
+    response_model=dict[str, Any],
+    summary="執行 AI Agent LOG controlled writeback metadata dispatch",
+    description=(
+        "把已通過 selector / diff / check-mode / rollback / post-apply verifier 的 "
+        "LOG feedback batches 寫入 automation_operation_log，形成 metadata-only "
+        "dispatch receipt。此端點不寫 KM、不寫 RAG index、不更新 PlayBook trust、"
+        "不呼叫 MCP tool、不發 Telegram、不觸發 workflow、不保存 raw log payload、"
+        "不讀 secret、不呼叫 GitHub。"
+    ),
+)
+async def post_agent_log_controlled_writeback_dispatch() -> dict[str, Any]:
+    """Persist metadata-only LOG controlled writeback dispatch receipts."""
+    try:
+        payload = await dispatch_latest_ai_agent_log_controlled_writeback()
+        return redact_public_lan_topology(payload)
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("ai_agent_log_controlled_writeback_dispatch_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI Agent LOG controlled writeback dispatch 無效",
+        ) from exc
+
+
 @router.get(
     "/agent-telegram-receipt-approval-package",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -18,6 +18,9 @@ from sqlalchemy import text
 from src.core.config import settings
 from src.core.logging import get_logger
 from src.db.base import get_db_context
+from src.services.ai_agent_log_controlled_writeback_dispatch import (
+    OPERATION_TYPE as LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE,
+)
 from src.services.ai_agent_log_controlled_writeback_executor_readback import (
     load_latest_ai_agent_log_controlled_writeback_executor_readback,
 )
@@ -44,6 +47,7 @@ _EXECUTOR_OPERATION_TYPES = (
     "ansible_learning_writeback_recorded",
     "ansible_rollback_executed",
     "ansible_execution_skipped",
+    LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE,
 )
 
 logger = get_logger(__name__)
@@ -2872,6 +2876,12 @@ def _attach_runtime_receipt_readback(
     log_executor_blockers = log_executor.get("active_blockers")
     if not isinstance(log_executor_blockers, list):
         log_executor_blockers = []
+    operation_counts = (readback.get("ansible_operations") or {}).get("counts")
+    if not isinstance(operation_counts, Mapping):
+        operation_counts = {}
+    log_dispatch_summary = (
+        operation_counts.get(LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE) or {}
+    )
     rollups.update({
         "live_ansible_apply_executed_count": _int_value(
             readback.get("ansible_apply_executed", {}).get("total")
@@ -2976,6 +2986,12 @@ def _attach_runtime_receipt_readback(
         "live_log_controlled_writeback_next_action_queue_count": len(
             log_executor_queue
         ),
+        "live_log_controlled_writeback_dispatch_count": _int_value(
+            log_dispatch_summary.get("total")
+        ),
+        "live_log_controlled_writeback_recent_dispatch_count": _int_value(
+            log_dispatch_summary.get("recent")
+        ),
         "live_agent_decision_wiring_stage_count": _int_value(
             ((readback.get("agent_decision_wiring") or {}).get("rollups") or {}).get(
                 "stage_count"
@@ -3193,6 +3209,12 @@ def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
             "purpose": "把已驗證執行沉澱成 KM / PlayBook trust 候選",
             "writes_runtime_state": True,
         },
+        {
+            "operation_type": LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE,
+            "owner_agent": "ai_agent_metadata_writeback_executor",
+            "purpose": "把 LOG feedback batch 寫入 metadata-only controlled dispatch ledger",
+            "writes_runtime_state": True,
+        },
     ]
     hard_blockers = [
         "secret_token_private_key_cookie_session_auth_header_cleartext",
@@ -3532,7 +3554,8 @@ _RUNTIME_OPERATION_COUNTS_SQL = """
         'ansible_apply_executed',
         'ansible_learning_writeback_recorded',
         'ansible_rollback_executed',
-        'ansible_execution_skipped'
+        'ansible_execution_skipped',
+        'log_controlled_writeback_dispatched'
     )
     GROUP BY operation_type, status
     ORDER BY operation_type, status
@@ -3564,7 +3587,8 @@ _RUNTIME_OPERATION_LATEST_SQL = """
         'ansible_apply_executed',
         'ansible_learning_writeback_recorded',
         'ansible_rollback_executed',
-        'ansible_execution_skipped'
+        'ansible_execution_skipped',
+        'log_controlled_writeback_dispatched'
     )
     ORDER BY created_at DESC
     LIMIT :limit

apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py

@@ -0,0 +1,340 @@
+"""AI Agent LOG controlled writeback dispatch.
+
+Records metadata-only controlled dispatch receipts for LOG feedback batches.
+This route writes only to the internal automation operation ledger; it does
+not write KM/RAG/PlayBook state, call MCP tools, trigger workflows, send
+Telegram messages, persist raw log payloads, or read secrets.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from sqlalchemy import text
+
+from src.db.base import get_db_context
+from src.services.ai_agent_log_controlled_writeback_executor_readback import (
+    load_latest_ai_agent_log_controlled_writeback_executor_readback,
+)
+
+SCHEMA_VERSION = "ai_agent_log_controlled_writeback_dispatch_receipt_v1"
+OPERATION_TYPE = "log_controlled_writeback_dispatched"
+EXECUTOR_ROUTE = "ai_agent_metadata_writeback_executor"
+DEFAULT_PROJECT_ID = "awoooi"
+
+
+async def dispatch_latest_ai_agent_log_controlled_writeback(
+    *,
+    project_id: str = DEFAULT_PROJECT_ID,
+) -> dict[str, Any]:
+    """Persist idempotent metadata-only dispatch receipts for ready LOG batches."""
+
+    executor = load_latest_ai_agent_log_controlled_writeback_executor_readback()
+    receipt = build_ai_agent_log_controlled_writeback_dispatch_receipt(executor)
+    if receipt["active_blockers"]:
+        return receipt
+
+    rows = []
+    async with get_db_context(project_id) as db:
+        for batch in receipt["dispatch_batches"]:
+            inserted = await _insert_dispatch_row(db, project_id=project_id, batch=batch)
+            rows.append(inserted)
+
+    inserted_count = sum(1 for row in rows if row["created"] is True)
+    existing_count = sum(1 for row in rows if row["created"] is False)
+    receipt["dispatch_result"] = {
+        "operation_type": OPERATION_TYPE,
+        "inserted_count": inserted_count,
+        "existing_count": existing_count,
+        "ledger_row_count": len(rows),
+        "rows": rows,
+    }
+    receipt["rollups"]["runtime_dispatch_performed"] = True
+    receipt["rollups"]["dispatch_ledger_row_count"] = len(rows)
+    receipt["rollups"]["inserted_dispatch_ledger_row_count"] = inserted_count
+    receipt["rollups"]["existing_dispatch_ledger_row_count"] = existing_count
+    receipt["operation_boundaries"]["executor_dispatch_performed"] = True
+    return receipt
+
+
+def build_ai_agent_log_controlled_writeback_dispatch_receipt(
+    executor: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build the deterministic dispatch receipt envelope from executor readback."""
+
+    source = executor or load_latest_ai_agent_log_controlled_writeback_executor_readback()
+    batches = [
+        batch
+        for batch in source.get("execution_batches", [])
+        if isinstance(batch, dict)
+    ]
+    dispatch_batches = [_dispatch_batch(batch) for batch in batches]
+    active_blockers = _active_blockers(source=source, dispatch_batches=dispatch_batches)
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
+        "scope": "ai_agent_log_controlled_writeback_dispatch",
+        "status": (
+            "controlled_writeback_dispatch_ready"
+            if not active_blockers
+            else "blocked_waiting_controlled_writeback_dispatch_inputs"
+        ),
+        "readback": {
+            "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-DISPATCH",
+            "workplan_title": "LOG feedback metadata-only controlled dispatch receipt",
+            "source_schema_version": source.get("schema_version"),
+            "source_status": source.get("status"),
+            "executor_route": EXECUTOR_ROUTE,
+            "operation_type": OPERATION_TYPE,
+            "safe_next_step": "post_apply_verify_dispatch_receipts_then_consume_learning_targets",
+        },
+        "dispatch_batches": dispatch_batches,
+        "rollups": {
+            "source_execution_batch_count": len(batches),
+            "ready_source_execution_batch_count": sum(
+                1
+                for batch in batches
+                if batch.get("status") == "ready_for_controlled_executor_dispatch"
+            ),
+            "dispatch_batch_count": len(dispatch_batches),
+            "ready_dispatch_batch_count": sum(
+                1 for batch in dispatch_batches if batch["status"] == "ready_for_ledger_write"
+            ),
+            "target_selector_count": sum(
+                batch["target_selector_count"] for batch in dispatch_batches
+            ),
+            "source_of_truth_diff_count": sum(
+                batch["source_of_truth_diff_count"] for batch in dispatch_batches
+            ),
+            "post_apply_verifier_ref_count": sum(
+                batch["post_apply_verifier_ref_count"] for batch in dispatch_batches
+            ),
+            "controlled_ledger_write_ready": not active_blockers,
+            "runtime_dispatch_performed": False,
+            "dispatch_ledger_row_count": 0,
+            "inserted_dispatch_ledger_row_count": 0,
+            "existing_dispatch_ledger_row_count": 0,
+        },
+        "active_blockers": active_blockers,
+        "operation_boundaries": {
+            "metadata_ledger_write_only": True,
+            "executor_dispatch_performed": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "workflow_trigger_performed": False,
+            "raw_log_payload_persisted": False,
+            "secret_value_collection_allowed": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def _dispatch_batch(batch: dict[str, Any]) -> dict[str, Any]:
+    verifier_refs = (batch.get("post_apply_verifier") or {}).get("verifier_refs")
+    if not isinstance(verifier_refs, list):
+        verifier_refs = []
+    return {
+        "dispatch_receipt_id": f"{OPERATION_TYPE}::{batch.get('target')}",
+        "batch_id": str(batch.get("batch_id") or ""),
+        "target": str(batch.get("target") or ""),
+        "target_surface": str(batch.get("target_surface") or ""),
+        "risk_tier": str(batch.get("risk_tier") or ""),
+        "executor_route": str(batch.get("executor_route") or ""),
+        "status": (
+            "ready_for_ledger_write"
+            if _batch_ready_for_dispatch(batch)
+            else "blocked_waiting_batch_controls"
+        ),
+        "plan_count": int(batch.get("plan_count") or 0),
+        "plan_ids": _strings(batch.get("plan_ids")),
+        "receipt_ids": _strings(batch.get("receipt_ids")),
+        "target_selector_count": int(batch.get("target_selector_count") or 0),
+        "source_of_truth_diff_count": int(batch.get("source_of_truth_diff_count") or 0),
+        "post_apply_verifier_ref_count": len(verifier_refs),
+        "post_apply_verifier_refs": [str(ref) for ref in verifier_refs],
+        "check_mode_verified": (batch.get("check_mode") or {}).get("enabled") is True,
+        "rollback_ready": (batch.get("rollback") or {}).get("required") is True,
+        "raw_payload_included": any(
+            (diff or {}).get("raw_payload_included") is not False
+            for diff in batch.get("source_of_truth_diffs") or []
+        ),
+    }
+
+
+def _batch_ready_for_dispatch(batch: dict[str, Any]) -> bool:
+    if batch.get("status") != "ready_for_controlled_executor_dispatch":
+        return False
+    if batch.get("dispatch_enabled_by_policy") is not True:
+        return False
+    if batch.get("executor_route") != EXECUTOR_ROUTE:
+        return False
+    if (batch.get("check_mode") or {}).get("enabled") is not True:
+        return False
+    if (batch.get("rollback") or {}).get("required") is not True:
+        return False
+    if (batch.get("post_apply_verifier") or {}).get("required") is not True:
+        return False
+    if int(batch.get("target_selector_count") or 0) <= 0:
+        return False
+    if int(batch.get("source_of_truth_diff_count") or 0) <= 0:
+        return False
+    for diff in batch.get("source_of_truth_diffs") or []:
+        if not isinstance(diff, dict):
+            return False
+        if diff.get("raw_payload_included") is not False:
+            return False
+    return True
+
+
+def _active_blockers(
+    *,
+    source: dict[str, Any],
+    dispatch_batches: list[dict[str, Any]],
+) -> list[str]:
+    blockers = []
+    if source.get("status") != "controlled_writeback_executor_ready":
+        blockers.append("controlled_writeback_executor_not_ready")
+    if (source.get("rollups") or {}).get("controlled_executor_dispatch_ready") is not True:
+        blockers.append("controlled_executor_dispatch_not_ready")
+    if not dispatch_batches:
+        blockers.append("dispatch_batches_missing")
+    for batch in dispatch_batches:
+        target = batch["target"] or "unknown"
+        if batch["status"] != "ready_for_ledger_write":
+            blockers.append(f"{target}_dispatch_batch_not_ready")
+        if batch["raw_payload_included"] is not False:
+            blockers.append(f"{target}_raw_payload_included")
+    return _unique(blockers)
+
+
+async def _insert_dispatch_row(
+    db: Any,
+    *,
+    project_id: str,
+    batch: dict[str, Any],
+) -> dict[str, Any]:
+    result = await db.execute(
+        text("""
+            INSERT INTO automation_operation_log (
+                operation_type, actor, status,
+                input, output, dry_run_result, tags
+            )
+            SELECT
+                :operation_type,
+                :actor,
+                'success',
+                CAST(:input AS jsonb),
+                CAST(:output AS jsonb),
+                CAST(:dry_run_result AS jsonb),
+                :tags
+            WHERE NOT EXISTS (
+                SELECT 1
+                FROM automation_operation_log existing
+                WHERE existing.operation_type = :operation_type
+                  AND existing.input ->> 'dispatch_receipt_id' = :dispatch_receipt_id
+            )
+            RETURNING op_id::text
+        """),
+        {
+            "operation_type": OPERATION_TYPE,
+            "actor": EXECUTOR_ROUTE,
+            "dispatch_receipt_id": batch["dispatch_receipt_id"],
+            "input": json.dumps(
+                {
+                    "schema_version": SCHEMA_VERSION,
+                    "project_id": project_id,
+                    "dispatch_receipt_id": batch["dispatch_receipt_id"],
+                    "batch_id": batch["batch_id"],
+                    "target": batch["target"],
+                    "target_surface": batch["target_surface"],
+                    "risk_tier": batch["risk_tier"],
+                    "plan_ids": batch["plan_ids"],
+                    "receipt_ids": batch["receipt_ids"],
+                    "raw_payload_included": False,
+                },
+                ensure_ascii=False,
+            ),
+            "output": json.dumps(
+                {
+                    "executor_route": EXECUTOR_ROUTE,
+                    "ledger_receipt_recorded": True,
+                    "post_apply_verifier_refs": batch["post_apply_verifier_refs"],
+                    "next_action": "consume_metadata_receipt_in_km_rag_playbook_agent_context",
+                    "km_write_performed": False,
+                    "rag_index_write_performed": False,
+                    "playbook_trust_write_performed": False,
+                    "mcp_tool_call_performed": False,
+                    "telegram_send_performed": False,
+                },
+                ensure_ascii=False,
+            ),
+            "dry_run_result": json.dumps(
+                {
+                    "target_selector_count": batch["target_selector_count"],
+                    "source_of_truth_diff_count": batch["source_of_truth_diff_count"],
+                    "check_mode_verified": batch["check_mode_verified"],
+                    "rollback_ready": batch["rollback_ready"],
+                    "post_apply_verifier_ref_count": batch["post_apply_verifier_ref_count"],
+                    "raw_payload_included": False,
+                },
+                ensure_ascii=False,
+            ),
+            "tags": [
+                "ai_agent",
+                "log_feedback",
+                "controlled_writeback",
+                batch["target"],
+            ],
+        },
+    )
+    op_id = result.scalar()
+    if op_id:
+        return {
+            "dispatch_receipt_id": batch["dispatch_receipt_id"],
+            "target": batch["target"],
+            "created": True,
+            "op_id": str(op_id),
+        }
+
+    existing = await db.execute(
+        text("""
+            SELECT op_id::text
+            FROM automation_operation_log
+            WHERE operation_type = :operation_type
+              AND input ->> 'dispatch_receipt_id' = :dispatch_receipt_id
+            ORDER BY created_at DESC
+            LIMIT 1
+        """),
+        {
+            "operation_type": OPERATION_TYPE,
+            "dispatch_receipt_id": batch["dispatch_receipt_id"],
+        },
+    )
+    return {
+        "dispatch_receipt_id": batch["dispatch_receipt_id"],
+        "target": batch["target"],
+        "created": False,
+        "op_id": str(existing.scalar() or ""),
+    }
+
+
+def _strings(value: Any) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    return [str(item) for item in value]
+
+
+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