feat(ai): enforce single-writer automation router

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -1075,6 +1075,18 @@ def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe():
     )
     assert "automation_run_id" in _RUNTIME_OPERATION_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_OPERATION_LATEST_DIRECT_SQL
+    for sql in (
+        _RUNTIME_OPERATION_LATEST_SQL,
+        _RUNTIME_OPERATION_LATEST_DIRECT_SQL,
+        runtime_control_module._RUNTIME_OPERATION_CHAIN_SQL,
+        runtime_control_module._RUNTIME_OPERATION_CHAIN_DIRECT_SQL,
+    ):
+        assert "single_writer_executor" in sql
+        assert "candidate_idempotency_key" in sql
+        assert "apply_idempotency_key" in sql
+        assert "source_truth_diff_required" in sql
+        assert "target_selector_present" in sql
+        assert "router_source_sha" in sql
     assert "automation_run_id" in runtime_control_module._RUNTIME_AUTO_REPAIR_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_VERIFIER_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_KM_LATEST_SQL
@@ -1102,6 +1114,99 @@ def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe():
     assert "channel_chat_id LIKE 'alert-group:%'" in _RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL
 
 
+def test_autonomous_single_writer_requires_same_run_runtime_and_deployed_source() -> None:
+    run_id = "11111111-1111-4111-8111-111111111111"
+    rows = [
+        {
+            "op_id": run_id,
+            "operation_type": "ansible_candidate_matched",
+            "status": "dry_run",
+            "automation_run_id": run_id,
+            "decision_path": "repair_candidate_controlled_queue",
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "candidate_idempotency_key": "ansible-candidate:awoooi:101",
+            "target_selector_present": True,
+            "source_truth_diff_required": "true",
+            "router_source_sha": "source123",
+        },
+        {
+            "op_id": "22222222-2222-4222-8222-222222222222",
+            "operation_type": "ansible_check_mode_executed",
+            "status": "success",
+            "automation_run_id": run_id,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+        },
+        {
+            "op_id": "33333333-3333-4333-8333-333333333333",
+            "operation_type": "ansible_apply_executed",
+            "status": "success",
+            "automation_run_id": run_id,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "apply_idempotency_key": "ansible-apply:run:catalog",
+        },
+    ]
+    runtime = runtime_control_module._build_autonomous_single_writer_runtime_readback(
+        operation_latest_rows=rows,
+        loop_ledger={
+            "automation_run_id": run_id,
+            "same_run_correlation": True,
+        },
+    )
+    assert runtime["runtime_evidence_ready"] is True
+    assert all(runtime["controls"].values())
+
+    payload = {
+        "strict_runtime_completion": {
+            "automation_run_id": run_id,
+            "closed": True,
+        },
+        "rollups": {},
+    }
+    source_controls = {
+        "decision_manager_queue_only": True,
+        "webhook_router_queue_only": True,
+        "failure_watcher_queue_only": True,
+        "auto_approved_direct_execution_blocked": True,
+        "candidate_idempotency_contract_verified": True,
+        "apply_idempotency_contract_verified": True,
+        "critical_break_glass_contract_verified": True,
+    }
+    runtime_control_module._attach_autonomous_single_writer_readback(
+        payload,
+        readback={"autonomous_single_writer_runtime": runtime},
+        boundary_readback={
+            "deployed_source_sha": "source123",
+            "autonomous_single_writer_source_boundary": {
+                "receipt_ref": "configmap:awoooi-prod/boundary",
+                "verifier": "cd_tests_and_production_source_sha_v1",
+                "controls": source_controls,
+                "active_blockers": [],
+            },
+        },
+    )
+
+    assert payload["autonomous_single_writer"]["closed"] is True
+    assert payload["autonomous_single_writer"]["active_blockers"] == []
+    assert payload["rollups"]["autonomous_single_writer_closed_count"] == 1
+
+    payload["strict_runtime_completion"]["automation_run_id"] = "other-run"
+    runtime_control_module._attach_autonomous_single_writer_readback(
+        payload,
+        readback={"autonomous_single_writer_runtime": runtime},
+        boundary_readback={
+            "deployed_source_sha": "source123",
+            "autonomous_single_writer_source_boundary": {
+                "controls": source_controls,
+                "active_blockers": [],
+            },
+        },
+    )
+    assert payload["autonomous_single_writer"]["closed"] is False
+    assert "strict_runtime_same_run_closed_not_verified" in payload[
+        "autonomous_single_writer"
+    ]["active_blockers"]
+
+
 def test_ai_agent_autonomous_runtime_control_uses_current_owner_directive():
     data = build_ai_agent_autonomous_runtime_control()
 

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -150,6 +150,30 @@ def _autonomous_runtime_control_trust_boundary_closed() -> dict:
     return payload
 
 
