From f1ad1f000c765a3d70d92491a784ee153c96641c Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 30 Jun 2026 01:03:56 +0800 Subject: [PATCH] feat(api): dispatch log writeback receipts --- .gitea/workflows/cd.yaml | 10 + ...lled_writeback_dispatch_operation_type.sql | 40 +++ ...writeback_dispatch_operation_type_down.sql | 40 +++ apps/api/src/api/v1/agents.py | 28 ++ .../ai_agent_autonomous_runtime_control.py | 28 +- ...agent_log_controlled_writeback_dispatch.py | 340 ++++++++++++++++++ ...est_ai_agent_autonomous_runtime_control.py | 15 + ...t_log_controlled_writeback_dispatch_api.py | 130 +++++++ .../tests/test_awooop_truth_chain_service.py | 19 +- .../test_cd_controlled_runtime_profile.py | 14 + 10 files changed, 660 insertions(+), 4 deletions(-) create mode 100644 apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql create mode 100644 apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql create mode 100644 apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py create mode 100644 apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 22f67c0d2..242080904 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -271,6 +271,8 @@ jobs: ;; apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py) ;; + apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py) + ;; apps/api/src/services/ai_agent_autonomous_runtime_control.py) ;; apps/api/src/services/awooop_ansible_audit_service.py) @@ -281,6 +283,10 @@ jobs: ;; apps/api/migrations/adr090e_ansible_learning_writeback_operation_type_down.sql) ;; + apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql) + ;; + apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql) + ;; apps/api/src/services/auto_approve.py) ;; apps/api/src/services/decision_fusion.py) @@ -363,6 +369,8 @@ jobs: ;; apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py) ;; + apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py) + ;; apps/api/tests/test_ai_agent_autonomous_runtime_control.py) ;; apps/api/tests/test_awooop_truth_chain_service.py) @@ -541,6 +549,7 @@ jobs: src/services/ai_agent_log_post_write_verifier_dry_run.py \ src/services/ai_agent_log_controlled_writeback_plan_readback.py \ src/services/ai_agent_log_controlled_writeback_executor_readback.py \ + src/services/ai_agent_log_controlled_writeback_dispatch.py \ src/services/ai_agent_autonomous_runtime_control.py \ src/services/awooop_ansible_audit_service.py \ src/services/awooop_ansible_check_mode_service.py \ @@ -592,6 +601,7 @@ jobs: tests/test_ai_agent_log_post_write_verifier_dry_run_api.py \ tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py \ tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py \ + tests/test_ai_agent_log_controlled_writeback_dispatch_api.py \ tests/test_ai_agent_autonomous_runtime_control.py \ tests/test_awooop_truth_chain_service.py \ tests/test_shadow_auto_approve.py \ diff --git a/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql b/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql new file mode 100644 index 000000000..03af4fadf --- /dev/null +++ b/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql @@ -0,0 +1,40 @@ +-- ADR-090-F: automation_operation_log.operation_type adds LOG metadata dispatch receipts +-- Created: 2026-06-30 Taipei +-- +-- Purpose: +-- P1-LOG-KM-RAG-MCP-PLAYBOOK controlled executor closure. This operation +-- type records metadata-only LOG feedback dispatch receipts after selector, +-- source-of-truth diff, check-mode, rollback, and post-apply verifier gates +-- are ready. +-- +-- Safety: +-- This migration only expands the CHECK allowlist. It does not execute +-- Ansible, write KM/RAG/PlayBook state, call MCP tools, send Telegram +-- messages, read secrets, or alter retained data. + +ALTER TABLE automation_operation_log + DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid; + +ALTER TABLE automation_operation_log + ADD CONSTRAINT automation_operation_log_type_valid CHECK (operation_type IN ( + 'monitor_configured','monitor_removed', + 'alert_fired','alert_suppressed','alert_routed', + 'rule_created','rule_updated','rule_matched','rule_rejected','rule_deprecated', + 'playbook_generated','playbook_updated','playbook_executed', + 'remediation_executed','remediation_verified','remediation_rolled_back', + 'self_correction_attempted', + 'km_created','km_updated','km_linked', + 'asset_discovered','coverage_recalculated', + 'capacity_recommendation','quota_enforced', + 'notification_formatted', + 'ansible_candidate_matched', + 'ansible_check_mode_executed', + 'ansible_apply_executed', + 'ansible_learning_writeback_recorded', + 'ansible_rollback_executed', + 'ansible_execution_skipped', + 'log_controlled_writeback_dispatched' + )); + +COMMENT ON CONSTRAINT automation_operation_log_type_valid ON automation_operation_log IS + 'ADR-090-F: allow LOG metadata-only controlled writeback dispatch receipt rows.'; diff --git a/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql b/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql new file mode 100644 index 000000000..72a8f13be --- /dev/null +++ b/apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql @@ -0,0 +1,40 @@ +-- ADR-090-F rollback: remove LOG metadata dispatch operation type. +-- Only apply after confirming no automation_operation_log rows use the type. + +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM automation_operation_log + WHERE operation_type = 'log_controlled_writeback_dispatched' + LIMIT 1 + ) THEN + RAISE EXCEPTION 'cannot remove log_controlled_writeback_dispatched while rows exist'; + END IF; +END $$; + +ALTER TABLE automation_operation_log + DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid; + +ALTER TABLE automation_operation_log + ADD CONSTRAINT automation_operation_log_type_valid CHECK (operation_type IN ( + 'monitor_configured','monitor_removed', + 'alert_fired','alert_suppressed','alert_routed', + 'rule_created','rule_updated','rule_matched','rule_rejected','rule_deprecated', + 'playbook_generated','playbook_updated','playbook_executed', + 'remediation_executed','remediation_verified','remediation_rolled_back', + 'self_correction_attempted', + 'km_created','km_updated','km_linked', + 'asset_discovered','coverage_recalculated', + 'capacity_recommendation','quota_enforced', + 'notification_formatted', + 'ansible_candidate_matched', + 'ansible_check_mode_executed', + 'ansible_apply_executed', + 'ansible_learning_writeback_recorded', + 'ansible_rollback_executed', + 'ansible_execution_skipped' + )); + +COMMENT ON CONSTRAINT automation_operation_log_type_valid ON automation_operation_log IS + 'ADR-090-E: allow Ansible learning writeback receipt rows for autonomous runtime closure.'; diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 38e0d75cb..5a9a11461 100644 --- a/apps/api/src/api/v1/agents.py +++ b/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], diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 6e2b322f5..18676dbdf 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/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 diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py b/apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py new file mode 100644 index 000000000..ca59a563a --- /dev/null +++ b/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 diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 3ebb0f102..0d5a0612a 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -114,6 +114,7 @@ def test_ai_agent_autonomous_runtime_control_exposes_reports_and_executor_receip "ansible_apply_executed", "incident_evidence.post_execution_state", "knowledge_entries", + "log_controlled_writeback_dispatched", }.issubset(operation_types) assert data["rollups"]["automated_risk_tier_count"] == 3 assert data["rollups"]["report_cadence_enabled_count"] == 3 @@ -128,6 +129,8 @@ def test_ai_agent_autonomous_runtime_control_exposes_reports_and_executor_receip assert data["rollups"]["live_log_controlled_writeback_executor_ready_count"] == 1 assert data["rollups"]["live_log_controlled_writeback_executor_blocker_count"] == 0 assert data["rollups"]["live_log_controlled_writeback_next_action_queue_count"] == 6 + assert data["rollups"]["live_log_controlled_writeback_dispatch_count"] == 0 + assert data["rollups"]["live_log_controlled_writeback_recent_dispatch_count"] == 0 assert data["runtime_receipt_readback"]["learning_loop"]["status"] == "in_progress" assert ( data["runtime_receipt_readback"]["learning_loop"]["rollups"][ @@ -234,6 +237,12 @@ def test_runtime_receipt_readback_summarizes_live_executor_closure_rows(): "total": 1, "recent": 1, }, + { + "operation_type": "log_controlled_writeback_dispatched", + "status": "success", + "total": 6, + "recent": 6, + }, ], operation_latest_rows=[ { @@ -384,6 +393,12 @@ def test_runtime_receipt_readback_summarizes_live_executor_closure_rows(): assert readback["db_read_status"] == "ok" assert readback["writes_on_read"] is False assert readback["ansible_apply_executed"]["total"] == 1 + assert ( + readback["ansible_operations"]["counts"]["log_controlled_writeback_dispatched"][ + "total" + ] + == 6 + ) assert readback["auto_repair_execution_receipt"]["by_status"]["success"] == 1 assert readback["post_apply_verifier"]["by_status"]["success"] == 1 assert readback["km_writeback"]["by_status"]["review"] == 1 diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py new file mode 100644 index 000000000..b092c5b4a --- /dev/null +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.api.v1 import agents +from src.api.v1.agents import router +from src.services import ai_agent_log_controlled_writeback_dispatch as dispatch_module +from src.services.ai_agent_log_controlled_writeback_dispatch import ( + build_ai_agent_log_controlled_writeback_dispatch_receipt, + dispatch_latest_ai_agent_log_controlled_writeback, +) + + +class _FakeResult: + def __init__(self, value: str | None): + self.value = value + + def scalar(self) -> str | None: + return self.value + + +class _FakeDb: + def __init__(self): + self.params: list[dict] = [] + + async def execute(self, _statement, params: dict): + self.params.append(params) + if "input" in params: + return _FakeResult(f"op-{params['dispatch_receipt_id'].rsplit('::', 1)[-1]}") + return _FakeResult("existing-op") + + +class _FakeContext: + def __init__(self, db: _FakeDb): + self.db = db + + async def __aenter__(self) -> _FakeDb: + return self.db + + async def __aexit__(self, _exc_type, _exc, _tb) -> bool: + return False + + +def test_log_controlled_writeback_dispatch_builder_creates_metadata_receipts(): + payload = build_ai_agent_log_controlled_writeback_dispatch_receipt() + + assert payload["schema_version"] == "ai_agent_log_controlled_writeback_dispatch_receipt_v1" + assert payload["status"] == "controlled_writeback_dispatch_ready" + assert payload["active_blockers"] == [] + assert payload["readback"]["operation_type"] == "log_controlled_writeback_dispatched" + assert payload["rollups"]["dispatch_batch_count"] == 6 + assert payload["rollups"]["ready_dispatch_batch_count"] == 6 + assert payload["rollups"]["target_selector_count"] == 12 + assert payload["rollups"]["source_of_truth_diff_count"] == 12 + assert payload["rollups"]["post_apply_verifier_ref_count"] == 12 + assert payload["rollups"]["runtime_dispatch_performed"] is False + + targets = {batch["target"] for batch in payload["dispatch_batches"]} + assert targets == {"km", "rag", "playbook", "mcp", "verifier", "ai_agent"} + for batch in payload["dispatch_batches"]: + assert batch["status"] == "ready_for_ledger_write" + assert batch["executor_route"] == "ai_agent_metadata_writeback_executor" + assert batch["raw_payload_included"] is False + assert batch["check_mode_verified"] is True + assert batch["rollback_ready"] is True + + boundaries = payload["operation_boundaries"] + assert boundaries["metadata_ledger_write_only"] is True + assert boundaries["executor_dispatch_performed"] is False + 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["telegram_send_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 + + +@pytest.mark.asyncio +async def test_log_controlled_writeback_dispatch_writes_idempotent_ledger_rows(monkeypatch): + fake_db = _FakeDb() + monkeypatch.setattr( + dispatch_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + payload = await dispatch_latest_ai_agent_log_controlled_writeback(project_id="awoooi") + + assert payload["status"] == "controlled_writeback_dispatch_ready" + assert payload["rollups"]["runtime_dispatch_performed"] is True + assert payload["rollups"]["dispatch_ledger_row_count"] == 6 + assert payload["rollups"]["inserted_dispatch_ledger_row_count"] == 6 + assert payload["operation_boundaries"]["executor_dispatch_performed"] is True + assert len(fake_db.params) == 6 + + for params in fake_db.params: + assert params["operation_type"] == "log_controlled_writeback_dispatched" + assert params["actor"] == "ai_agent_metadata_writeback_executor" + assert "dispatch_receipt_id" in params + assert "raw_payload_included" in params["input"] + assert "False" not in params["input"] + assert "telegram_send_performed" in params["output"] + + +def test_log_controlled_writeback_dispatch_endpoint_returns_receipt(monkeypatch): + async def fake_dispatch(): + payload = build_ai_agent_log_controlled_writeback_dispatch_receipt() + payload["rollups"]["runtime_dispatch_performed"] = True + return payload + + monkeypatch.setattr( + agents, + "dispatch_latest_ai_agent_log_controlled_writeback", + fake_dispatch, + ) + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.post("/api/v1/agents/agent-log-controlled-writeback-dispatch") + + assert response.status_code == 200 + payload = response.json() + assert payload["schema_version"] == "ai_agent_log_controlled_writeback_dispatch_receipt_v1" + assert payload["status"] == "controlled_writeback_dispatch_ready" + assert payload["rollups"]["runtime_dispatch_performed"] is True diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index ce7ecd3c0..f186e3b11 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -23,10 +23,10 @@ from src.services.awooop_ansible_check_mode_service import ( _record_auto_repair_execution_receipt, _record_learning_writeback_receipt, _send_controlled_apply_telegram_receipt, + backfill_missing_auto_repair_execution_receipts_once, build_ansible_apply_command, build_ansible_check_mode_claim_input, build_ansible_check_mode_command, - backfill_missing_auto_repair_execution_receipts_once, claim_pending_check_modes, detect_ansible_transport_blockers, recent_ansible_transport_blockers, @@ -43,8 +43,8 @@ from src.services.awooop_truth_chain_service import ( _truth_status, build_automation_quality, build_incident_reconciliation, - fetch_truth_chain, fetch_automation_quality_summary, + fetch_truth_chain, summarize_automation_quality_records, ) from src.services.drift_repeat_state import ( @@ -1687,6 +1687,21 @@ def test_ansible_learning_writeback_operation_type_has_schema_migration() -> Non assert "cannot remove ansible_learning_writeback_recorded" in down +def test_log_controlled_writeback_dispatch_operation_type_has_schema_migration() -> None: + migrations_dir = Path(__file__).resolve().parents[1] / "migrations" + migration = Path( + migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type.sql" + ).read_text() + down = Path( + migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type_down.sql" + ).read_text() + + assert "log_controlled_writeback_dispatched" in migration + assert "automation_operation_log_type_valid" in migration + assert "DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid" in migration + assert "cannot remove log_controlled_writeback_dispatched" in down + + def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_not() -> None: live_source = inspect.getsource(run_controlled_apply_for_claim) backfill_source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) diff --git a/ops/runner/test_cd_controlled_runtime_profile.py b/ops/runner/test_cd_controlled_runtime_profile.py index e6a738138..8334065bf 100644 --- a/ops/runner/test_cd_controlled_runtime_profile.py +++ b/ops/runner/test_cd_controlled_runtime_profile.py @@ -170,6 +170,20 @@ def test_ai_log_controlled_writeback_executor_stays_on_controlled_runtime_profil assert source in text +def test_ai_log_controlled_writeback_dispatch_stays_on_controlled_runtime_profile() -> None: + text = _workflow_text() + expected_sources = [ + "apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py)", + "apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py)", + "apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql)", + "apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql)", + "src/services/ai_agent_log_controlled_writeback_dispatch.py", + "tests/test_ai_agent_log_controlled_writeback_dispatch_api.py", + ] + for source in expected_sources: + assert source in text + + def test_awooop_ansible_check_mode_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [