Add maker-checker approval gate for high-risk actions

README.md

@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-19) — Maker-Checker Approval Gate](#whats-new-2026-06-19--maker-checker-approval-gate)
 - [What's new (2026-06-19) — Plugin SDK](#whats-new-2026-06-19--plugin-sdk)
 - [What's new (2026-06-19) — MCP Structured Output](#whats-new-2026-06-19--mcp-structured-output)
 - [What's new (2026-06-19) — Tweened Drag](#whats-new-2026-06-19--tweened-drag)
@@ -84,6 +85,12 @@
 
 ---
 
+## What's new (2026-06-19) — Maker-Checker Approval Gate
+
+Segregation of duties for high-risk steps. Full reference: [`docs/source/Eng/doc/new_features/v32_features_doc.rst`](docs/source/Eng/doc/new_features/v32_features_doc.rst).
+
+- **`ApprovalGate`** (`AC_approval_request` / `AC_approval_approve` / `AC_approval_reject` / `AC_approval_status`, `ac_*`): a *maker* files a high-risk action and gets a token; a *checker* — required to be a **different** principal — approves or rejects it; the action proceeds only once `is_approved` is true. State is an optional shared JSON file so the dispatcher and the human approver can run as separate processes. Pure-stdlib, SOC2-style four-eyes control.
+
 ## What's new (2026-06-19) — Plugin SDK
 
 Third-party `AC_*` commands via entry points. Full reference: [`docs/source/Eng/doc/new_features/v31_features_doc.rst`](docs/source/Eng/doc/new_features/v31_features_doc.rst).

README/README_zh-CN.md

@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-19) — Maker-Checker 审批闸门](#本次更新-2026-06-19--maker-checker-审批闸门)
 - [本次更新 (2026-06-19) — Plugin SDK](#本次更新-2026-06-19--plugin-sdk)
 - [本次更新 (2026-06-19) — MCP 结构化输出](#本次更新-2026-06-19--mcp-结构化输出)
 - [本次更新 (2026-06-19) — 缓动拖拽](#本次更新-2026-06-19--缓动拖拽)
@@ -83,6 +84,12 @@
 
 ---
 
+## 本次更新 (2026-06-19) — Maker-Checker 审批闸门
+
+高风险步骤的职责分离。完整参考:[`docs/source/Zh/doc/new_features/v32_features_doc.rst`](../docs/source/Zh/doc/new_features/v32_features_doc.rst)。
+
+- **`ApprovalGate`**(`AC_approval_request` / `AC_approval_approve` / `AC_approval_reject` / `AC_approval_status`、`ac_*`):由 *maker* 提出高风险动作并取得 token;*checker*(必须为**不同**主体)核准或驳回;只有在 `is_approved` 为真后动作才继续。状态为选用的共享 JSON 文件,让派发器与人工审批者可分属不同进程。纯标准库,SOC2 式四眼原则控制。
+
 ## 本次更新 (2026-06-19) — Plugin SDK
 
 通过 entry points 注册第三方 `AC_*` 指令。完整参考:[`docs/source/Zh/doc/new_features/v31_features_doc.rst`](../docs/source/Zh/doc/new_features/v31_features_doc.rst)。

README/README_zh-TW.md

@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-19) — Maker-Checker 審批閘門](#本次更新-2026-06-19--maker-checker-審批閘門)
 - [本次更新 (2026-06-19) — Plugin SDK](#本次更新-2026-06-19--plugin-sdk)
 - [本次更新 (2026-06-19) — MCP 結構化輸出](#本次更新-2026-06-19--mcp-結構化輸出)
 - [本次更新 (2026-06-19) — 緩動拖曳](#本次更新-2026-06-19--緩動拖曳)
@@ -83,6 +84,12 @@
 
 ---
 
+## 本次更新 (2026-06-19) — Maker-Checker 審批閘門
+
+高風險步驟的職責分離。完整參考:[`docs/source/Zh/doc/new_features/v32_features_doc.rst`](../docs/source/Zh/doc/new_features/v32_features_doc.rst)。
+
+- **`ApprovalGate`**(`AC_approval_request` / `AC_approval_approve` / `AC_approval_reject` / `AC_approval_status`、`ac_*`):由 *maker* 提出高風險動作並取得 token;*checker*(必須為**不同**主體)核准或駁回;只有在 `is_approved` 為真後動作才繼續。狀態為選用的共用 JSON 檔,讓派發器與人工審批者可分屬不同程序。純標準函式庫,SOC2 式四眼原則控制。
+
 ## 本次更新 (2026-06-19) — Plugin SDK
 
 透過 entry points 註冊第三方 `AC_*` 指令。完整參考:[`docs/source/Zh/doc/new_features/v31_features_doc.rst`](../docs/source/Zh/doc/new_features/v31_features_doc.rst)。

docs/source/Eng/doc/new_features/v32_features_doc.rst

@@ -0,0 +1,52 @@
+Maker-Checker Approval Gate
+===========================
+
+Some automation steps are too consequential to fire on one party's say-so —
+deleting production data, wiring money, promoting a release. ``ApprovalGate``
+adds a **segregation of duties** control: a *maker* files a request and gets a
+token; a *checker*, who must be a **different** principal, approves or rejects
+it; the action proceeds only once the token is approved.
+
+State is an optional JSON file, so the maker (e.g. a CI dispatcher) and the
+checker (e.g. a human approver) can run as separate processes. The module is
+pure standard library and imports no ``PySide6``; tokens use :mod:`secrets`.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import ApprovalGate
+
+    gate = ApprovalGate("approvals.json")          # shared across processes
+    token = gate.request("delete prod table", requester="alice")
+
+    # Self-approval is refused — the checker must differ from the maker.
+    gate.approve(token, "alice")    # -> False
+    gate.approve(token, "bob")      # -> True
+
+    if gate.is_approved(token):
+        run_high_risk_action()
+
+``reject(token, approver)`` blocks an action; a request that has already been
+decided cannot be re-decided. ``status(token)`` returns
+``pending`` / ``approved`` / ``rejected`` (or ``None`` for an unknown token),
+``get(token)`` returns the full record, and ``pending()`` lists every request
+still awaiting a decision.
+
+Executor commands
+-----------------
+
+================================ ===================================================
+Command                          Effect
+================================ ===================================================
+``AC_approval_request``          File a request for ``action``; returns ``{token}``.
+``AC_approval_approve``          Approve ``token`` as ``approver``; ``{approved}``.
+``AC_approval_reject``           Reject ``token`` as ``approver``; ``{rejected}``.
+``AC_approval_status``           Return ``{status, approved}`` to gate an action.
+================================ ===================================================
+
+Each command accepts an optional ``db`` path so a flow can persist requests to
+a shared JSON file. The same operations are exposed as MCP tools
+(``ac_approval_request`` / ``ac_approval_approve`` / ``ac_approval_reject`` /
+``ac_approval_status``) and as Script Builder commands under **Tools**.

docs/source/Eng/eng_index.rst

@@ -54,6 +54,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v29_features_doc
    doc/new_features/v30_features_doc
    doc/new_features/v31_features_doc
+   doc/new_features/v32_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

docs/source/Zh/doc/new_features/v32_features_doc.rst

@@ -0,0 +1,49 @@
+Maker-Checker 審批閘門
+======================
+
+有些自動化步驟後果太重大,不該由單一方說了算 —— 刪除正式環境資料、匯款、發布版
+本。``ApprovalGate`` 提供**職責分離(segregation of duties)**控制:由 *maker*
+提出請求並取得 token;*checker*(必須是**不同**的主體)核准或駁回;只有在 token
+被核准後動作才會繼續。
+
+狀態存於選用的 JSON 檔,因此 maker(例如 CI 派發器)與 checker(例如人工審批者)
+可分屬不同程序執行。本模組為純標準函式庫,不匯入 ``PySide6``;token 使用
+:mod:`secrets`。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import ApprovalGate
+
+    gate = ApprovalGate("approvals.json")          # 跨程序共用
+    token = gate.request("delete prod table", requester="alice")
+
+    # 自我核准會被拒絕 —— checker 必須與 maker 不同。
+    gate.approve(token, "alice")    # -> False
+    gate.approve(token, "bob")      # -> True
+
+    if gate.is_approved(token):
+        run_high_risk_action()
+
+``reject(token, approver)`` 會封鎖動作;已決議的請求無法再次決議。
+``status(token)`` 回傳 ``pending`` / ``approved`` / ``rejected``(未知 token 為
+``None``),``get(token)`` 回傳完整紀錄,``pending()`` 則列出所有仍待決議的請求。
+
+執行器指令
+----------
+
+================================ ===================================================
+指令                             效果
+================================ ===================================================
+``AC_approval_request``          為 ``action`` 提出請求;回傳 ``{token}``。
+``AC_approval_approve``          以 ``approver`` 核准 ``token``;``{approved}``。
+``AC_approval_reject``           以 ``approver`` 駁回 ``token``;``{rejected}``。
+``AC_approval_status``           回傳 ``{status, approved}`` 以閘控動作。
+================================ ===================================================
+
+每個指令都接受選用的 ``db`` 路徑,讓流程可將請求保存到共用 JSON 檔。相同操作亦提供
+為 MCP 工具(``ac_approval_request`` / ``ac_approval_approve`` /
+``ac_approval_reject`` / ``ac_approval_status``),以及 Script Builder 中 **Tools**
+分類下的指令。

docs/source/Zh/zh_index.rst

@@ -54,6 +54,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v29_features_doc
    doc/new_features/v30_features_doc
    doc/new_features/v31_features_doc
+   doc/new_features/v32_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

je_auto_control/__init__.py

@@ -203,6 +203,8 @@
 from je_auto_control.utils.plugin_sdk import (
     COMMANDS_GROUP, discover_plugins, load_plugins,
 )
+# Maker-checker approval gate (segregation of duties for high-risk actions)
+from je_auto_control.utils.governance import ApprovalGate
 # Background popup/interrupt watchdog (unattended automation)
 from je_auto_control.utils.watchdog import (
     PopupWatchdog, WatchdogRule, default_popup_watchdog,
@@ -638,6 +640,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "describe_step", "generate_sop", "write_sop",
     "easing_names", "tween_drag", "tween_points",
     "COMMANDS_GROUP", "discover_plugins", "load_plugins",
+    "ApprovalGate",
     # MCP server
     "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt",
     "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool",

je_auto_control/gui/script_builder/command_schema.py

@@ -693,6 +693,41 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None:
                           default="je_auto_control.commands"),),
         description="Discover + register third-party plugin commands.",
     ))
+    specs.append(CommandSpec(
+        "AC_approval_request", "Tools", "Approval: Request",
+        fields=(
+            FieldSpec("action", FieldType.STRING),
+            FieldSpec("requester", FieldType.STRING, optional=True),
+            FieldSpec("db", FieldType.STRING, optional=True),
+        ),
+        description="Maker-checker: file a high-risk action for approval.",
+    ))
+    specs.append(CommandSpec(
+        "AC_approval_approve", "Tools", "Approval: Approve",
+        fields=(
+            FieldSpec("token", FieldType.STRING),
+            FieldSpec("approver", FieldType.STRING),
+            FieldSpec("db", FieldType.STRING, optional=True),
+        ),
+        description="Approve a request (approver must differ from requester).",
+    ))
+    specs.append(CommandSpec(
+        "AC_approval_reject", "Tools", "Approval: Reject",
+        fields=(
+            FieldSpec("token", FieldType.STRING),
+            FieldSpec("approver", FieldType.STRING),
+            FieldSpec("db", FieldType.STRING, optional=True),
+        ),
+        description="Reject a request (approver must differ from requester).",
+    ))
+    specs.append(CommandSpec(
+        "AC_approval_status", "Tools", "Approval: Status",
+        fields=(
+            FieldSpec("token", FieldType.STRING),
+            FieldSpec("db", FieldType.STRING, optional=True),
+        ),
+        description="Report a request's status and approved flag.",
+    ))
     specs.append(CommandSpec(
         "AC_generate_sop", "Report", "Generate SOP Document",
         fields=(

je_auto_control/utils/executor/action_executor.py

@@ -2894,6 +2894,34 @@ def _load_plugins(group: str = "je_auto_control.commands") -> Dict[str, Any]:
     return {"loaded": load_plugins(group)}
 
 
+def _approval_request(action: str, requester: str = "",
+                      db: Optional[str] = None) -> Dict[str, Any]:
+    """Adapter: file a maker-checker approval request; return its token."""
+    from je_auto_control.utils.governance import ApprovalGate
+    return {"token": ApprovalGate(db).request(action, requester)}
+
+
+def _approval_approve(token: str, approver: str,
+                      db: Optional[str] = None) -> Dict[str, Any]:
+    """Adapter: approve a request as ``approver`` (must differ from maker)."""
+    from je_auto_control.utils.governance import ApprovalGate
+    return {"approved": ApprovalGate(db).approve(token, approver)}
+
+
+def _approval_reject(token: str, approver: str,
+                     db: Optional[str] = None) -> Dict[str, Any]:
+    """Adapter: reject a request as ``approver`` (must differ from maker)."""
+    from je_auto_control.utils.governance import ApprovalGate
+    return {"rejected": ApprovalGate(db).reject(token, approver)}
+
+
+def _approval_status(token: str, db: Optional[str] = None) -> Dict[str, Any]:
+    """Adapter: report the status and approved flag of a request token."""
+    from je_auto_control.utils.governance import ApprovalGate
+    gate = ApprovalGate(db)
+    return {"status": gate.status(token), "approved": gate.is_approved(token)}
+
+
 class Executor:
     """
     Executor
@@ -3125,6 +3153,10 @@ def __init__(self):
             "AC_tween_drag": _tween_drag,
             "AC_list_plugins": _list_plugins,
             "AC_load_plugins": _load_plugins,
+            "AC_approval_request": _approval_request,
+            "AC_approval_approve": _approval_approve,
+            "AC_approval_reject": _approval_reject,
+            "AC_approval_status": _approval_status,
             "AC_a11y_record_start": _a11y_record_start,
             "AC_a11y_record_stop": _a11y_record_stop,
             "AC_a11y_record_events": _a11y_record_events,

je_auto_control/utils/governance/__init__.py

@@ -0,0 +1,4 @@
+"""Governance: maker-checker approval gate for high-risk actions."""
+from je_auto_control.utils.governance.governance import ApprovalGate
+
+__all__ = ["ApprovalGate"]

je_auto_control/utils/governance/governance.py

@@ -0,0 +1,95 @@
+"""Maker-checker approval gate for high-risk automation actions.
+
+Some automation steps (deleting records, sending money, deploying) should not
+fire on one person's say-so. This implements a *segregation of duties* gate: a
+**maker** files a request and receives a token; a **checker** — who must be a
+different principal — approves or rejects it; the action proceeds only once
+``is_approved`` is true. State is an optional JSON file so the maker and checker
+can run as separate processes (CI dispatcher and a human approver).
+
+Pure standard library; imports no ``PySide6``. Tokens use :mod:`secrets`.
+"""
+import json
+import secrets
+import time
+from pathlib import Path
+from typing import Dict, List, Optional
+
+STATUS_PENDING = "pending"
+STATUS_APPROVED = "approved"
+STATUS_REJECTED = "rejected"
+
+
+class ApprovalGate:
+    """A maker-checker approval registry backed by an optional JSON file."""
+
+    def __init__(self, db_path: Optional[str] = None) -> None:
+        """Open the gate; ``db_path`` persists state across processes."""
+        self._path = Path(db_path) if db_path else None
+        self._items: Dict[str, Dict[str, object]] = self._load()
+
+    def _load(self) -> Dict[str, Dict[str, object]]:
+        if self._path is None or not self._path.is_file():
+            return {}
+        try:
+            data = json.loads(self._path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            return {}
+        return data if isinstance(data, dict) else {}
+
+    def _flush(self) -> None:
+        if self._path is None:
+            return
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._path.write_text(
+            json.dumps(self._items, ensure_ascii=False, indent=2),
+            encoding="utf-8")
+
+    def request(self, action: str, requester: str = "") -> str:
+        """File an approval request for ``action``; return its token."""
+        token = secrets.token_hex(8)
+        self._items[token] = {
+            "token": token, "action": action, "requester": requester,
+            "status": STATUS_PENDING, "approver": "", "created": time.time(),
+        }
+        self._flush()
+        return token
+
+    def _decide(self, token: str, approver: str, status: str) -> bool:
+        record = self._items.get(token)
+        if record is None or record["status"] != STATUS_PENDING:
+            return False
+        if approver and approver == record["requester"]:
+            return False  # segregation of duties: checker must differ from maker
+        record["status"] = status
+        record["approver"] = approver
+        self._flush()
+        return True
+
+    def approve(self, token: str, approver: str) -> bool:
+        """Approve ``token`` as ``approver`` (must differ from the requester)."""
+        return self._decide(token, approver, STATUS_APPROVED)
+
+    def reject(self, token: str, approver: str) -> bool:
+        """Reject ``token`` as ``approver`` (must differ from the requester)."""
+        return self._decide(token, approver, STATUS_REJECTED)
+
+    def status(self, token: str) -> Optional[str]:
+        """Return the status string for ``token``, or ``None`` if unknown."""
+        record = self._items.get(token)
+        return str(record["status"]) if record else None
+
+    def is_approved(self, token: str) -> bool:
+        """Return ``True`` only when ``token`` has been approved."""
+        record = self._items.get(token)
+        return record is not None and record["status"] == STATUS_APPROVED
+
+    def get(self, token: str) -> Optional[Dict[str, object]]:
+        """Return a copy of the request record for ``token``, or ``None``."""
+        record = self._items.get(token)
+        return dict(record) if record else None
+
+    def pending(self) -> List[Dict[str, object]]:
+        """Return copies of all requests still awaiting a decision."""
+        return [dict(r) for r in self._items.values()
+                if r["status"] == STATUS_PENDING]