+def _autonomous_runtime_control_single_writer_closed() -> dict:
+    payload = _autonomous_runtime_control_trust_boundary_closed()
+    run_id = payload["strict_runtime_completion"]["automation_run_id"]
+    payload["autonomous_single_writer"] = {
+        "schema_version": "awoooi_autonomous_single_writer_readback_v1",
+        "closed": True,
+        "automation_run_id": run_id,
+        "router_source_sha": "source-p0-002",
+        "candidate_op_id": "candidate-p0-002",
+        "check_mode_op_id": "check-p0-002",
+        "apply_op_id": "apply-p0-002",
+        "source_receipt_ref": (
+            "configmap:awoooi-prod/awoooi-executor-boundary-verification"
+        ),
+        "source_verifier": "cd_tests_and_production_source_sha_v1",
+        "controls": {
+            control_id: True
+            for control_id in priority_service._AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS
+        },
+        "active_blockers": [],
+    }
+    return payload
+
+
 async def _autonomous_runtime_control_closed_async() -> dict:
     return _autonomous_runtime_control_closed()
 
@@ -2031,6 +2055,35 @@ def test_ai_automation_trust_boundary_requires_db_identity_and_budget_receipt()
     assert program["summary"]["completion_percent"] == 5
 
 
+def test_ai_automation_single_writer_runtime_receipt_advances_order() -> None:
+    payload = load_latest_awoooi_priority_work_order_readback()
+
+    apply_ai_automation_live_closure_readbacks(
+        payload,
+        autonomous_runtime_control=(
+            _autonomous_runtime_control_single_writer_closed()
+        ),
+    )
+
+    program = payload["ai_automation_program_ledger"]
+    items = {item["id"]: item for item in program["work_items"]}
+    assert items["AIA-P0-001"]["status"] == "done"
+    assert items["AIA-P0-011"]["status"] == "done"
+    single_writer = items["AIA-P0-002"]
+    assert single_writer["status"] == "done"
+    assert single_writer["runtime_progress"]["runtime_closed"] is True
+    assert single_writer["runtime_progress"]["completion_percent"] == 100
+    assert single_writer["completion_evidence"]["automation_run_id"] == (
+        "run-p0-001-closed"
+    )
+    assert single_writer["completion_evidence"]["single_writer_executor"] == (
+        "awoooi-ansible-executor-broker"
+    )
+    assert program["summary"]["next_work_item_id"] == "AIA-P0-003"
+    assert program["summary"]["done_count"] == 3
+    assert program["summary"]["completion_percent"] == 15
+
+
 def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot(
     monkeypatch: pytest.MonkeyPatch,
 ):

apps/api/tests/test_awooop_truth_chain_service.py

@@ -9,6 +9,9 @@ from types import SimpleNamespace
 import pytest
 
 from src.core.config import settings
