diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml
index ae1ee238b..bdf3d345e 100644
--- a/.gitea/workflows/cd.yaml
+++ b/.gitea/workflows/cd.yaml
@@ -3140,6 +3140,14 @@ jobs:
             --from-literal=worker_ssh_egress_denied=true \
             --from-literal=broker_ssh_egress_allowlisted=true \
             --from-literal=legacy_namespace_egress_allow_absent=true \
+            --from-literal=single_writer_source_verifier=cd_tests_and_production_source_sha_v1 \
+            --from-literal=decision_manager_queue_only=true \
+            --from-literal=webhook_router_queue_only=true \
+            --from-literal=failure_watcher_queue_only=true \
+            --from-literal=auto_approved_direct_execution_blocked=true \
+            --from-literal=candidate_idempotency_contract_verified=true \
+            --from-literal=apply_idempotency_contract_verified=true \
+            --from-literal=critical_break_glass_contract_verified=true \
             --from-literal=db_identity_verifier=runtime_role_fingerprint_secret_projection_and_representative_preflight_v2 \
             --from-literal=api_database_secret_ref="$API_DATABASE_SECRET_REF" \
             --from-literal=worker_database_secret_ref="$WORKER_DATABASE_SECRET_REF" \
@@ -3242,12 +3250,21 @@ jobs:
                   database_boundary = (
                       boundary.get("database_identity_boundary") or {}
                   )
+                  single_writer_source_boundary = (
+                      boundary.get(
+                          "autonomous_single_writer_source_boundary"
+                      )
+                      or {}
+                  )
                   last_status = str(boundary.get("status") or "missing")
                   if (
                       boundary.get("production_boundary_verified") is True
                       and boundary.get("full_workload_boundary_verified") is True
                       and database_boundary.get("db_identity_isolated") is True
                       and database_boundary.get("connection_budget_verified") is True
+                      and single_writer_source_boundary.get(
+                          "source_boundary_verified"
+                      ) is True
                       and str(boundary.get("verified_source_sha") or "").lower()
                       == expected
                   ):
diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py
index 8ea6a7ca6..9dab1239f 100644
--- a/apps/api/src/api/v1/webhooks.py
+++ b/apps/api/src/api/v1/webhooks.py
@@ -223,6 +223,90 @@ async def _try_auto_repair_background(
     alert_type: str,
     target_resource: str,
     namespace: str,
+    risk_level: str,
+) -> dict:
+    """Queue an alert decision for the single-writer executor."""
+
+    from src.repositories.alert_operation_log_repository import (
+        get_alert_operation_log_repository,
+    )
+    from src.jobs.awooop_ansible_candidate_backfill_job import (
+        enqueue_ai_decision_ansible_candidate,
+    )
+
+    op_log = get_alert_operation_log_repository()
+    incident = await get_incident_service().get_from_working_memory(incident_id)
+    if incident is None:
+        handoff = {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "incident_not_found",
+            "queued": False,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": ["incident_not_found"],
+        }
+    else:
+        handoff = await enqueue_ai_decision_ansible_candidate(
+            incident=incident,
+            proposal_data={
+                "source": "alert_webhook_controlled_router",
+                "risk_level": risk_level,
+                "action": target_resource,
+                "alert_type": alert_type,
+                "namespace": namespace,
+                "approval_id": approval_id,
+            },
+            project_id=str(getattr(incident, "project_id", None) or "awoooi"),
+        )
+
+    queued = handoff.get("queued") is True
+    await op_log.append(
+        "AUTO_REPAIR_TRIGGERED",
+        incident_id=incident_id,
+        approval_id=approval_id,
+        actor="single_writer_decision_router",
+        action_detail=(
+            "controlled_check_mode_queued"
+            if queued
+            else "controlled_queue_blocked"
+        ),
+        success=queued,
+        error_message=(
+            None
+            if queued
+            else ",".join(str(value) for value in handoff.get("active_blockers") or [])
+        ),
+        context={
+            "schema_version": handoff.get("schema_version"),
+            "status": handoff.get("status"),
+            "automation_run_id": handoff.get("automation_run_id"),
+            "single_writer_executor": handoff.get("single_writer_executor"),
+            "side_effect_performed": False,
+            "safe_next_action": (
+                "awooop_ansible_check_mode_worker_claims_candidate"
+                if queued
+                else "repair_candidate_backfill_or_playbook_coverage_gap"
+            ),
+        },
+    )
+    logger.info(
+        "auto_repair_routed_to_single_writer_executor",
+        incident_id=incident_id,
+        approval_id=approval_id,
+        queued=queued,
+        status=handoff.get("status"),
+        automation_run_id=handoff.get("automation_run_id"),
+        side_effect_performed=False,
+    )
+    return handoff
+
+
+async def _legacy_try_auto_repair_background_disabled(
+    incident_id: str,
+    approval_id: str,
+    alert_type: str,
+    target_resource: str,
+    namespace: str,
 ) -> None:
     """
     背景評估並執行自動修復
@@ -234,6 +318,9 @@ async def _try_auto_repair_background(
     4. 不可修復 → 靜默，等人工批准
     所有步驟都寫入 alert_operation_log
     """
+    raise RuntimeError(
+        "webhook_direct_auto_repair_superseded_by_single_writer_broker"
+    )
     from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
     op_log = get_alert_operation_log_repository()
 
@@ -1366,11 +1453,36 @@ async def receive_alert(
             request=approval_create,
             fingerprint=fingerprint,
         )
