From be6af5f82c8d81d76cb3e827db905bfa96a6cf13 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 19 Jun 2026 22:22:45 +0800 Subject: [PATCH] Add maker-checker approval gate for high-risk actions --- README.md | 7 ++ README/README_zh-CN.md | 7 ++ README/README_zh-TW.md | 7 ++ .../Eng/doc/new_features/v32_features_doc.rst | 52 ++++++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v32_features_doc.rst | 49 ++++++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 3 + .../gui/script_builder/command_schema.py | 35 +++++++ .../utils/executor/action_executor.py | 32 +++++++ je_auto_control/utils/governance/__init__.py | 4 + .../utils/governance/governance.py | 95 +++++++++++++++++++ .../utils/mcp_server/tools/_factories.py | 48 +++++++++- .../utils/mcp_server/tools/_handlers.py | 22 +++++ .../headless/test_governance_batch.py | 94 ++++++++++++++++++ 15 files changed, 456 insertions(+), 1 deletion(-) create mode 100644 docs/source/Eng/doc/new_features/v32_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v32_features_doc.rst create mode 100644 je_auto_control/utils/governance/__init__.py create mode 100644 je_auto_control/utils/governance/governance.py create mode 100644 test/unit_test/headless/test_governance_batch.py diff --git a/README.md b/README.md index 3fb07228..e22c0751 100644 --- a/README.md +++ b/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). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index 83844b58..be131988 100644 --- a/README/README_zh-CN.md +++ b/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)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 330afaa6..64b9a5dc 100644 --- a/README/README_zh-TW.md +++ b/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)。 diff --git a/docs/source/Eng/doc/new_features/v32_features_doc.rst b/docs/source/Eng/doc/new_features/v32_features_doc.rst new file mode 100644 index 00000000..ec1541b6 --- /dev/null +++ b/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**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 0cfff060..8cb1d0f5 100644 --- a/docs/source/Eng/eng_index.rst +++ b/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 diff --git a/docs/source/Zh/doc/new_features/v32_features_doc.rst b/docs/source/Zh/doc/new_features/v32_features_doc.rst new file mode 100644 index 00000000..aa0b6ea0 --- /dev/null +++ b/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** +分類下的指令。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 9525b554..dfcac399 100644 --- a/docs/source/Zh/zh_index.rst +++ b/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 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index eef767c8..b633ac61 100644 --- a/je_auto_control/__init__.py +++ b/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", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 1ce4401f..359fb34e 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/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=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index eba28f5a..bcacf8da 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/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, diff --git a/je_auto_control/utils/governance/__init__.py b/je_auto_control/utils/governance/__init__.py new file mode 100644 index 00000000..a6eb4515 --- /dev/null +++ b/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"] diff --git a/je_auto_control/utils/governance/governance.py b/je_auto_control/utils/governance/governance.py new file mode 100644 index 00000000..be10de9b --- /dev/null +++ b/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] diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index a4d8e6c4..09f97d6f 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -2573,6 +2573,52 @@ def plugin_sdk_tools() -> List[MCPTool]: ] +def governance_tools() -> List[MCPTool]: + _AD = {"action": {"type": "string"}, "requester": {"type": "string"}, + "db": {"type": "string"}} + _TA = {"token": {"type": "string"}, "approver": {"type": "string"}, + "db": {"type": "string"}} + return [ + MCPTool( + name="ac_approval_request", + description=("Maker-checker gate: file an approval request for a " + "high-risk 'action' and get a token. The action must " + "wait until a different principal approves. 'db' is an " + "optional JSON file shared across processes."), + input_schema=schema(_AD, ["action"]), + handler=h.approval_request, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_approval_approve", + description=("Approve a pending request token as 'approver'. " + "Rejected (returns approved=False) if the approver " + "equals the requester (segregation of duties)."), + input_schema=schema(_TA, ["token", "approver"]), + handler=h.approval_approve, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_approval_reject", + description=("Reject a pending request token as 'approver' (must " + "differ from the requester). Returns {rejected}."), + input_schema=schema(_TA, ["token", "approver"]), + handler=h.approval_reject, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_approval_status", + description=("Report a request token's status (pending/approved/" + "rejected) and an 'approved' boolean to gate an " + "action on."), + input_schema=schema({"token": {"type": "string"}, + "db": {"type": "string"}}, ["token"]), + handler=h.approval_status, + annotations=READ_ONLY, + ), + ] + + def unattended_tools() -> List[MCPTool]: return [ MCPTool( @@ -3630,7 +3676,7 @@ def media_assert_tools() -> List[MCPTool]: checkpoint_tools, set_of_marks_tools, screen_state_tools, input_macro_tools, resilience_tools, ci_annotation_tools, clipboard_history_tools, audit_analysis_tools, - process_doc_tools, tween_drag_tools, plugin_sdk_tools, + process_doc_tools, tween_drag_tools, plugin_sdk_tools, governance_tools, screen_record_tools, process_and_shell_tools, remote_desktop_tools, gamepad_tools, usb_passthrough_tools, assertion_tools, data_source_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 01a5f882..785dd30c 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1246,6 +1246,28 @@ def load_plugins(group="je_auto_control.commands"): return {"loaded": _load(group)} +def approval_request(action: str, requester: str = "", + db: Optional[str] = None): + 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): + 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): + from je_auto_control.utils.governance import ApprovalGate + return {"rejected": ApprovalGate(db).reject(token, approver)} + + +def approval_status(token: str, db: Optional[str] = None): + from je_auto_control.utils.governance import ApprovalGate + gate = ApprovalGate(db) + return {"status": gate.status(token), "approved": gate.is_approved(token)} + + def vlm_locate(description: str, screen_region: Optional[List[int]] = None, model: Optional[str] = None) -> Optional[List[int]]: diff --git a/test/unit_test/headless/test_governance_batch.py b/test/unit_test/headless/test_governance_batch.py new file mode 100644 index 00000000..cae4d514 --- /dev/null +++ b/test/unit_test/headless/test_governance_batch.py @@ -0,0 +1,94 @@ +"""Headless tests for the maker-checker approval gate. State is in-memory +(no db_path) or a tmp JSON file; pure stdlib, no Qt imports.""" +import je_auto_control as ac +from je_auto_control.utils.governance import ApprovalGate + + +def test_request_returns_pending_token(): + gate = ApprovalGate() + token = gate.request("delete prod table", requester="alice") + assert token and gate.status(token) == "pending" + assert gate.is_approved(token) is False + + +def test_checker_must_differ_from_maker(): + gate = ApprovalGate() + token = gate.request("wire funds", requester="alice") + # self-approval is refused (segregation of duties) + assert gate.approve(token, "alice") is False + assert gate.is_approved(token) is False + # a different checker succeeds + assert gate.approve(token, "bob") is True + assert gate.is_approved(token) is True + assert gate.status(token) == "approved" + + +def test_reject_blocks_approval(): + gate = ApprovalGate() + token = gate.request("deploy", requester="alice") + assert gate.reject(token, "bob") is True + assert gate.is_approved(token) is False + # a decided request cannot be re-decided + assert gate.approve(token, "carol") is False + + +def test_pending_lists_only_open_requests(): + gate = ApprovalGate() + open_token = gate.request("a", requester="alice") + closed = gate.request("b", requester="alice") + gate.approve(closed, "bob") + tokens = {r["token"] for r in gate.pending()} + assert tokens == {open_token} + + +def test_persists_across_instances(tmp_path): + db = str(tmp_path / "gate.json") + token = ApprovalGate(db).request("ship", requester="alice") + # a fresh instance (e.g. the checker's process) sees the request + assert ApprovalGate(db).approve(token, "bob") is True + assert ApprovalGate(db).is_approved(token) is True + + +def test_unknown_token_is_safe(): + gate = ApprovalGate() + assert gate.status("nope") is None + assert gate.is_approved("nope") is False + assert gate.approve("nope", "bob") is False + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(tmp_path): + db = str(tmp_path / "g.json") + rec = ac.execute_action([ + ["AC_approval_request", {"action": "x", "requester": "alice", + "db": db}], + ]) + token = next(v for v in rec.values() if isinstance(v, dict))["token"] + rec2 = ac.execute_action([ + ["AC_approval_approve", {"token": token, "approver": "bob", + "db": db}], + ["AC_approval_status", {"token": token, "db": db}], + ]) + statuses = [v for v in rec2.values() if isinstance(v, dict)] + assert any(v.get("approved") is True for v in statuses) + + +def test_wiring(): + known = ac.executor.known_commands() + assert {"AC_approval_request", "AC_approval_approve", + "AC_approval_reject", "AC_approval_status"} <= known + from je_auto_control.utils.mcp_server.tools import ( + build_default_tool_registry) + names = {t.name for t in build_default_tool_registry()} + assert {"ac_approval_request", "ac_approval_approve", + "ac_approval_reject", "ac_approval_status"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_approval_request", "AC_approval_approve", + "AC_approval_reject", "AC_approval_status"} <= cmds + + +def test_facade_export(): + assert hasattr(ac, "ApprovalGate") + assert "ApprovalGate" in ac.__all__