+from src.jobs.awooop_ansible_candidate_backfill_job import (
+    enqueue_ai_decision_ansible_candidate,
+)
 from src.services.awooop_ansible_audit_service import (
     build_ansible_decision_audit_payload,
     build_ansible_truth,
@@ -1385,7 +1388,10 @@ def test_ansible_decision_audit_payload_marks_repair_candidate_queue_claimable()
     assert payload["input"]["project_id"] == "awoooi"
 
 
-def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags() -> None:
+def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv("AWOOOI_BUILD_COMMIT_SHA", "source-sha-123")
     incident = SimpleNamespace(
         incident_id="INC-MOMO",
         project_id="awoooi",
@@ -1417,6 +1423,7 @@ def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags() -> Non
     assert candidate["auto_apply_enabled"] is True
     assert candidate["approval_required"] is False
     assert candidate["risk_level"] == "low"
+    assert payload["input"]["router_source_sha"] == "source-sha-123"
 
 
 def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None:
@@ -1424,6 +1431,11 @@ def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None:
         "incident_id": "INC-MOMO",
         "automation_run_id": "untrusted-upstream-run-id",
         "executor": "ansible",
+        "decision_path": "repair_candidate_controlled_queue",
+        "idempotency_key": "ansible-candidate:awoooi:INC-MOMO",
+        "router_source_sha": "source-sha-123",
+        "target_selector": {"catalog_ids": ["ansible:188-ai-web"]},
+        "source_truth_diff": {"required_before_apply": True},
         "executor_candidates": [
             {
                 "catalog_id": "ansible:188-ai-web",
@@ -1448,6 +1460,19 @@ def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None:
     assert claim["approval_required_before_apply"] is False
     assert claim["controlled_apply_allowed"] is True
     assert claim["controlled_apply_blocker"] is None
+    assert claim["single_writer_executor"] == "awoooi-ansible-executor-broker"
+    assert claim["candidate_idempotency_key"] == (
+        "ansible-candidate:awoooi:INC-MOMO"
+    )
+    assert claim["apply_idempotency_key"] == (
+        "ansible-apply:00000000-0000-0000-0000-000000000001:"
+        "ansible:188-ai-web"
+    )
+    assert claim["router_source_sha"] == "source-sha-123"
+    assert claim["target_selector"] == {
+        "catalog_ids": ["ansible:188-ai-web"]
+    }
+    assert claim["source_truth_diff"]["required_before_apply"] is True
     assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
     assert claim["apply_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
     assert claim["source_candidate_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
@@ -2231,6 +2256,165 @@ def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> N
     assert '"automation_run_id": str(' in apply_source
 
 
+@pytest.mark.asyncio
+async def test_ai_decision_handoff_queues_evidence_backed_single_writer_candidate() -> None:
+    incident = SimpleNamespace(
+        incident_id="INC-AIA-P0-002",
+        project_id="awoooi",
+        severity=SimpleNamespace(value="low"),
+        alertname="MomoPostgresBackupFailed",
+        alert_category="backup_failure",
+        notification_type="TYPE-3",
+        affected_services=["momo-postgres"],
+        signals=[
+            SimpleNamespace(
+                alert_name="MomoPostgresBackupFailed",
+                labels={"alertname": "MomoPostgresBackupFailed", "instance": "188"},
+                annotations={},
+            )
+        ],
+    )
+    recorded: list[dict] = []
+
+    async def collect(**_kwargs):
+        return SimpleNamespace(snapshot_id="evidence-1")
+
+    async def verify(**_kwargs):
+        return True
+
+    async def record(**kwargs):
+        recorded.append(kwargs)
+        return True
+
+    handoff = await enqueue_ai_decision_ansible_candidate(
+        incident=incident,
+        proposal_data={
+            "source": "decision_manager",
+            "risk_level": "low",
+            "action": "systemctl restart momo-backup",
+        },
+        recorder=record,
+        evidence_collector=collect,
+        evidence_verifier=verify,
+    )
+
+    assert handoff["status"] == "controlled_check_mode_queued"
+    assert handoff["queued"] is True
+    assert handoff["side_effect_performed"] is False
+    assert handoff["single_writer_executor"] == "awoooi-ansible-executor-broker"
+    assert handoff["source_truth_diff"]["required_before_apply"] is True
+    assert handoff["risk_policy_decision"]["risk_level"] == "low"
+    assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue"
+
+
+@pytest.mark.asyncio
+async def test_ai_decision_handoff_blocks_critical_before_evidence_or_queue() -> None:
+    calls = 0
+
+    async def must_not_run(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return True
+
+    handoff = await enqueue_ai_decision_ansible_candidate(
+        incident=SimpleNamespace(incident_id="INC-CRITICAL"),
+        proposal_data={"risk_level": "critical", "action": "restart"},
+        recorder=must_not_run,
+        evidence_collector=must_not_run,
+        evidence_verifier=must_not_run,
+    )
+
+    assert handoff["status"] == "critical_break_glass_required"
+    assert handoff["queued"] is False
+    assert handoff["side_effect_performed"] is False
+    assert calls == 0
+
+
+def test_controlled_apply_uses_durable_idempotency_guard() -> None:
+    source = inspect.getsource(run_controlled_apply_for_claim)
+    broker_source = inspect.getsource(run_pending_check_modes_once)
+
+    assert "pg_advisory_xact_lock" in source
+    assert "apply_idempotency_key" in source
+    assert "ansible_controlled_apply_duplicate_suppressed" in source
+    assert "apply_duplicate_suppressed" in broker_source
+
+
+@pytest.mark.asyncio
+async def test_controlled_apply_duplicate_never_runs_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.services import awooop_ansible_check_mode_service as service
+
+    class ExistingResult:
+        def mappings(self):
+            return self
+
+        def first(self):
+            return {
+                "op_id": "00000000-0000-0000-0000-000000000103",
+                "status": "success",
+            }
+
+    class FakeDb:
+        async def execute(self, *_args, **_kwargs):
+            return ExistingResult()
+
+    class FakeContext:
+        async def __aenter__(self):
+            return FakeDb()
+
+        async def __aexit__(self, *_args):
+            return False
+
+    subprocess_started = False
+
+    async def must_not_run(*_args, **_kwargs):
+        nonlocal subprocess_started
+        subprocess_started = True
+        raise AssertionError("duplicate apply must not execute")
+
+    monkeypatch.setattr(
+        service,
+        "get_db_context",
+        lambda _project_id: FakeContext(),
+    )
+    monkeypatch.setattr(
+        service,
+        "_execution_capability_timeout_seconds",
+        lambda *_args, **_kwargs: 30,
+    )
+    monkeypatch.setattr(service, "_run_ansible_command", must_not_run)
+
+    claim = AnsibleCheckModeClaim(
+        op_id="00000000-0000-0000-0000-000000000102",
+        source_candidate_op_id="00000000-0000-0000-0000-000000000101",
+        incident_id="INC-IDEMPOTENT",
+        catalog_id="ansible:188-momo-backup-user",
+        playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
+        apply_playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
+        inventory_hosts=("host_188",),
+        risk_level="low",
+        input_payload={
+            "controlled_apply_allowed": True,
+            "apply_idempotency_key": (
+                "ansible-apply:00000000-0000-0000-0000-000000000101:"
+                "ansible:188-momo-backup-user"
+            ),
+        },
+    )
+
+    result = await service.run_controlled_apply_for_claim(
+        claim,
+        timeout_seconds=30,
+    )
+
+    assert result is None
+    assert subprocess_started is False
+    assert claim.input_payload["apply_duplicate_suppressed"] is True
+    assert claim.input_payload["existing_apply_status"] == "success"
+
+
 def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> None:
     source = inspect.getsource(_run_ansible_command)
 

apps/api/tests/test_cs1_auto_execute.py

@@ -1,43 +1,35 @@
-# apps/api/tests/test_cs1_auto_execute.py
-# 2026-04-27 ogt + Claude Sonnet 4.6 — CS1 LLM 高信心度自動執行邏輯單元測試
-"""
-測試覆蓋：
-1. confidence=0.90 + low risk + kubectl 有值 → execute_approved_action 被呼叫
-2. confidence=0.70 → 不執行（低信心度）
-3. confidence=0.85 + CRITICAL → 不執行
-4. confidence=0.90 + DESTRUCTIVE_PATTERN → 不執行
-5. confidence=0.90 + NO_ACTION → 不執行
-6. confidence=0.90 執行失敗（exception）→ 降級 PENDING 不 crash
-
-測試分類：unit（mock ApprovalExecutionService / service，無 DB / Redis 依賴）
-"""
+"""Webhook AI decisions must route through the single-writer broker."""
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, MagicMock, patch
+import inspect
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 
-from src.models.ai import AIBlastRadius, AIDataImpact, AIRiskLevel, OpenClawDecision, SuggestedAction
+from src.api.v1 import webhooks
+from src.models.ai import (
+    AIBlastRadius,
+    AIDataImpact,
+    OpenClawDecision,
+    SuggestedAction,
+)
 from src.models.approval import RiskLevel
-
-
-# =============================================================================
-# Helpers
-# =============================================================================
+from src.services.approval_execution import ApprovalExecutionService
 
 
 def _make_analysis(
+    *,
     confidence: float = 0.90,
     kubectl_command: str = "kubectl rollout restart deployment/api -n prod",
-    risk_level_str: str = "low",
     suggested_action: SuggestedAction = SuggestedAction.RESTART_DEPLOYMENT,
 ) -> OpenClawDecision:
     return OpenClawDecision(
         action_title="Restart deployment",
         kubectl_command=kubectl_command,
-        description="Auto restart",
-        risk_level=risk_level_str,
+        description="Controlled restart candidate",
+        risk_level="low",
         suggested_action=suggested_action,
         confidence=confidence,
         blast_radius=AIBlastRadius(
@@ -53,176 +45,124 @@ def _make_analysis(
     )
 
 
-def _run_cs1_block(
-    analysis_result: OpenClawDecision | None,
-    risk_level: RiskLevel,
-    exec_side_effect=None,
-) -> tuple[MagicMock, MagicMock]:
-    """
-    從 webhooks.py CS1 auto-execute 邏輯提取的同等邏輯，
-    直接呼叫，驗證 execute_approved_action 的呼叫情況。
-
-    回傳 (mock_executor_class, mock_execute_method)
-    """
+def _can_route(analysis: OpenClawDecision, risk_level: RiskLevel) -> bool:
     from src.services.action_parser import is_safe_kubectl_action
 
-    mock_exec_instance = MagicMock()
-    if exec_side_effect is not None:
-        mock_exec_instance.execute_approved_action = AsyncMock(side_effect=exec_side_effect)
-    else:
-        mock_exec_instance.execute_approved_action = AsyncMock(return_value=True)
+    non_actions = {"NO_ACTION", "INVESTIGATE", "OBSERVE"}
+    action_name = analysis.suggested_action.value
+    command = (analysis.kubectl_command or "").strip()
+    return (
+        bool(command)
+        and analysis.confidence >= 0.85
+        and risk_level != RiskLevel.CRITICAL
+        and action_name not in non_actions
+        and is_safe_kubectl_action(command)
+    )
 
-    mock_executor_cls = MagicMock(return_value=mock_exec_instance)
 
-    with (
-        patch("src.services.approval_execution.ApprovalExecutionService", mock_executor_cls),
-        patch("src.models.approval.ApprovalRequest", MagicMock()),
-        patch("src.models.approval.ApprovalStatus", MagicMock()),
+@pytest.mark.parametrize(
+    ("analysis", "risk_level", "expected"),
+    [
+        (_make_analysis(), RiskLevel.LOW, True),
+        (_make_analysis(confidence=0.70), RiskLevel.LOW, False),
+        (_make_analysis(confidence=0.85), RiskLevel.MEDIUM, True),
+        (_make_analysis(), RiskLevel.CRITICAL, False),
+        (
+            _make_analysis(
+                kubectl_command="kubectl delete pods --all -n prod",
+            ),
+            RiskLevel.LOW,
+            False,
+        ),
+        (
+            _make_analysis(suggested_action=SuggestedAction.NO_ACTION),
+            RiskLevel.LOW,
+            False,
+        ),
+    ],
+)
+def test_cs1_controlled_queue_eligibility(
+    analysis: OpenClawDecision,
+    risk_level: RiskLevel,
+    expected: bool,
+) -> None:
+    assert _can_route(analysis, risk_level) is expected
+
+
+def test_active_webhook_paths_do_not_call_direct_executor() -> None:
+    receive_source = inspect.getsource(webhooks.receive_alert)
+    alertmanager_source = inspect.getsource(webhooks._process_new_alert_background)
+    router_source = inspect.getsource(webhooks._try_auto_repair_background)
+    legacy_source = inspect.getsource(
+        webhooks._legacy_try_auto_repair_background_disabled
+    )
+
+    for source in (receive_source, alertmanager_source, router_source):
+        assert "execute_approved_action" not in source
+        assert "ApprovalExecutionService" not in source
+        assert "execute_auto_repair(" not in source
+    assert "_try_auto_repair_background" in receive_source
+    assert "_try_auto_repair_background" in alertmanager_source
+    assert "enqueue_ai_decision_ansible_candidate" in router_source
+    assert "webhook_direct_auto_repair_superseded" in legacy_source
+
+
+@pytest.mark.asyncio
+async def test_webhook_router_writes_queue_receipt_without_side_effect(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    incident = SimpleNamespace(project_id="awoooi")
+    incident_service = SimpleNamespace(
+        get_from_working_memory=AsyncMock(return_value=incident)
+    )
+    append = AsyncMock()
+
+    async def queue(**_kwargs):
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "controlled_check_mode_queued",
+            "automation_run_id": "00000000-0000-0000-0000-000000000101",
+            "queued": True,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": [],
+        }
+
+    monkeypatch.setattr(webhooks, "get_incident_service", lambda: incident_service)
+    monkeypatch.setattr(
+        "src.jobs.awooop_ansible_candidate_backfill_job.enqueue_ai_decision_ansible_candidate",
+        queue,
+    )
+    monkeypatch.setattr(
+        "src.repositories.alert_operation_log_repository.get_alert_operation_log_repository",
+        lambda: SimpleNamespace(append=append),
+    )
+
+    handoff = await webhooks._try_auto_repair_background(
+        incident_id="INC-QUEUE",
+        approval_id="00000000-0000-0000-0000-000000000201",
+        alert_type="MomoPostgresBackupFailed",
+        target_resource="momo-postgres",
+        namespace="awoooi-prod",
+        risk_level="low",
+    )
+
+    assert handoff["queued"] is True
+    assert handoff["side_effect_performed"] is False
+    append.assert_awaited_once()
+    assert append.await_args.kwargs["context"]["side_effect_performed"] is False
+
+
+@pytest.mark.asyncio
+async def test_approval_executor_rejects_auto_approved_direct_caller() -> None:
+    approval = SimpleNamespace(
+        id="00000000-0000-0000-0000-000000000301",
+        incident_id="INC-DIRECT-BLOCK",
+        requested_by="auto_approve_rule_engine",
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="auto_approved_direct_execution_disabled_use_single_writer_broker",
     ):
-        # Replicate the exact condition logic from webhooks.py CS1 block
-        _non_destructive_actions = {"NO_ACTION", "INVESTIGATE", "OBSERVE"}
-        _sa_val = (
-            analysis_result.suggested_action.value
-            if analysis_result and hasattr(analysis_result.suggested_action, "value")
-            else str(getattr(analysis_result, "suggested_action", ""))
-        )
-
-        if analysis_result:
-            _cs1_kubectl = analysis_result.kubectl_command.strip() if analysis_result.kubectl_command else ""
-            _cs1_can_auto = (
-                bool(_cs1_kubectl)
-                and analysis_result.confidence >= 0.85
-                and risk_level != RiskLevel.CRITICAL
-                and _sa_val not in _non_destructive_actions
-                and is_safe_kubectl_action(_cs1_kubectl)
-            )
-            if _cs1_can_auto:
-                import asyncio
-
-                from src.models.approval import ApprovalRequest, ApprovalStatus
-                from src.services.approval_execution import ApprovalExecutionService
-
-                _cs1_auto_approval = MagicMock()
-                _cs1_executor = ApprovalExecutionService()
-                asyncio.run(_cs1_executor.execute_approved_action(_cs1_auto_approval))
-
-    return mock_executor_cls, mock_exec_instance
-
-
-# =============================================================================
-# Tests
-# =============================================================================
-
-
-class TestCS1AutoExecuteConditions:
-    """測試 CS1 自動執行的觸發條件"""
-
-    def test_high_confidence_low_risk_executes(self):
-        """confidence=0.90 + LOW risk + kubectl 有值 → execute 被呼叫"""
-        analysis = _make_analysis(confidence=0.90)
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_called_once()
-
-    def test_low_confidence_does_not_execute(self):
-        """confidence=0.70 → 不執行"""
-        analysis = _make_analysis(confidence=0.70)
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_not_called()
-
-    def test_boundary_confidence_085_executes(self):
-        """confidence=0.85 剛好等於門檻 → 執行"""
-        analysis = _make_analysis(confidence=0.85)
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_called_once()
-
-    def test_critical_risk_does_not_execute(self):
-        """confidence=0.90 + CRITICAL → 不執行"""
-        analysis = _make_analysis(confidence=0.90, risk_level_str="critical")
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.CRITICAL)
-        mock_exec.execute_approved_action.assert_not_called()
-
-    def test_destructive_pattern_does_not_execute(self):
-        """kubectl 含 destructive pattern → 不執行"""
-        from src.services.auto_approve import _DESTRUCTIVE_PATTERNS
-        # 取第一個 pattern 構造一個含危險詞的指令
-        bad_pattern = _DESTRUCTIVE_PATTERNS[0]
-        analysis = _make_analysis(
-            confidence=0.90,
-            kubectl_command=f"kubectl {bad_pattern} deployment/api -n prod",
-        )
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_not_called()
-
-    def test_single_delete_pod_executes(self):
-        """單一 Pod delete 是可恢復操作，parser 不應誤殺"""
-        analysis = _make_analysis(
-            confidence=0.90,
-            kubectl_command="kubectl delete pod api-xxx-yyy -n prod",
-        )
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_called_once()
-
-    def test_no_action_does_not_execute(self):
-        """suggested_action=NO_ACTION → 不執行"""
-        analysis = _make_analysis(
-            confidence=0.90,
-            suggested_action=SuggestedAction.NO_ACTION,
-        )
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_not_called()
-
-    def test_investigate_does_not_execute(self):
-        """suggested_action=INVESTIGATE → 不執行"""
-        analysis = _make_analysis(
-            confidence=0.90,
-            suggested_action=SuggestedAction.INVESTIGATE,
-        )
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_not_called()
-
-
-class TestCS1AutoExecuteFailureDegradation:
-    """測試執行失敗時的降級行為"""
-
-    def test_execution_exception_does_not_crash(self):
-        """execute_approved_action 拋 exception → 捕捉後繼續，不 crash"""
-        analysis = _make_analysis(confidence=0.90)
-
-        # 直接測試條件邏輯，確保例外被吞掉
-        from src.services.action_parser import is_safe_kubectl_action
-
-        _non_destructive_actions = {"NO_ACTION", "INVESTIGATE", "OBSERVE"}
-        _sa_val = analysis.suggested_action.value
-        _cs1_kubectl = analysis.kubectl_command.strip()
-        _cs1_can_auto = (
-            bool(_cs1_kubectl)
-            and analysis.confidence >= 0.85
-            and RiskLevel.LOW != RiskLevel.CRITICAL
-            and _sa_val not in _non_destructive_actions
-            and is_safe_kubectl_action(_cs1_kubectl)
-        )
-        assert _cs1_can_auto, "前置條件必須為 True 才能測試降級"
-
-        raised = False
-        try:
-            import asyncio
-
-            async def _simulate():
-                # 模擬整個 if _cs1_can_auto: try/except 區塊
-                if _cs1_can_auto:
-                    try:
-                        raise RuntimeError("executor exploded")
-                    except Exception:
-                        pass  # 降級：維持 PENDING
-
-            asyncio.run(_simulate())
-        except Exception:
-            raised = True
-
-        assert not raised, "例外應被 CS1 try/except 吞掉，不應傳播"
-
-    def test_empty_kubectl_does_not_execute(self):
-        """kubectl_command 為空字串 → 不執行"""
-        analysis = _make_analysis(confidence=0.90, kubectl_command="")
-        _, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
-        mock_exec.execute_approved_action.assert_not_called()
+        await ApprovalExecutionService().execute_approved_action(approval)

apps/api/tests/test_cs3_auto_execute.py

@@ -1,127 +1,69 @@
-# apps/api/tests/test_cs3_auto_execute.py
-# 2026-04-27 ogt + Claude Sonnet 4.6 — CS3 alertmanager AI path 高信心自動執行單元測試
-"""
-測試覆蓋：
-1. confidence=0.90 + MEDIUM risk + kubectl 有值 → can_auto=True
-2. confidence=0.70 → blocked
-3. CRITICAL risk → blocked
-4. kubectl="" → blocked
-5. NO_ACTION title → blocked
-6. destructive kubectl (delete) → blocked
-7. destructive --force pattern → isinstance check
-8. execute_approved_action 被呼叫
-9. execute 拋例外不向上傳播
-"""
+"""Alertmanager high-confidence routing remains guarded and queue-only."""
+
 from __future__ import annotations
 
+import inspect
+from types import SimpleNamespace
+
 import pytest
-from unittest.mock import AsyncMock, MagicMock, patch
+
+from src.api.v1 import webhooks
+from src.models.approval import RiskLevel
 
 
-def _make_analysis(
-    confidence: float = 0.9,
+def _analysis(
+    *,
+    confidence: float = 0.90,
     action_title: str = "restart pod",
     kubectl: str = "kubectl rollout restart deployment/foo",
-):
-    a = MagicMock()
-    a.confidence = confidence
-    a.action_title = action_title
-    a.kubectl_command = kubectl
-    a.description = "test desc"
-    a.affected_services = []
-    a.primary_responsibility = "COLLAB"
-    return a
-
-
-def _can_auto(analysis, risk_level, patterns):
-    from src.models.approval import RiskLevel
-    from src.services.action_parser import is_safe_kubectl_action
-    kubectl = (analysis.kubectl_command or "").strip()
-    return (
-        bool(kubectl)
-        and analysis.confidence >= 0.85
-        and risk_level != RiskLevel.CRITICAL
-        and "NO_ACTION" not in (analysis.action_title or "")
-        and is_safe_kubectl_action(kubectl)
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        confidence=confidence,
+        action_title=action_title,
+        kubectl_command=kubectl,
     )
 
 
-@pytest.fixture(scope="module")
-def patterns():
-    from src.services.auto_approve import _DESTRUCTIVE_PATTERNS
-    return _DESTRUCTIVE_PATTERNS
+def _can_route(analysis: SimpleNamespace, risk_level: RiskLevel) -> bool:
+    from src.services.action_parser import is_safe_kubectl_action
+
+    command = str(analysis.kubectl_command or "").strip()
+    return (
+        bool(command)
+        and analysis.confidence >= 0.85
+        and risk_level != RiskLevel.CRITICAL
+        and "NO_ACTION" not in str(analysis.action_title or "")
+        and is_safe_kubectl_action(command)
+    )
 
 
-class TestCS3AutoExecute:
+@pytest.mark.parametrize(
+    ("analysis", "risk_level", "expected"),
+    [
+        (_analysis(), RiskLevel.MEDIUM, True),
+        (_analysis(confidence=0.70), RiskLevel.MEDIUM, False),
+        (_analysis(), RiskLevel.CRITICAL, False),
+        (_analysis(kubectl=""), RiskLevel.MEDIUM, False),
+        (_analysis(action_title="NO_ACTION: no fix"), RiskLevel.MEDIUM, False),
+        (
+            _analysis(kubectl="kubectl delete pods --all -n prod"),
+            RiskLevel.MEDIUM,
+            False,
+        ),
+    ],
+)
+def test_cs3_controlled_queue_eligibility(
+    analysis: SimpleNamespace,
+    risk_level: RiskLevel,
+    expected: bool,
+) -> None:
+    assert _can_route(analysis, risk_level) is expected
 
-    def test_high_confidence_eligible(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(confidence=0.9)
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is True
 
-    def test_low_confidence_blocked(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(confidence=0.7)
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
+def test_alertmanager_source_uses_queue_not_approval_executor() -> None:
+    source = inspect.getsource(webhooks._process_new_alert_background)
 
-    def test_critical_risk_blocked(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(confidence=0.95)
-        assert _can_auto(a, RiskLevel.CRITICAL, patterns) is False
-
-    def test_empty_kubectl_blocked(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(kubectl="")
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
-
-    def test_no_action_blocked(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(action_title="NO_ACTION: no fix needed")
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
-
-    def test_single_delete_pod_eligible(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(kubectl="kubectl delete pod foo-123")
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is True
-
-    def test_delete_pods_all_blocked(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(kubectl="kubectl delete pods --all -n prod")
-        assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
-
-    def test_destructive_force_check(self, patterns):
-        # --force 不一定在 pattern；只驗 _can_auto 回傳 bool
-        from src.models.approval import RiskLevel
-        a = _make_analysis(kubectl="kubectl rollout restart --force deployment/bar")
-        result = _can_auto(a, RiskLevel.MEDIUM, patterns)
-        assert isinstance(result, bool)
-
-    @pytest.mark.asyncio
-    async def test_execute_called_when_eligible(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(confidence=0.9)
-        risk_level = RiskLevel.MEDIUM
-
-        mock_svc = AsyncMock()
-        mock_svc.execute_approved_action = AsyncMock(return_value=True)
-
-        assert _can_auto(a, risk_level, patterns) is True
-        await mock_svc.execute_approved_action(MagicMock())
-        mock_svc.execute_approved_action.assert_called_once()
-
-    @pytest.mark.asyncio
-    async def test_execute_exception_does_not_propagate(self, patterns):
-        from src.models.approval import RiskLevel
-        a = _make_analysis(confidence=0.9)
-        risk_level = RiskLevel.MEDIUM
-
-        mock_svc = AsyncMock()
-        mock_svc.execute_approved_action = AsyncMock(side_effect=RuntimeError("boom"))
-
-        try:
-            if _can_auto(a, risk_level, patterns):
-                await mock_svc.execute_approved_action(MagicMock())
-        except Exception:
-            pass  # prod code wraps in try/except; test confirms pattern
-
-        assert True
+    assert "_try_auto_repair_background" in source
+    assert "execute_approved_action" not in source
+    assert "ApprovalExecutionService" not in source
+    assert "execute_auto_repair(" not in source

apps/api/tests/test_decision_manager_bare_metal_kubectl_guard.py

@@ -13,8 +13,9 @@ LLM 亂提「kubectl rollout restart awoooi-api」，要被擋下。
 
 from __future__ import annotations
 
+import inspect
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -59,6 +60,21 @@ def manager(monkeypatch):
         "src.services.decision_manager._escalate_decision_auto_repair_unavailable",
         lambda **k: None,
     )
+    async def _queue_candidate(**_kwargs):
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "controlled_check_mode_queued",
+            "automation_run_id": "00000000-0000-0000-0000-000000000101",
+            "queued": True,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": [],
+        }
+
+    monkeypatch.setattr(
+        "src.jobs.awooop_ansible_candidate_backfill_job.enqueue_ai_decision_ansible_candidate",
+        _queue_candidate,
+    )
     return mgr
 
 
@@ -146,3 +162,27 @@ class TestBareMetalKubectlGuard:
             pass
         br = token.proposal_data.get("blocked_reason", "")
         assert "host_type=bare_metal" not in br
+
+    @pytest.mark.asyncio
+    async def test_critical_decision_remains_break_glass(self, manager):
+        incident = _fake_incident(host_type="kubernetes")
+        token = _fake_token("kubectl rollout restart deployment awoooi-api")
+        token.proposal_data["risk_level"] = "critical"
+
+        await manager._auto_execute(incident, token)
+
+        assert token.state == DecisionState.READY
+        assert token.proposal_data["blocked_reason"] == "critical_break_glass_required"
+        assert token.proposal_data["direct_runtime_execution_performed"] is False
+
+
+def test_active_auto_execute_only_routes_to_single_writer_queue() -> None:
+    source = inspect.getsource(DecisionManager._auto_execute)
+    legacy_source = inspect.getsource(DecisionManager._legacy_auto_execute_disabled)
+
+    assert "enqueue_ai_decision_ansible_candidate" in source
+    assert "ApprovalExecutionService" not in source
+    assert "execute_approved_action" not in source
+    assert "_ssh_execute" not in source
+    assert "critical_break_glass_required" in source
+    assert "decision_manager_direct_execution_superseded" in legacy_source

apps/api/tests/test_executor_trust_boundary_readback.py

@@ -25,6 +25,16 @@ def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]:
         "worker_ssh_egress_denied": "true",
         "broker_ssh_egress_allowlisted": "true",
         "legacy_namespace_egress_allow_absent": "true",
+        "single_writer_source_verifier": (
+            "cd_tests_and_production_source_sha_v1"
+        ),
+        "decision_manager_queue_only": "true",
+        "webhook_router_queue_only": "true",
+        "failure_watcher_queue_only": "true",
+        "auto_approved_direct_execution_blocked": "true",
+        "candidate_idempotency_contract_verified": "true",
+        "apply_idempotency_contract_verified": "true",
+        "critical_break_glass_contract_verified": "true",
         "db_identity_verifier": (
             "runtime_role_fingerprint_secret_projection_and_"
             "representative_preflight_v2"
@@ -92,6 +102,27 @@ def test_executor_trust_boundary_requires_live_controls_and_matching_source() ->
     ] is True
     assert database_boundary["secret_values_read"] is False
     assert database_boundary["role_names_exposed"] is False
+    source_boundary = readback["autonomous_single_writer_source_boundary"]
+    assert source_boundary["source_boundary_verified"] is True
+    assert source_boundary["active_blockers"] == []
+    assert all(source_boundary["controls"].values())
+
+
+def test_executor_trust_boundary_single_writer_source_controls_fail_closed() -> None:
+    receipt = _verified_receipt()
+    receipt["failure_watcher_queue_only"] = "false"
+
+    readback = build_executor_trust_boundary_readback(
+        receipt,
+        deployed_source_sha="abc123",
+    )
+
+    source_boundary = readback["autonomous_single_writer_source_boundary"]
+    assert readback["production_boundary_verified"] is True
+    assert source_boundary["source_boundary_verified"] is False
+    assert "failure_watcher_queue_only_not_verified" in source_boundary[
+        "active_blockers"
+    ]
 
 
 def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None:

apps/api/tests/test_failure_watcher.py

@@ -8,6 +8,8 @@ FailureWatcher Service Tests - Phase 18 失敗自動修復閉環
 建立者: Claude Code (Phase 18.6 E2E 驗證)
 """
 
+import inspect
+
 import pytest
 
 from src.services.failure_watcher import (
@@ -22,6 +24,34 @@ class TestFailureClassification:
     def setup_method(self):
         self.service = FailureWatcherService()
 
+    def test_active_failure_watcher_has_no_recursive_direct_executor(self):
+        process_source = inspect.getsource(self.service.process_failure)
+        legacy_source = inspect.getsource(
+            self.service._legacy_execute_auto_repair_disabled
+        )
+
+        assert "self.execute_auto_repair(" not in process_source
+        assert "single_writer_retry_or_repair_queued" in process_source
+        assert "get_executor" not in inspect.getsource(
+            self.service.execute_auto_repair
+        )
+        assert "failure_watcher_direct_execution_disabled" in inspect.getsource(
+            self.service.execute_auto_repair
+        )
+        assert "restart_deployment" in legacy_source
+
+    @pytest.mark.asyncio
+    async def test_legacy_direct_repair_entrypoint_is_fail_closed(self):
+        with pytest.raises(
+            RuntimeError,
+            match="failure_watcher_direct_execution_disabled_use_single_writer_broker",
+        ):
+            await self.service.execute_auto_repair(
+                audit_log_id="audit-1",
+                repair_strategy="restart_deployment",
+                failure_data={"target_resource": "deployment/api"},
+            )
+
     def test_classify_timeout(self):
         """測試超時錯誤分類"""
         result = self.service._classify_by_rules("Connection timed out after 30s")

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -156,9 +156,19 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "legacy cluster-wide executor binding still exists" in workflow
     assert "awoooi-executor-boundary-verification" in workflow
     assert "awoooi_executor_boundary_verification_v3" in workflow
+    assert "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" in workflow
+    assert "decision_manager_queue_only=true" in workflow
+    assert "webhook_router_queue_only=true" in workflow
+    assert "failure_watcher_queue_only=true" in workflow
+    assert "auto_approved_direct_execution_blocked=true" in workflow
+    assert "candidate_idempotency_contract_verified=true" in workflow
+    assert "apply_idempotency_contract_verified=true" in workflow
+    assert "critical_break_glass_contract_verified=true" in workflow
     assert "executor_boundary_public_readback=verified_ready" in workflow
     assert 'boundary.get("full_workload_boundary_verified") is True' in workflow
     assert 'database_boundary.get("db_identity_isolated") is True' in workflow
+    assert "autonomous_single_writer_source_boundary" in workflow
+    assert '"source_boundary_verified"' in workflow
     assert 'database_boundary.get("connection_budget_verified") is True' in workflow
     assert "verified_source_sha" in workflow
     assert (