+        _cs1_risk_level = approval_create.risk_level
+        _cs1_alert_category = {
+            "k8s_node_failure": "kubernetes",
+            "k8s_pod_crash": "kubernetes",
+            "db_connection_timeout": "database",
+            "high_cpu": "host_resource",
+            "high_memory": "host_resource",
+            "disk_full": "storage",
+            "ssl_expiry": "ssl_cert",
+        }.get(alert.alert_type, "general")
+        _cs1_incident_id = await create_incident_for_approval(
+            approval_id=str(approval.id),
+            risk_level=_cs1_risk_level.value,
+            target_resource=alert.target_resource,
+            namespace=alert.namespace,
+            alert_type=alert.alert_type,
+            message=alert.message,
+            source=alert.source,
+            alertname=str((alert.labels or {}).get("alertname") or alert.alert_type),
+            alert_labels=alert.labels or {},
+            notification_type="generic",
+            alert_category=_cs1_alert_category,
+        )
+        await service.update_incident_id(approval.id, _cs1_incident_id)
+        approval.incident_id = _cs1_incident_id
 
         # 2026-04-27 Claude Sonnet 4.6: shadow-run Step2 — 只記 log，不改執行決策
         try:
             _shadow_proposal = {
-                "risk_level": risk_level.value,
+                "risk_level": _cs1_risk_level.value,
                 "confidence": getattr(approval_create, "metadata", {}).get("confidence_score", 0.0) if approval_create.metadata else 0.0,
                 "action": approval_create.action,
                 "kubectl_command": approval_create.action,
@@ -1396,66 +1508,19 @@ async def receive_alert(
             _cs1_can_auto = (
                 bool(_cs1_kubectl)
                 and analysis_result.confidence >= 0.85
-                and risk_level != RiskLevel.CRITICAL
+                and _cs1_risk_level != RiskLevel.CRITICAL
                 and _sa_val not in _non_destructive_actions
                 and is_safe_kubectl_action(_cs1_kubectl)
             )
             if _cs1_can_auto:
-                try:
-                    from src.models.approval import ApprovalRequest, ApprovalStatus
-                    from src.services.approval_execution import ApprovalExecutionService
-
-                    _cs1_auto_approval = ApprovalRequest(
-                        incident_id=str(approval.incident_id) if approval.incident_id else None,
-                        action=approval_create.action,
-                        description=approval_create.description,
-                        requested_by="auto_approve_llm_high_confidence",
-                        required_signatures=0,
-                        status=ApprovalStatus.APPROVED,
-                        risk_level=risk_level.value,
-                        matched_playbook_id=_matched_playbook_id_cs1,
-                        metadata={
-                            **_approval_metadata_cs1,
-                            "is_high_confidence": True,
-                            "policy_reason": _shadow_result.reason.value
-                            if hasattr(_shadow_result, "reason")
-                            else "cs1_auto_confident_execution",
-                        },
-                    )
-                    _cs1_auto_approval.id = approval.id
-
-                    _cs1_executor = ApprovalExecutionService()
-                    _cs1_exec_success = await _cs1_executor.execute_approved_action(_cs1_auto_approval)
-
-                    try:
-                        await service.update_execution_status(approval.id, _cs1_exec_success)
-                    except Exception as _cs1_upd_err:
-                        logger.warning(
-                            "cs1_auto_execute_status_update_failed",
-                            approval_id=str(approval.id),
-                            error=str(_cs1_upd_err),
-                        )
-
-                        logger.info(
-                            "llm_high_confidence_auto_executed",
-                            approval_id=str(approval.id),
-                            confidence=analysis_result.confidence,
-                            exec_success=_cs1_exec_success,
-                            action=_cs1_kubectl[:80],
-                            is_high_confidence=True,
-                            policy_reason=(
-                                _shadow_result.reason.value
-                                if hasattr(_shadow_result, "reason")
-                                else "cs1_auto_confident_execution"
-                            ),
-                        )
-                except Exception as _cs1_auto_err:
-                    logger.warning(
-                        "llm_high_confidence_auto_execute_failed",
-                        approval_id=str(approval.id),
-                        error=str(_cs1_auto_err),
-                    )
-                    # 降級：維持 PENDING，流程繼續到 Telegram 推送
+                await _try_auto_repair_background(
+                    incident_id=_cs1_incident_id,
+                    approval_id=str(approval.id),
+                    alert_type=alert.alert_type,
+                    target_resource=alert.target_resource,
+                    namespace=alert.namespace,
+                    risk_level=_cs1_risk_level.value,
+                )
 
         logger.info(
             "approval_auto_created_with_fingerprint",
@@ -1524,7 +1589,7 @@ async def receive_alert(
             ai_cost=ai_cost,
             ai_provider=ai_provider,
             # 2026-04-08 ogt: 補傳 incident_id 以啟用詳情/重診/歷史按鈕
-            incident_id="",  # /alerts 路徑尚無 incident，detail/reanalyze/history 按鈕不顯示
+            incident_id=_cs1_incident_id,
             # /alerts 路徑沒有 notification_type（非 Alertmanager 路徑），不需 TYPE-4D routing
             # ADR-075: alert_type → category 對應，啟用動態按鈕
             alert_category={
@@ -1795,62 +1860,19 @@ async def _process_new_alert_background(
             # 2026-04-27 ogt + Claude Sonnet 4.6: CS2 規則引擎自動執行
             # 設計：is_rule_based=True 確定性高，滿足條件直接執行，不等人工審核
             # 安全防線：CRITICAL / destructive patterns / NO_ACTION / 空 kubectl → 全部降級 PENDING
-            _cs2_auto_approval = None
-            _cs2_executor = None
-            _cs2_exec_success: bool | None = None
-            _cs2_exec_error: str | None = None
-            try:
-                from src.models.approval import ApprovalRequest, ApprovalStatus
-                from src.services.approval_execution import ApprovalExecutionService
-
-                _can_auto = (
-                    bool(rule_kubectl)
-                    and rule_risk != RiskLevel.CRITICAL
-                    and is_safe_kubectl_action(rule_kubectl)
-                    and "NO_ACTION" not in rule_action
-                )
-                if _can_auto:
-                    _auto_approval = ApprovalRequest(
-                        incident_id=None,  # 尚未建立，稍後 update_incident_id 補上
-                        action=rule_action,
-                        description=approval_create.description,
-                        requested_by="auto_approve_rule_engine",
-                        required_signatures=0,
-                        status=ApprovalStatus.APPROVED,
-                        risk_level=rule_risk.value,
-                        matched_playbook_id=_matched_playbook_id_cs2,
-                    )
-                    # 使用 DB 中剛建立的 approval.id 讓 executor 可回寫
-                    _auto_approval.id = approval.id
-                    _cs2_auto_approval = _auto_approval
-
-                    _cs2_executor = ApprovalExecutionService()
-                    _cs2_exec_success = await _cs2_executor.execute_approved_action(_auto_approval)
-
-                    # 更新 DB approval 執行狀態
-                    try:
-                        await service.update_execution_status(approval.id, _cs2_exec_success)
-                    except Exception as _upd_err:
-                        logger.warning(
-                            "cs2_auto_execute_status_update_failed",
-                            approval_id=str(approval.id),
-                            error=str(_upd_err),
-                        )
-
-                    logger.info(
-                        "rule_engine_auto_executed",
-                        approval_id=str(approval.id),
-                        rule_id=rule_response.get("rule_id", "unknown"),
-                        kubectl=rule_kubectl,
-                        exec_success=_cs2_exec_success,
-                    )
-            except Exception as _auto_err:
-                _cs2_exec_success = False if _cs2_auto_approval is not None else None
-                _cs2_exec_error = str(_auto_err)
-                logger.warning(
-                    "cs2_auto_execute_failed_degraded_to_pending",
+            _can_auto = (
+                bool(rule_kubectl)
+                and rule_risk != RiskLevel.CRITICAL
+                and is_safe_kubectl_action(rule_kubectl)
+                and "NO_ACTION" not in rule_action
+            )
+            if _can_auto:
+                logger.info(
+                    "rule_engine_auto_execution_deferred_to_single_writer",
                     approval_id=str(approval.id),
-                    error=str(_auto_err),
+                    rule_id=rule_response.get("rule_id", "unknown"),
+                    kubectl=rule_kubectl,
+                    direct_execution_performed=False,
                 )
 
             incident_id = await create_incident_for_approval(
@@ -1896,23 +1918,6 @@ async def _process_new_alert_background(
                 annotations=alert_context.get("annotations", {}),
             )
 
-            if _cs2_auto_approval is not None and _cs2_exec_success is not None:
-                try:
-                    _cs2_auto_approval.incident_id = incident_id
-                    _cs2_executor = _cs2_executor or ApprovalExecutionService()
-                    await _cs2_executor.finalize_auto_approved_execution(
-                        _cs2_auto_approval,
-                        success=_cs2_exec_success,
-                        error_message=_cs2_exec_error,
-                    )
-                except Exception as _cs2_finalize_err:
-                    logger.warning(
-                        "cs2_auto_execute_finalize_failed",
-                        approval_id=str(approval.id),
-                        incident_id=incident_id,
-                        error=str(_cs2_finalize_err),
-                    )
-
             _is_heartbeat = is_heartbeat_alertname(alertname)
             if can_auto_repair and not _is_heartbeat:
                 await _try_auto_repair_background(
@@ -1921,6 +1926,7 @@ async def _process_new_alert_background(
                     alert_type=alert_type,
                     target_resource=target_resource,
                     namespace=namespace,
+                    risk_level=rule_risk.value,
                 )
             elif not can_auto_repair and not _is_heartbeat:
                 from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
@@ -2081,60 +2087,15 @@ async def _process_new_alert_background(
                 and "NO_ACTION" not in (analysis_result.action_title or "")
                 and is_safe_kubectl_action(_cs3_kubectl)
             )
-            _cs3_auto_approval = None
-            _cs3_executor = None
-            _cs3_exec_success: bool | None = None
-            _cs3_exec_error: str | None = None
             if _cs3_can_auto:
-                try:
-                    from src.models.approval import ApprovalRequest, ApprovalStatus
-                    from src.services.approval_execution import ApprovalExecutionService
-
-                    _cs3_auto_approval = ApprovalRequest(
-                        action=approval_create.action,
-                        description=approval_create.description,
-                        requested_by="auto_approve_llm_cs3",
-                        required_signatures=0,
-                        status=ApprovalStatus.APPROVED,
-                        risk_level=risk_level.value,
-                        matched_playbook_id=_matched_playbook_id_cs3,
-                        metadata={
-                            **_approval_metadata_cs3,
-                            "is_high_confidence": True,
-                            "policy_reason": _shadow_result_cs3.reason.value
-                            if hasattr(_shadow_result_cs3, "reason")
-                            else "cs3_auto_confident_execution",
-                        },
-                    )
-                    _cs3_auto_approval.id = approval.id
-                    _cs3_executor = ApprovalExecutionService()
-                    _cs3_exec_success = await _cs3_executor.execute_approved_action(_cs3_auto_approval)
-                    try:
-                        await service.update_execution_status(approval.id, _cs3_exec_success)
-                    except Exception as _cs3_upd_err:
-                        logger.warning(
-                            "cs3_auto_execute_status_update_failed",
-                            approval_id=str(approval.id),
-                            error=str(_cs3_upd_err),
-                        )
-                    logger.info(
-                        "cs3_llm_auto_executed",
-                        approval_id=str(approval.id),
-                        kubectl=_cs3_kubectl,
-                        confidence=analysis_result.confidence,
-                        success=_cs3_exec_success,
-                        provider=ai_provider,
-                        is_high_confidence=True,
-                        policy_reason=(
-                            _shadow_result_cs3.reason.value
-                            if hasattr(_shadow_result_cs3, "reason")
-                            else "cs3_auto_confident_execution"
-                        ),
-                    )
-                except Exception as _cs3_exec_err:
-                    _cs3_exec_success = False if _cs3_auto_approval is not None else None
-                    _cs3_exec_error = str(_cs3_exec_err)
-                    logger.warning("cs3_llm_auto_execute_failed", error=str(_cs3_exec_err))
+                logger.info(
+                    "cs3_llm_auto_execution_deferred_to_single_writer",
+                    approval_id=str(approval.id),
+                    kubectl=_cs3_kubectl,
+                    confidence=analysis_result.confidence,
+                    provider=ai_provider,
+                    direct_execution_performed=False,
+                )
 
             incident_id = await create_incident_for_approval(
                 approval_id=str(approval.id),
@@ -2179,23 +2140,6 @@ async def _process_new_alert_background(
                 annotations=alert_context.get("annotations", {}),
             )
 
-            if _cs3_auto_approval is not None and _cs3_exec_success is not None:
-                try:
-                    _cs3_auto_approval.incident_id = incident_id
-                    _cs3_executor = _cs3_executor or ApprovalExecutionService()
-                    await _cs3_executor.finalize_auto_approved_execution(
-                        _cs3_auto_approval,
-                        success=_cs3_exec_success,
-                        error_message=_cs3_exec_error,
-                    )
-                except Exception as _cs3_finalize_err:
-                    logger.warning(
-                        "cs3_auto_execute_finalize_failed",
-                        approval_id=str(approval.id),
-                        incident_id=incident_id,
-                        error=str(_cs3_finalize_err),
-                    )
-
             root_cause = analysis_result.description or message
             estimated_downtime = blast.estimated_downtime if blast else "~30s"
             primary_responsibility = analysis_result.primary_responsibility or "COLLAB"
@@ -2216,6 +2160,7 @@ async def _process_new_alert_background(
                     alert_type=alert_type,
                     target_resource=target_resource,
                     namespace=namespace,
+                    risk_level=risk_level.value,
                 )
             else:
                 from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
index 2d9fc6871..0099534b5 100644
--- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
+++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
@@ -88,6 +88,7 @@ async def _fetch_missing_candidate_incidents(
                       WHERE existing.operation_type = 'ansible_candidate_matched'
                         AND existing.created_at >= NOW() - (:window_hours * INTERVAL '1 hour')
                         AND existing.input ->> 'executor' = 'ansible'
+                        AND existing.input ->> 'decision_path' = 'repair_candidate_controlled_queue'
                         AND coalesce(existing.incident_id::text, existing.input ->> 'incident_id') = incidents.incident_id::text
                   )
                 ORDER BY created_at DESC
@@ -163,12 +164,17 @@ def _incident_for_evidence(incident: dict[str, Any]) -> SimpleNamespace:
 
 async def collect_ansible_candidate_pre_decision_evidence(
     *,
-    incident: dict[str, Any],
+    incident: Any,
     project_id: str,
     automation_run_id: str,
 ) -> EvidenceSnapshot | None:
+    evidence_incident = (
+        _incident_for_evidence(incident)
+        if isinstance(incident, dict)
+        else incident
+    )
     return await get_pre_decision_investigator().investigate(
-        _incident_for_evidence({**incident, "project_id": project_id}),
+        evidence_incident,
         project_id=project_id,
         automation_run_id=automation_run_id,
     )
@@ -218,6 +224,125 @@ async def verify_ansible_candidate_pre_decision_evidence(
         return verified.scalar() is True
 
 
+async def enqueue_ai_decision_ansible_candidate(
+    *,
+    incident: Any,
+    proposal_data: dict[str, Any],
+    project_id: str = "awoooi",
+    recorder: Recorder = record_ansible_decision_audit,
+    evidence_collector: EvidenceCollector = (
+        collect_ansible_candidate_pre_decision_evidence
+    ),
+    evidence_verifier: EvidenceVerifier = (
+        verify_ansible_candidate_pre_decision_evidence
+    ),
+) -> dict[str, Any]:
+    """Queue one AI decision for the broker without executing a side effect."""
+
+    risk_value = proposal_data.get("risk_level")
+    risk_level = str(getattr(risk_value, "value", risk_value) or "").strip().lower()
+    if risk_level == "critical":
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "critical_break_glass_required",
+            "queued": False,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": ["critical_break_glass_required"],
+        }
+    if risk_level not in {"low", "medium", "high"}:
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "risk_policy_not_resolved",
+            "queued": False,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": ["risk_policy_not_resolved"],
+        }
+
+    normalized_proposal = {
+        **proposal_data,
+        "source": str(proposal_data.get("source") or "decision_manager"),
+        "risk_level": risk_level,
+    }
+    payload = build_ansible_decision_audit_payload(
+        incident=incident,
+        proposal_data=normalized_proposal,
+        decision_path=_BACKFILL_DECISION_PATH,
+        not_used_reason=(
+            "AI decision routed to the single-writer Ansible broker; "
+            "check-mode and diff are required before controlled apply"
+        ),
+    )
+    if payload is None:
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "no_allowlisted_ansible_candidate",
+            "queued": False,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": ["no_allowlisted_ansible_candidate"],
+        }
+
+    automation_run_id = str(uuid4())
+    try:
+        snapshot = await evidence_collector(
+            incident=incident,
+            project_id=project_id,
+            automation_run_id=automation_run_id,
+        )
+        evidence_ready = await evidence_verifier(
+            snapshot=snapshot,
+            project_id=project_id,
+        )
+    except Exception as exc:
+        logger.warning(
+            "ai_decision_controlled_executor_evidence_failed",
+            incident_id=payload["input"].get("incident_id"),
+            automation_run_id=automation_run_id,
+            error_type=type(exc).__name__,
+        )
+        evidence_ready = False
+    if not evidence_ready:
+        return {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "pre_decision_evidence_not_verified",
+            "automation_run_id": automation_run_id,
+            "queued": False,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "active_blockers": ["pre_decision_evidence_not_verified"],
+        }
+
+    inserted = await recorder(
+        incident=incident,
+        proposal_data=normalized_proposal,
+        decision_path=_BACKFILL_DECISION_PATH,
+        not_used_reason=(
+            "AI decision routed to the single-writer Ansible broker; "
+            "check-mode and diff are required before controlled apply"
+        ),
+        automation_run_id=automation_run_id,
+    )
+    return {
+        "schema_version": "ai_decision_controlled_executor_handoff_v1",
+        "status": (
+            "controlled_check_mode_queued"
+            if inserted
+            else "existing_candidate_or_write_not_acknowledged"
+        ),
+        "automation_run_id": automation_run_id,
+        "idempotency_key": payload["input"]["idempotency_key"],
+        "queued": inserted,
+        "side_effect_performed": False,
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+        "target_selector": payload["input"]["target_selector"],
+        "source_truth_diff": payload["input"]["source_truth_diff"],
+        "risk_policy_decision": payload["input"]["risk_policy_decision"],
+        "active_blockers": [] if inserted else ["candidate_write_not_acknowledged"],
+    }
+
+
 async def enqueue_missing_ansible_candidates_once(
     *,
     project_id: str = "awoooi",
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 f2e523a1d..7642e2828 100644
--- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py
+++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py
@@ -4460,6 +4460,97 @@ def _control_plane_integration() -> dict[str, Any]:
     }
 
 
+def _runtime_truthy(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    return str(value or "").strip().lower() == "true"
+
+
+def _build_autonomous_single_writer_runtime_readback(
+    *,
+    operation_latest_rows: Iterable[Mapping[str, Any] | Any],
+    loop_ledger: Mapping[str, Any],
+) -> dict[str, Any]:
+    rows = [_row_mapping(row) for row in operation_latest_rows]
+    automation_run_id = str(loop_ledger.get("automation_run_id") or "")
+    run_rows = [
+        row
+        for row in rows
+        if str(row.get("automation_run_id") or "") == automation_run_id
+    ]
+
+    def _operation(operation_type: str) -> dict[str, Any]:
+        return next(
+            (
+                row
+                for row in run_rows
+                if str(row.get("operation_type") or "") == operation_type
+            ),
+            {},
+        )
+
+    candidate = _operation("ansible_candidate_matched")
+    check_mode = _operation("ansible_check_mode_executed")
+    controlled_apply = _operation("ansible_apply_executed")
+    writer_rows = [candidate, check_mode, controlled_apply]
+    router_source_sha = str(candidate.get("router_source_sha") or "").lower()
+    controls = {
+        "controlled_queue_decision_path": (
+            candidate.get("decision_path") == "repair_candidate_controlled_queue"
+        ),
+        "single_writer_executor": bool(
+            all(writer_rows)
+            and all(
+                row.get("single_writer_executor")
+                == "awoooi-ansible-executor-broker"
+                for row in writer_rows
+            )
+        ),
+        "candidate_idempotency_key_present": bool(
+            str(candidate.get("candidate_idempotency_key") or "")
+        ),
+        "target_selector_present": _runtime_truthy(
+            candidate.get("target_selector_present")
+        ),
+        "source_truth_diff_required": _runtime_truthy(
+            candidate.get("source_truth_diff_required")
+        ),
+        "check_mode_receipt_present": bool(
+            check_mode
+            and str(check_mode.get("status") or "") == "success"
+        ),
+        "apply_idempotency_key_present": bool(
+            str(controlled_apply.get("apply_idempotency_key") or "")
+        ),
+        "same_run_apply_chain": bool(
+            automation_run_id
+            and candidate
+            and check_mode
+            and controlled_apply
+            and loop_ledger.get("same_run_correlation") is True
+        ),
+        "router_source_sha_present": bool(router_source_sha),
+    }
+    blockers = [
+        f"{control_id}_not_verified"
+        for control_id, verified in controls.items()
+        if verified is not True
+    ]
+    return {
+        "schema_version": "awoooi_autonomous_single_writer_runtime_v1",
+        "status": "runtime_evidence_ready" if not blockers else "in_progress",
+        "automation_run_id": automation_run_id or None,
+        "candidate_op_id": str(candidate.get("op_id") or "") or None,
+        "check_mode_op_id": str(check_mode.get("op_id") or "") or None,
+        "apply_op_id": str(controlled_apply.get("op_id") or "") or None,
+        "router_source_sha": router_source_sha or None,
+        "runtime_evidence_ready": not blockers,
+        "controls": controls,
+        "active_blockers": blockers,
+        "writes_on_read": False,
+    }
+
+
 def build_runtime_receipt_readback_from_rows(
     *,
     project_id: str = _DEFAULT_PROJECT_ID,
@@ -4598,6 +4689,10 @@ def build_runtime_receipt_readback_from_rows(
         latest_failure_classification=latest_failure,
         controlled_retry_package=retry_package,
     )
+    single_writer_runtime = _build_autonomous_single_writer_runtime_readback(
+        operation_latest_rows=operation_latest,
+        loop_ledger=loop_ledger,
+    )
     trace_ledger = _build_trace_ledger(
         operation_summary=operation_summary,
         auto_repair_summary=auto_repair_summary,
@@ -4715,6 +4810,13 @@ def build_runtime_receipt_readback_from_rows(
                     "check_mode_op_id",
                     "risk_level",
                     "controlled_apply_allowed",
+                    "decision_path",
+                    "single_writer_executor",
+                    "candidate_idempotency_key",
+                    "apply_idempotency_key",
+                    "source_truth_diff_required",
+                    "target_selector_present",
+                    "router_source_sha",
                     "capability_op_id",
                     "capability_issued_at",
                     "capability_expires_at",
@@ -4725,6 +4827,7 @@ def build_runtime_receipt_readback_from_rows(
                 ),
             ),
         },
+        "autonomous_single_writer_runtime": single_writer_runtime,
         "auto_repair_execution_receipt": {
             **auto_repair_summary,
             "latest": _sanitize_latest_rows(
@@ -6260,6 +6363,11 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
         )
     _attach_runtime_receipt_readback(payload, readback)
     payload["executor_trust_boundary"] = boundary_readback
+    _attach_autonomous_single_writer_readback(
+        payload,
+        readback=readback,
+        boundary_readback=boundary_readback,
+    )
     payload.setdefault("rollups", {})[
         "executor_trust_boundary_verified_count"
     ] = 1 if boundary_readback.get("production_boundary_verified") is True else 0
@@ -6267,6 +6375,89 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
     return payload
 
 
+def _attach_autonomous_single_writer_readback(
+    payload: dict[str, Any],
+    *,
+    readback: Mapping[str, Any],
+    boundary_readback: Mapping[str, Any],
+) -> None:
+    runtime = readback.get("autonomous_single_writer_runtime")
+    if not isinstance(runtime, Mapping):
+        runtime = {}
+    source_boundary = boundary_readback.get(
+        "autonomous_single_writer_source_boundary"
+    )
+    if not isinstance(source_boundary, Mapping):
+        source_boundary = {}
+    strict_runtime = payload.get("strict_runtime_completion")
+    if not isinstance(strict_runtime, Mapping):
+        strict_runtime = {}
+    runtime_controls = runtime.get("controls")
+    if not isinstance(runtime_controls, Mapping):
+        runtime_controls = {}
+    source_controls = source_boundary.get("controls")
+    if not isinstance(source_controls, Mapping):
+        source_controls = {}
+
+    runtime_run_id = str(runtime.get("automation_run_id") or "")
+    strict_run_id = str(strict_runtime.get("automation_run_id") or "")
+    deployed_source_sha = str(
+        boundary_readback.get("deployed_source_sha") or ""
+    ).lower()
+    router_source_sha = str(runtime.get("router_source_sha") or "").lower()
+    controls = {
+        "strict_runtime_same_run_closed": bool(
+            strict_runtime.get("closed") is True
+            and runtime_run_id
+            and runtime_run_id == strict_run_id
+        ),
+        "router_source_sha_matches_deployment": bool(
+            deployed_source_sha
+            and router_source_sha == deployed_source_sha
+        ),
+        **{
+            str(control_id): verified is True
+            for control_id, verified in runtime_controls.items()
+        },
+        **{
+            str(control_id): verified is True
+            for control_id, verified in source_controls.items()
+        },
+    }
+    blockers = [
+        *list(runtime.get("active_blockers") or []),
+        *list(source_boundary.get("active_blockers") or []),
+        *[
+            f"{control_id}_not_verified"
+            for control_id, verified in controls.items()
+            if verified is not True
+        ],
+    ]
+    blockers = list(dict.fromkeys(str(blocker) for blocker in blockers if blocker))
+    closed = bool(controls and not blockers)
+    payload["autonomous_single_writer"] = {
+        "schema_version": "awoooi_autonomous_single_writer_readback_v1",
+        "work_item_id": "AIA-P0-002",
+        "status": "verified_ready" if closed else "in_progress",
+        "closed": closed,
+        "automation_run_id": runtime_run_id or None,
+        "router_source_sha": router_source_sha or None,
+        "deployed_source_sha": deployed_source_sha or None,
+        "candidate_op_id": runtime.get("candidate_op_id"),
+        "check_mode_op_id": runtime.get("check_mode_op_id"),
+        "apply_op_id": runtime.get("apply_op_id"),
+        "source_receipt_ref": source_boundary.get("receipt_ref"),
+        "source_verifier": source_boundary.get("verifier"),
+        "controls": controls,
+        "active_blockers": blockers,
+        "writes_on_read": False,
+        "source_only_counts_as_completion": False,
+    }
+    payload.setdefault("rollups", {})[
+        "autonomous_single_writer_closed_count"
+    ] = 1 if closed else 0
+
+
 _RUNTIME_OPERATION_COUNTS_SQL = """
     SELECT
         coalesce(input ->> 'semantic_operation_type', operation_type)
@@ -6353,6 +6544,18 @@ _RUNTIME_OPERATION_LATEST_SQL = """
         input ->> 'check_mode_op_id' AS check_mode_op_id,
         input ->> 'risk_level' AS risk_level,
         input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
+        input ->> 'decision_path' AS decision_path,
+        input ->> 'single_writer_executor' AS single_writer_executor,
+        coalesce(
+            input ->> 'candidate_idempotency_key',
+            input ->> 'idempotency_key'
+        ) AS candidate_idempotency_key,
+        input ->> 'apply_idempotency_key' AS apply_idempotency_key,
+        input #>> '{source_truth_diff,required_before_apply}'
+            AS source_truth_diff_required,
+        coalesce(input -> 'target_selector', '{}'::jsonb) <> '{}'::jsonb
+            AS target_selector_present,
+        input ->> 'router_source_sha' AS router_source_sha,
         coalesce(
             input ->> 'capability_op_id',
             input #>> '{execution_capability,capability_op_id}'
@@ -6453,6 +6656,18 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """
         input ->> 'check_mode_op_id' AS check_mode_op_id,
         input ->> 'risk_level' AS risk_level,
         input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
+        input ->> 'decision_path' AS decision_path,
+        input ->> 'single_writer_executor' AS single_writer_executor,
+        coalesce(
+            input ->> 'candidate_idempotency_key',
+            input ->> 'idempotency_key'
+        ) AS candidate_idempotency_key,
+        input ->> 'apply_idempotency_key' AS apply_idempotency_key,
+        input #>> '{source_truth_diff,required_before_apply}'
+            AS source_truth_diff_required,
+        coalesce(input -> 'target_selector', '{}'::jsonb) <> '{}'::jsonb
+            AS target_selector_present,
+        input ->> 'router_source_sha' AS router_source_sha,
         coalesce(
             input ->> 'capability_op_id',
             input #>> '{execution_capability,capability_op_id}'
@@ -6542,6 +6757,18 @@ _RUNTIME_OPERATION_CHAIN_SQL = """
         input ->> 'check_mode_op_id' AS check_mode_op_id,
         input ->> 'risk_level' AS risk_level,
         input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
+        input ->> 'decision_path' AS decision_path,
+        input ->> 'single_writer_executor' AS single_writer_executor,
+        coalesce(
+            input ->> 'candidate_idempotency_key',
+            input ->> 'idempotency_key'
+        ) AS candidate_idempotency_key,
+        input ->> 'apply_idempotency_key' AS apply_idempotency_key,
+        input #>> '{source_truth_diff,required_before_apply}'
+            AS source_truth_diff_required,
+        coalesce(input -> 'target_selector', '{}'::jsonb) <> '{}'::jsonb
+            AS target_selector_present,
+        input ->> 'router_source_sha' AS router_source_sha,
         coalesce(
             input ->> 'capability_op_id',
             input #>> '{execution_capability,capability_op_id}'
@@ -6585,6 +6812,18 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """
         input ->> 'check_mode_op_id' AS check_mode_op_id,
         input ->> 'risk_level' AS risk_level,
         input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
+        input ->> 'decision_path' AS decision_path,
+        input ->> 'single_writer_executor' AS single_writer_executor,
+        coalesce(
+            input ->> 'candidate_idempotency_key',
+            input ->> 'idempotency_key'
+        ) AS candidate_idempotency_key,
+        input ->> 'apply_idempotency_key' AS apply_idempotency_key,
+        input #>> '{source_truth_diff,required_before_apply}'
+            AS source_truth_diff_required,
+        coalesce(input -> 'target_selector', '{}'::jsonb) <> '{}'::jsonb
+            AS target_selector_present,
+        input ->> 'router_source_sha' AS router_source_sha,
         coalesce(
             input ->> 'capability_op_id',
             input #>> '{execution_capability,capability_op_id}'
diff --git a/apps/api/src/services/approval_execution.py b/apps/api/src/services/approval_execution.py
index eaee13e45..c609957ef 100644
--- a/apps/api/src/services/approval_execution.py
+++ b/apps/api/src/services/approval_execution.py
@@ -193,6 +193,18 @@ class ApprovalExecutionService:
         """
         from src.services.notifications import ExecutionStatus
 
+        if self._is_auto_approved_request(approval):
+            logger.error(
+                "auto_approved_direct_execution_blocked",
+                approval_id=str(getattr(approval, "id", "")),
+                incident_id=getattr(approval, "incident_id", None),
+                requested_by=getattr(approval, "requested_by", None),
+                required_executor="awoooi-ansible-executor-broker",
+            )
+            raise RuntimeError(
+                "auto_approved_direct_execution_disabled_use_single_writer_broker"
+            )
+
         logger.info(
             "background_execution_start",
             approval_id=str(approval.id),
diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py
index daa4d4e18..6ad0f0325 100644
--- a/apps/api/src/services/awoooi_priority_work_order_readback.py
+++ b/apps/api/src/services/awoooi_priority_work_order_readback.py
@@ -271,6 +271,26 @@ _AI_AUTOMATION_PROGRAM_DEPENDENCIES = {
     "AIA-P0-010": ["AIA-P0-006", "AIA-P0-011"],
     "AIA-P0-007": ["AIA-P0-002", "AIA-P0-003", "AIA-P0-004", "AIA-P0-005", "AIA-P0-006", "AIA-P0-008"],
 }
+_AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS = (
+    "strict_runtime_same_run_closed",
+    "router_source_sha_matches_deployment",
+    "controlled_queue_decision_path",
+    "single_writer_executor",
+    "candidate_idempotency_key_present",
+    "target_selector_present",
+    "source_truth_diff_required",
+    "check_mode_receipt_present",
+    "apply_idempotency_key_present",
+    "same_run_apply_chain",
+    "router_source_sha_present",
+    "decision_manager_queue_only",
+    "webhook_router_queue_only",
+    "failure_watcher_queue_only",
+    "auto_approved_direct_execution_blocked",
+    "candidate_idempotency_contract_verified",
+    "apply_idempotency_contract_verified",
+    "critical_break_glass_contract_verified",
+)
 _AI_AUTOMATION_PROGRAM_SCOPE_CLASSES: list[dict[str, Any]] = [
     {
         "id": "hosts",
@@ -7404,6 +7424,80 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
         ]
     elif boundary_verified or capability_lifecycle_closed:
         trust_boundary_work_item["status"] = "in_progress"
+    single_writer_work_item = next(
+        item for item in work_items if item["id"] == "AIA-P0-002"
+    )
+    single_writer_closed = bool(
+        summary.get("ai_agent_single_writer_runtime_closed") is True
+    )
+    observed_single_writer_controls = _dict(
+        summary.get("ai_agent_single_writer_controls")
+    )
+    single_writer_controls = {
+        control_id: observed_single_writer_controls.get(control_id) is True
+        for control_id in _AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS
+    }
+    single_writer_present_count = sum(single_writer_controls.values())
+    single_writer_blockers = list(
+        summary.get("ai_agent_single_writer_active_blockers") or []
+    )
+    single_writer_blockers.extend(
+        f"{control_id}_not_verified"
+        for control_id, verified in single_writer_controls.items()
+        if verified is not True
+    )
+    single_writer_work_item["runtime_progress"] = {
+        "completion_percent": round(
+            single_writer_present_count
+            / len(_AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS)
+            * 100
+        ),
+        "present_control_count": single_writer_present_count,
+        "required_control_count": len(_AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS),
+        "controls": single_writer_controls,
+        "missing": list(dict.fromkeys(single_writer_blockers)),
+        "runtime_closed": single_writer_closed,
+    }
+    if single_writer_closed and all(single_writer_controls.values()):
+        single_writer_work_item["status"] = "done"
+        single_writer_work_item["completion_evidence"] = {
+            "source": "agent-autonomous-runtime-control",
+            "automation_run_id": str(
+                summary.get("ai_agent_single_writer_automation_run_id") or ""
+            ),
+            "router_source_sha": str(
+                summary.get("ai_agent_single_writer_router_source_sha") or ""
+            ),
+            "source_receipt_ref": str(
+                summary.get("ai_agent_single_writer_source_receipt_ref") or ""
+            ),
+            "source_verifier": str(
+                summary.get("ai_agent_single_writer_source_verifier") or ""
+            ),
+            "candidate_op_id": str(
+                summary.get("ai_agent_single_writer_candidate_op_id") or ""
+            ),
+            "check_mode_op_id": str(
+                summary.get("ai_agent_single_writer_check_mode_op_id") or ""
+            ),
+            "apply_op_id": str(
+                summary.get("ai_agent_single_writer_apply_op_id") or ""
+            ),
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+        }
+        single_writer_work_item["runtime_evidence_refs"] = [
+            "automation-run:"
+            + single_writer_work_item["completion_evidence"]["automation_run_id"],
+            "candidate:"
+            + single_writer_work_item["completion_evidence"]["candidate_op_id"],
+            "check-mode:"
+            + single_writer_work_item["completion_evidence"]["check_mode_op_id"],
+            "apply:"
+            + single_writer_work_item["completion_evidence"]["apply_op_id"],
+            single_writer_work_item["completion_evidence"]["source_receipt_ref"],
+        ]
+    elif observed_single_writer_controls or single_writer_blockers:
+        single_writer_work_item["status"] = "in_progress"
     priority_rank = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
     for item in work_items:
         item_id = str(item["id"])
@@ -7764,6 +7858,9 @@ def apply_ai_automation_live_closure_readbacks(
     database_identity_boundary = _dict(
         executor_trust_boundary.get("database_identity_boundary")
     )
+    autonomous_single_writer = _dict(
+        runtime_control.get("autonomous_single_writer")
+    )
     strict_runtime_closed = bool(
         strict_runtime.get("schema_version")
         == "ai_agent_strict_runtime_completion_v2"
@@ -7889,6 +7986,46 @@ def apply_ai_automation_live_closure_readbacks(
         rollups["ai_agent_executor_trust_boundary_runtime_closed_count"] = (
             1 if trust_boundary_closed else 0
         )
+        single_writer_controls = _dict(
+            autonomous_single_writer.get("controls")
+        )
+        single_writer_closed = bool(
+            autonomous_single_writer.get("schema_version")
+            == "awoooi_autonomous_single_writer_readback_v1"
+            and autonomous_single_writer.get("closed") is True
+            and str(autonomous_single_writer.get("automation_run_id") or "")
+            == str(strict_runtime.get("automation_run_id") or "")
+            and strict_runtime_closed
+        )
+        summary["ai_agent_single_writer_runtime_closed"] = single_writer_closed
+        summary["ai_agent_single_writer_controls"] = single_writer_controls
+        summary["ai_agent_single_writer_active_blockers"] = list(
+            autonomous_single_writer.get("active_blockers") or []
+        )
+        summary["ai_agent_single_writer_automation_run_id"] = str(
+            autonomous_single_writer.get("automation_run_id") or ""
+        )
+        summary["ai_agent_single_writer_router_source_sha"] = str(
+            autonomous_single_writer.get("router_source_sha") or ""
+        )
+        summary["ai_agent_single_writer_source_receipt_ref"] = str(
+            autonomous_single_writer.get("source_receipt_ref") or ""
+        )
+        summary["ai_agent_single_writer_source_verifier"] = str(
+            autonomous_single_writer.get("source_verifier") or ""
+        )
+        summary["ai_agent_single_writer_candidate_op_id"] = str(
+            autonomous_single_writer.get("candidate_op_id") or ""
+        )
+        summary["ai_agent_single_writer_check_mode_op_id"] = str(
+            autonomous_single_writer.get("check_mode_op_id") or ""
+        )
+        summary["ai_agent_single_writer_apply_op_id"] = str(
+            autonomous_single_writer.get("apply_op_id") or ""
+        )
+        rollups["ai_agent_single_writer_runtime_closed_count"] = (
+            1 if single_writer_closed else 0
+        )
 
     executor = _dict(executor_readback)
     executor_rollups = _dict(executor.get("rollups"))
diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py
index 345885210..d3c0d0839 100644
--- a/apps/api/src/services/awooop_ansible_audit_service.py
+++ b/apps/api/src/services/awooop_ansible_audit_service.py
@@ -8,6 +8,7 @@ allowlisted playbooks with check-mode support can move from dry-run to apply.
 from __future__ import annotations
 
 import json
+import os
 from typing import Any
 from uuid import UUID, uuid4
 
@@ -525,12 +526,31 @@ def build_ansible_decision_audit_payload(
         return None
 
     incident_id = str(incident_payload.get("incident_id") or "")
+    project_id = str(incident_payload.get("project_id") or "awoooi")
+    affected_services = [
+        str(value)
+        for value in incident_payload.get("affected_services") or []
+        if str(value).strip()
+    ]
+    namespaces = sorted({
+        str((signal.get("labels") or {}).get("namespace") or "").strip()
+        for signal in incident_payload.get("signals") or []
+        if isinstance(signal, dict)
+        and str((signal.get("labels") or {}).get("namespace") or "").strip()
+    })
+    proposal_risk_level = str(proposal_data.get("risk_level") or "").strip().lower()
+    idempotency_key = f"ansible-candidate:{project_id}:{incident_id}"
     input_payload = {
         "incident_id": incident_id,
-        "project_id": str(incident_payload.get("project_id") or "awoooi"),
+        "project_id": project_id,
         "executor": "ansible",
         "execution_backend": "ansible",
+        "single_writer_executor": "awoooi-ansible-executor-broker",
         "decision_path": decision_path,
+        "idempotency_key": idempotency_key,
+        "router_source_sha": os.getenv("AWOOOI_BUILD_COMMIT_SHA", "")
+        .strip()
+        .lower(),
         "check_mode": True,
         "apply_enabled": False,
         "approval_required": True,
@@ -551,12 +571,35 @@ def build_ansible_decision_audit_payload(
             for row in candidates[:5]
         ],
         "proposal_source": proposal_data.get("source", ""),
-        "proposal_risk_level": proposal_data.get("risk_level", ""),
+        "proposal_risk_level": proposal_risk_level,
         "proposal_action_preview": str(
             proposal_data.get("action")
             or proposal_data.get("kubectl_command")
             or ""
         )[:240],
+        "target_selector": {
+            "schema_version": "ai_decision_target_selector_v1",
+            "project_id": project_id,
+            "incident_id": incident_id,
+            "affected_services": affected_services,
+            "namespaces": namespaces,
+            "catalog_ids": [str(row["catalog_id"]) for row in candidates[:5]],
+            "inventory_hosts": sorted({
+                str(host)
+                for row in candidates[:5]
+                for host in row.get("inventory_hosts") or []
+            }),
+        },
+        "source_truth_diff": {
+            "required_before_apply": True,
+            "producer": "awooop_ansible_check_mode_worker",
+            "receipt_status": "pending_check_mode",
+        },
+        "risk_policy_decision": {
+            "risk_level": proposal_risk_level,
+            "controlled_apply_candidate": proposal_risk_level in {"low", "medium", "high"},
+            "critical_break_glass_required": proposal_risk_level == "critical",
+        },
     }
     controlled_queue = decision_path == "repair_candidate_controlled_queue"
     output_payload = {
@@ -610,6 +653,14 @@ async def record_ansible_decision_audit(
     payload["input"]["automation_run_id"] = candidate_op_id
     try:
         async with get_db_context(str(project_id)) as db:
+            await db.execute(
+                text("""
+                    SELECT pg_advisory_xact_lock(
+                        hashtextextended(:idempotency_key, 0)
+                    )
+                """),
+                {"idempotency_key": payload["input"]["idempotency_key"]},
+            )
             existing = await db.execute(
                 text("""
                     SELECT op_id
@@ -617,6 +668,7 @@ async def record_ansible_decision_audit(
                     WHERE operation_type = 'ansible_candidate_matched'
                       AND coalesce(incident_id::text, input ->> 'incident_id') = :incident_id
                       AND input ->> 'executor' = 'ansible'
+                      AND input ->> 'decision_path' = 'repair_candidate_controlled_queue'
                     LIMIT 1
                 """),
                 {"incident_id": incident_id},
diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py
index 73d62c5b3..a80d652cb 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -238,6 +238,12 @@ def build_ansible_check_mode_claim_input(
     safe = _safe_candidate(candidate_input)
     incident_id = _incident_id_from_payload(candidate_input)
     controlled_apply_allowed, controlled_apply_blocker = _controlled_apply_allowed(safe)
+    proposal_risk_level = str(
+        candidate_input.get("proposal_risk_level") or ""
+    ).strip().lower()
+    if proposal_risk_level == "critical":
+        controlled_apply_allowed = False
+        controlled_apply_blocker = "critical_break_glass_required"
     return {
         "automation_run_id": str(source_candidate_op_id),
         "incident_id": incident_id,
@@ -251,6 +257,20 @@ def build_ansible_check_mode_claim_input(
         "approval_required_before_apply": not controlled_apply_allowed,
         "controlled_apply_allowed": controlled_apply_allowed,
         "controlled_apply_blocker": controlled_apply_blocker,
+        "candidate_idempotency_key": str(
+            candidate_input.get("idempotency_key") or ""
+        ),
+        "apply_idempotency_key": (
+            f"ansible-apply:{source_candidate_op_id}:{safe['catalog_id']}"
+        ),
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+        "decision_path": str(candidate_input.get("decision_path") or ""),
+        "router_source_sha": str(candidate_input.get("router_source_sha") or "")
+        .strip()
+        .lower(),
+        "target_selector": candidate_input.get("target_selector") or {},
+        "source_truth_diff": candidate_input.get("source_truth_diff") or {},
+        "risk_policy_decision": candidate_input.get("risk_policy_decision") or {},
         "source_candidate_op_id": source_candidate_op_id,
         "catalog_id": safe["catalog_id"],
         "playbook_path": safe["playbook_path"],
@@ -3004,7 +3024,48 @@ async def run_controlled_apply_for_claim(
         requested_timeout_seconds=timeout_seconds,
     )
 
+    apply_idempotency_key = str(
+        claim.input_payload.get("apply_idempotency_key")
+        or f"ansible-apply:{claim.source_candidate_op_id}:{claim.catalog_id}"
+    )
     async with get_db_context(project_id) as db:
+        await db.execute(
+            text("""
+                SELECT pg_advisory_xact_lock(
+                    hashtextextended(:idempotency_key, 0)
+                )
+            """),
+            {"idempotency_key": apply_idempotency_key},
+        )
+        existing = await db.execute(
+            text("""
+                SELECT op_id, status
+                FROM automation_operation_log
+                WHERE operation_type = 'ansible_apply_executed'
+                  AND parent_op_id = CAST(:parent_op_id AS uuid)
+                ORDER BY created_at ASC
+                LIMIT 1
+            """),
+            {"parent_op_id": claim.op_id},
+        )
+        existing_row = existing.mappings().first()
+        if existing_row is not None:
+            claim.input_payload["apply_duplicate_suppressed"] = True
+            claim.input_payload["existing_apply_op_id"] = str(
+                existing_row.get("op_id") or ""
+            )
+            claim.input_payload["existing_apply_status"] = str(
+                existing_row.get("status") or ""
+            )
+            logger.warning(
+                "ansible_controlled_apply_duplicate_suppressed",
+                source_candidate_op_id=claim.source_candidate_op_id,
+                check_mode_op_id=claim.op_id,
+                existing_apply_op_id=claim.input_payload["existing_apply_op_id"],
+                existing_apply_status=claim.input_payload["existing_apply_status"],
+                idempotency_key=apply_idempotency_key,
+            )
+            return None
         inserted = await db.execute(
             text("""
                 INSERT INTO automation_operation_log (
@@ -3042,6 +3103,13 @@ async def run_controlled_apply_for_claim(
                     "apply_enabled": True,
                     "approval_required_before_apply": False,
                     "controlled_apply_allowed": True,
+                    "apply_idempotency_key": apply_idempotency_key,
+                    "single_writer_executor": "awoooi-ansible-executor-broker",
+                    "router_source_sha": str(
+                        claim.input_payload.get("router_source_sha") or ""
+                    )
+                    .strip()
+                    .lower(),
                     "runtime_stage_receipts": (
                         build_ansible_pre_apply_runtime_stage_receipts(claim)
                     ),
@@ -3322,6 +3390,7 @@ async def run_pending_check_modes_once(
     apply_completed = 0
     apply_failed = 0
     apply_blocked = 0
+    apply_duplicate_suppressed = 0
     capability_issued = 0
     capability_revoked = 0
     capability_revoke_failed = 0
@@ -3355,8 +3424,12 @@ async def run_pending_check_modes_once(
                 project_id=project_id,
             )
             if apply_result is None:
-                apply_blocked += 1
-                terminal_status = "check_mode_passed_apply_blocked_by_policy"
+                if claim.input_payload.get("apply_duplicate_suppressed") is True:
+                    apply_duplicate_suppressed += 1
+                    terminal_status = "controlled_apply_duplicate_suppressed"
+                else:
+                    apply_blocked += 1
+                    terminal_status = "check_mode_passed_apply_blocked_by_policy"
             else:
                 apply_completed += 1
                 if apply_result.returncode != 0:
@@ -3426,6 +3499,7 @@ async def run_pending_check_modes_once(
         "apply_completed": apply_completed,
         "apply_failed": apply_failed,
         "apply_blocked": apply_blocked,
+        "apply_duplicate_suppressed": apply_duplicate_suppressed,
         "capability_issued": capability_issued,
         "capability_revoked": capability_revoked,
         "capability_expired": expired_capability_count,
diff --git a/apps/api/src/services/decision_manager.py b/apps/api/src/services/decision_manager.py
index f14af5149..e59d3a3f9 100644
--- a/apps/api/src/services/decision_manager.py
+++ b/apps/api/src/services/decision_manager.py
@@ -1823,26 +1823,9 @@ class DecisionManager:
                 token.proposal_data["auto_approve_reason"] = auto_decision.reason_detail
                 await self._save_token(token)
 
-                try:
-                    from src.services.awooop_ansible_audit_service import (
-                        record_ansible_decision_audit as _record_ansible_decision_audit,
-                    )
-
-                    _fire_and_forget(
-                        _record_ansible_decision_audit(
-                            incident=incident,
-                            proposal_data=token.proposal_data,
-                            decision_path="auto_execute",
-                            not_used_reason=(
-                                "auto_execute selected existing executor path; "
-                                "Ansible check-mode is not wired yet"
-                            ),
-                        )
-                    )
-                except Exception as _ansible_audit_err:
-                    logger.debug("ansible_decision_audit_schedule_error", error=str(_ansible_audit_err))
-
-                # 觸發自動執行 (非阻塞)
+                # The active auto-execute path owns the durable controlled-queue
+                # receipt. Do not write an audit-only candidate here because it
+                # would race with, and suppress, the claimable broker candidate.
                 _fire_and_forget(
                     self._auto_execute(incident, token)
                 )
@@ -1890,13 +1873,148 @@ class DecisionManager:
         return token
 
     async def _auto_execute(self, incident: Incident, token: "DecisionToken") -> None:
-        """
-        ADR-030 Phase 4: 自動執行已批准的操作
+        """Route an approved AI decision to the single-writer executor queue."""
 
-        僅當 AutoApprovePolicy 判斷可自動執行時呼叫
-        執行後發 Telegram 結果通知 (統帥要求: 修復結果對應同一告警)
-        2026-04-09 Claude Sonnet 4.6 Asia/Taipei
+        proposal_data = token.proposal_data or {}
+        token.proposal_data = proposal_data
+        action = str(
+            proposal_data.get("kubectl_command")
+            or proposal_data.get("action")
+            or ""
+        ).strip()
+        labels = incident.signals[0].labels if incident.signals else {}
+        host_type = str(labels.get("host_type") or "").lower()
+
+        if host_type == "bare_metal" and action.lower().startswith("kubectl"):
+            token.state = DecisionState.READY
+            proposal_data.update({
+                "auto_executed": False,
+                "direct_runtime_execution_performed": False,
+                "automation_state": "blocked_with_safe_next_action",
+                "blocked_reason": (
+                    "host_type=bare_metal but the proposed action uses kubectl; "
+                    "a host-scoped allowlisted PlayBook target is required"
+                ),
+                "safe_next_action": "queue_host_scoped_playbook_candidate",
+            })
+            await self._save_token(token)
+            logger.warning(
+                "ai_decision_wrong_domain_blocked_before_queue",
+                incident_id=incident.incident_id,
+                host_type=host_type,
+                action=action[:120],
+            )
+            return
+
+        risk_value = proposal_data.get("risk_level")
+        risk_level = str(
+            getattr(risk_value, "value", risk_value) or ""
+        ).strip().lower()
+        if risk_level == "critical":
+            token.state = DecisionState.READY
+            proposal_data.update({
+                "auto_executed": False,
+                "direct_runtime_execution_performed": False,
+                "automation_state": "critical_break_glass_required",
+                "blocked_reason": "critical_break_glass_required",
+                "safe_next_action": "open_bounded_break_glass_contract",
+            })
+            await self._save_token(token)
+            logger.warning(
+                "ai_decision_critical_break_glass_required",
+                incident_id=incident.incident_id,
+            )
+            return
+
+        if not action or proposal_data.get("suggested_action") == "NO_ACTION":
+            token.state = DecisionState.READY
+            proposal_data.update({
+                "auto_executed": False,
+                "direct_runtime_execution_performed": False,
+                "automation_state": "verified_no_write_terminal",
+                "blocked_reason": "no_runtime_action_selected",
+            })
+            await self._save_token(token)
+            return
+
+        try:
+            from src.jobs.awooop_ansible_candidate_backfill_job import (
+                enqueue_ai_decision_ansible_candidate,
+            )
+
+            handoff = await enqueue_ai_decision_ansible_candidate(
+                incident=incident,
+                proposal_data=proposal_data,
+                project_id=str(getattr(incident, "project_id", None) or "awoooi"),
+            )
+        except Exception as exc:
+            logger.warning(
+                "ai_decision_controlled_executor_queue_failed",
+                incident_id=incident.incident_id,
+                error_type=type(exc).__name__,
+            )
+            handoff = {
+                "schema_version": "ai_decision_controlled_executor_handoff_v1",
+                "status": "controlled_executor_queue_failed",
+                "queued": False,
+                "side_effect_performed": False,
+                "single_writer_executor": "awoooi-ansible-executor-broker",
+                "active_blockers": ["controlled_executor_queue_failed"],
+            }
+
+        queued = handoff.get("queued") is True
+        proposal_data.update({
+            "auto_executed": False,
+            "direct_runtime_execution_performed": False,
+            "approval_execution_invoked": False,
+            "check_mode_required_before_apply": True,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "controlled_executor_handoff": handoff,
+            "automation_run_id": str(handoff.get("automation_run_id") or ""),
+            "automation_state": (
+                "controlled_check_mode_queued"
+                if queued
+                else "blocked_with_safe_next_action"
+            ),
+            "safe_next_action": (
+                "awooop_ansible_check_mode_worker_claims_candidate"
+                if queued
+                else "repair_candidate_backfill_or_playbook_coverage_gap"
+            ),
+        })
+        if queued:
+            token.state = DecisionState.EXECUTING
+            token.error = None
+            proposal_data.pop("blocked_reason", None)
+        else:
+            token.state = DecisionState.READY
+            blockers = handoff.get("active_blockers") or [handoff.get("status")]
+            proposal_data["blocked_reason"] = ",".join(
+                str(value) for value in blockers if value
+            )
+        await self._save_token(token)
+        logger.info(
+            "ai_decision_controlled_executor_handoff",
+            incident_id=incident.incident_id,
+            status=handoff.get("status"),
+            queued=queued,
+            automation_run_id=handoff.get("automation_run_id"),
+            side_effect_performed=False,
+        )
+
+    async def _legacy_auto_execute_disabled(
+        self,
+        incident: Incident,
+        token: "DecisionToken",
+    ) -> None:
         """
+        Superseded direct executor retained temporarily for bounded removal.
+
+        It is intentionally fail-closed and must never produce a side effect.
+        """
+        raise RuntimeError(
+            "decision_manager_direct_execution_superseded_by_single_writer_broker"
+        )
         action = token.proposal_data.get("kubectl_command", "") or token.proposal_data.get("action", "")
         _alert_labels = incident.signals[0].labels if incident.signals else {}
         _host_type_for_domain_guard = (_alert_labels.get("host_type") or "").lower()
diff --git a/apps/api/src/services/executor_trust_boundary_readback.py b/apps/api/src/services/executor_trust_boundary_readback.py
index e75daa202..d736f2048 100644
--- a/apps/api/src/services/executor_trust_boundary_readback.py
+++ b/apps/api/src/services/executor_trust_boundary_readback.py
@@ -33,6 +33,15 @@ _REQUIRED_TRUE_FIELDS = (
     "broker_ssh_egress_allowlisted",
     "legacy_namespace_egress_allow_absent",
 )
+_SINGLE_WRITER_REQUIRED_TRUE_FIELDS = (
+    "decision_manager_queue_only",
+    "webhook_router_queue_only",
+    "failure_watcher_queue_only",
+    "auto_approved_direct_execution_blocked",
+    "candidate_idempotency_contract_verified",
+    "apply_idempotency_contract_verified",
+    "critical_break_glass_contract_verified",
+)
 _DB_WORKLOAD_IDS = ("api", "worker", "broker")
 _ROLE_FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$")
 _cache: tuple[float, dict[str, Any]] | None = None
@@ -270,6 +279,41 @@ def _build_database_identity_boundary(
     }
 
 
+def _build_autonomous_single_writer_source_boundary(
+    receipt: Mapping[str, str],
+    *,
+    receipt_ref: str,
+    source_sha_matches_deployment: bool,
+) -> dict[str, Any]:
+    controls = {
+        field: _truthy(receipt.get(field))
+        for field in _SINGLE_WRITER_REQUIRED_TRUE_FIELDS
+    }
+    blockers = [
+        f"{field}_not_verified"
+        for field, verified in controls.items()
+        if verified is not True
+    ]
+    verifier = receipt.get("single_writer_source_verifier") or None
+    if verifier != "cd_tests_and_production_source_sha_v1":
+        blockers.append("single_writer_source_verifier_not_verified")
+    if not source_sha_matches_deployment:
+        blockers.append("single_writer_source_sha_not_deployed")
+    verified = not blockers
+    return {
+        "schema_version": "awoooi_autonomous_single_writer_source_boundary_v1",
+        "status": "verified_ready" if verified else "in_progress",
+        "receipt_ref": receipt_ref,
+        "verifier": verifier,
+        "source_boundary_verified": verified,
+        "source_sha_matches_deployment": source_sha_matches_deployment,
+        "controls": controls,
+        "active_blockers": blockers,
+        "writes_on_read": False,
+        "secret_values_read": False,
+    }
+
+
 def build_executor_trust_boundary_readback(
     data: Mapping[str, Any] | None,
     *,
@@ -296,10 +340,21 @@ def build_executor_trust_boundary_readback(
             blockers.append(f"{field}_not_verified")
 
     ready = not blockers
+    source_sha_matches_deployment = bool(
+        normalized_deployed_sha
+        and verified_source_sha == normalized_deployed_sha
+    )
     database_identity_boundary = _build_database_identity_boundary(
         receipt,
         receipt_ref=receipt_ref,
     )
+    autonomous_single_writer_source_boundary = (
+        _build_autonomous_single_writer_source_boundary(
+            receipt,
+            receipt_ref=receipt_ref,
+            source_sha_matches_deployment=source_sha_matches_deployment,
+        )
+    )
     full_workload_boundary_verified = bool(
         ready
         and database_identity_boundary["db_identity_isolated"] is True
@@ -323,10 +378,7 @@ def build_executor_trust_boundary_readback(
         "receipt_schema_version": receipt.get("schema_version") or None,
         "verified_source_sha": verified_source_sha or None,
         "deployed_source_sha": normalized_deployed_sha or None,
-        "source_sha_matches_deployment": bool(
-            normalized_deployed_sha
-            and verified_source_sha == normalized_deployed_sha
-        ),
+        "source_sha_matches_deployment": source_sha_matches_deployment,
         "verified_at": receipt.get("verified_at") or None,
         "workflow_run_id": receipt.get("workflow_run_id") or None,
         "verifier": receipt.get("verifier") or None,
@@ -340,6 +392,9 @@ def build_executor_trust_boundary_readback(
             for field in _REQUIRED_TRUE_FIELDS
         },
         "database_identity_boundary": database_identity_boundary,
+        "autonomous_single_writer_source_boundary": (
+            autonomous_single_writer_source_boundary
+        ),
         "active_blockers": blockers,
         "full_boundary_active_blockers": [
             *blockers,
diff --git a/apps/api/src/services/failure_watcher.py b/apps/api/src/services/failure_watcher.py
index 304a083af..e6b537e67 100644
--- a/apps/api/src/services/failure_watcher.py
+++ b/apps/api/src/services/failure_watcher.py
@@ -216,44 +216,27 @@ class FailureWatcherService(IFailureWatcher):
             result["risk_level"] = "MEDIUM"
 
         if risk_level != "CRITICAL" and RISK_LEVELS.get(risk_level, {}).get("auto_repair"):
-            # 自動修復 (Phase 18.3: 傳入完整 failure_data)
-            success, repair_result = await self.execute_auto_repair(
+            # Executor failures must not recursively invoke a second writer.
+            # Queue the repair context for the controlled broker path instead.
+            await self._queue_ai_repair_followup(
                 audit_log_id=audit_log_id,
-                repair_strategy=analysis["suggested_repair"],
-                failure_data=failure_data,
+                analysis=analysis,
+                reason=(
+                    "executor failure routed to single-writer retry/repair queue; "
+                    "a new check-mode receipt is required before apply"
+                ),
             )
-            result["repair_attempted"] = True
-            result["repair_result"] = repair_result
-            result["next_action"] = "auto_repaired" if success else "escalate"
+            result["repair_attempted"] = False
+            result["repair_result"] = "single_writer_retry_or_repair_queued"
+            result["next_action"] = "ai_retry_queued"
 
             # 更新 AuditLog
             await self._update_audit_log_classification(
                 audit_log_id=audit_log_id,
                 classification=analysis["classification"],
-                auto_repair_attempted=True,
-                auto_repair_result=repair_result,
+                auto_repair_attempted=False,
+                auto_repair_result="single_writer_retry_or_repair_queued",
             )
-
-            if success:
-                # P0-1 補充: 記錄全域修復動作 (ADR-040)
-                from src.services.global_repair_cooldown import record_global_repair_action
-                await record_global_repair_action()
-
-                # 推送揭露通知 (自動修復成功)
-                await self._push_repair_notification(
-                    audit_log_id=audit_log_id,
-                    repair_result=repair_result,
-                    auto=True,
-                )
-            else:
-                # 失敗後排入 AI controlled retry / rollback，不再直接轉人工。
-                result["risk_level"] = "MEDIUM"
-                await self._queue_ai_repair_followup(
-                    audit_log_id=audit_log_id,
-                    analysis=analysis,
-                    reason="AI 受控自動修復失敗，已排入 rollback / transport / PlayBook 修復重試",
-                )
-                result["next_action"] = "ai_retry_queued"
         else:
             # CRITICAL: break-glass / hard blocker，不執行寫入。
             await self._queue_ai_repair_followup(
@@ -322,6 +305,18 @@ class FailureWatcherService(IFailureWatcher):
         audit_log_id: str,
         repair_strategy: str,
         failure_data: dict | None = None,
+    ) -> tuple[bool, str]:
+        """Fail closed when legacy code tries to bypass the single writer."""
+
+        raise RuntimeError(
+            "failure_watcher_direct_execution_disabled_use_single_writer_broker"
+        )
+
+    async def _legacy_execute_auto_repair_disabled(
+        self,
+        audit_log_id: str,
+        repair_strategy: str,
+        failure_data: dict | None = None,
     ) -> tuple[bool, str]:
         """
         執行自動修復 (僅限 LOW 風險)
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 298a61398..fb31c7cd3 100644
--- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py
+++ b/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()
 
diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py
index c06452eee..f7aa7fce7 100644
--- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py
+++ b/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,
 ):
diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py
index 49315047a..a100bd560 100644
--- a/apps/api/tests/test_awooop_truth_chain_service.py
+++ b/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)
 
diff --git a/apps/api/tests/test_cs1_auto_execute.py b/apps/api/tests/test_cs1_auto_execute.py
index 4808d6705..e1c7762db 100644
--- a/apps/api/tests/test_cs1_auto_execute.py
+++ b/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)
diff --git a/apps/api/tests/test_cs3_auto_execute.py b/apps/api/tests/test_cs3_auto_execute.py
index dd5744057..06759e582 100644
--- a/apps/api/tests/test_cs3_auto_execute.py
+++ b/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
diff --git a/apps/api/tests/test_decision_manager_bare_metal_kubectl_guard.py b/apps/api/tests/test_decision_manager_bare_metal_kubectl_guard.py
index 2b3778e2f..d7cc82904 100644
--- a/apps/api/tests/test_decision_manager_bare_metal_kubectl_guard.py
+++ b/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
diff --git a/apps/api/tests/test_executor_trust_boundary_readback.py b/apps/api/tests/test_executor_trust_boundary_readback.py
index 82bdac559..ade763657 100644
--- a/apps/api/tests/test_executor_trust_boundary_readback.py
+++ b/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:
diff --git a/apps/api/tests/test_failure_watcher.py b/apps/api/tests/test_failure_watcher.py
index 62d54506a..0e542fe17 100644
--- a/apps/api/tests/test_failure_watcher.py
+++ b/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")
diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py
index 171539536..c476893a0 100644
--- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py
+++ b/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 (