slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(drone): joint-decision gate on @git merge (DPLAN-0256, todo #92) — TTY gets y/N prompt, headless refused without --confirm, merge_pr unreachable ungated, all decisions logged. 6 new tests (86 green), live refusal verified, seedgo 31/31. Patrick ruling S330: merges together, never accidental

Browse Source
This commit is contained in:
AIOSAI
2026-07-21 18:05:39 -07:00
parent d1a6c874e8
commit 7a2700d649
3 changed files with 142 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+9
View File
@@ -11,6 +11,15 @@ PyPI version — not the changelog header.
## [2026-07-21]
**feat(drone)** — joint-decision gate on `drone @git merge` (DPLAN-0256,
Patrick ruling S330: merges are always done together, never accidental).
The gate sits in `_handle_merge` before the plugin import — `merge_pr()` is
unreachable without confirmation. A real terminal gets an interactive y/N
prompt; headless callers (agent Bash) are refused unless `--confirm` is
passed explicitly. Every gate decision (confirm / tty-yes / tty-abort /
headless-refused) is logged via json_handler. 6 new tests (86 green),
live-fired refusal verified, seedgo 31/31.
**feat(skills)** — telegram user_message_relay joins the sound layer: relay
events now carry their own sound key so an inbound user message is audible
like every other hook event (59/59 + 252 green).
src/aipass/drone/apps/modules/git_module.py
+45 -5
View File
@@ -17,6 +17,7 @@ from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
from aipass.prax import logger
@@ -302,10 +303,47 @@ def _handle_close_pr(args: list[str]) -> dict:
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _confirm_merge(pr_number: str, caller: str, confirmed: bool) -> dict | None:
"""Joint-decision gate: merges must never happen accidentally (DPLAN-0256).
Returns None when the merge may proceed, or a refusal/abort result dict.
Order matters: an explicit --confirm always passes; otherwise a real
terminal gets an interactive y/N prompt; headless callers are refused.
"""
if confirmed:
json_handler.log_operation("merge_gate", {"pr_number": pr_number, "caller": caller, "path": "--confirm"})
return None
if sys.stdin.isatty():
answer = input(f"Merge PR #{pr_number}? Merges are a joint decision. [y/N] ")
if answer.strip().lower() in ("y", "yes"):
json_handler.log_operation("merge_gate", {"pr_number": pr_number, "caller": caller, "path": "tty-yes"})
return None
json_handler.log_operation("merge_gate", {"pr_number": pr_number, "caller": caller, "path": "tty-abort"})
return {"stdout": "", "stderr": f"Merge of PR #{pr_number} aborted at prompt.", "exit_code": 1}
json_handler.log_operation("merge_gate", {"pr_number": pr_number, "caller": caller, "path": "headless-refused"})
logger.info("merge gate: refused headless merge of PR #%s by %s (no --confirm)", pr_number, caller)
return {
"stdout": "",
"stderr": (
f"Merge of PR #{pr_number} requires explicit confirmation — merges are a joint decision.\n"
f"Re-run once agreed: drone @git merge {pr_number} --confirm"
),
"exit_code": 1,
}
def _handle_merge(args: list[str], caller: str) -> dict:
"""Handle the merge subcommand (owner-tier, auth pre-checked)."""
if not args:
return {"stdout": "", "stderr": "Usage: drone @git merge <PR#>", "exit_code": 1}
confirmed = "--confirm" in args
pr_args = [a for a in args if not a.startswith("--")]
if not pr_args:
return {"stdout": "", "stderr": "Usage: drone @git merge <PR#> [--confirm]", "exit_code": 1}
refusal = _confirm_merge(pr_args[0], caller, confirmed)
if refusal is not None:
return refusal
try:
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
@@ -313,7 +351,7 @@ def _handle_merge(args: list[str], caller: str) -> dict:
logger.error("Failed to import devpulse_ops merge plugin: %s", exc)
return {"stdout": "", "stderr": f"devpulse_ops plugin not available: {exc}", "exit_code": 1}
result = merge_pr(args[0], caller)
result = merge_pr(pr_args[0], caller)
if result["success"]:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
@@ -637,8 +675,10 @@ def get_help(command: str | None = None) -> str:
return "git unlock --force — Force-release the PR lock [owner]\n"
if command == "merge":
return (
"git merge <PR#> — Merge a PR and sync local main [owner]\n"
"git merge <PR#> [--confirm] — Merge a PR and sync local main [owner]\n"
" Runs gh pr merge --merge --delete-branch, then git pull --rebase.\n"
" Joint-decision gate: terminal sessions get a y/N prompt; headless\n"
" callers must pass --confirm or the merge is refused.\n"
)
if command == "smart-sync":
return "git smart-sync — Fetch origin and rebase if behind [owner]\n"
@@ -686,7 +726,7 @@ def get_help(command: str | None = None) -> str:
" dev-pr <desc> Push dev and create PR to main\n"
" delete-branch <name> Delete a remote branch\n"
" close-pr <number> Close a PR\n"
" merge <PR#> Merge a PR\n"
" merge <PR#> [--confirm] Merge a PR (gated: y/N prompt or --confirm)\n"
" sync [--autostash] Sync with origin/main (FF on dev)\n"
" smart-sync Fetch + rebase if behind\n"
" unlock --force Force-release the PR lock\n"
src/aipass/drone/tests/test_git_module.py
+88
View File
@@ -1471,3 +1471,91 @@ class TestScopeFooter:
result = handle_command("diff", ["--all"])
assert "showing drone scope" not in result["stdout"]
# ===========================================================================
# Merge joint-decision gate (DPLAN-0256)
# ===========================================================================
_MERGE_PR = "aipass.drone.apps.plugins.devpulse_ops.merge_plugin.merge_pr"
_MERGE_OK = {
"success": True,
"pr_number": "123",
"title": "t",
"merge_commit": "abc123",
"message": "PR #123 merged: t (abc123)",
}
class TestMergeGate:
"""merge must never run without explicit confirmation."""
@patch(_AUTH, return_value="devpulse")
def test_headless_without_confirm_refused(self, _mock_auth: MagicMock) -> None:
"""Non-TTY caller without --confirm is refused before the plugin loads."""
with patch(_MERGE_PR) as mock_merge:
with patch(f"{_GIT_MOD}.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = False
result = handle_command("merge", ["123"])
assert result["exit_code"] == 1
assert "requires explicit confirmation" in result["stderr"]
assert "--confirm" in result["stderr"]
mock_merge.assert_not_called()
@patch(_AUTH, return_value="devpulse")
def test_confirm_flag_proceeds(self, _mock_auth: MagicMock) -> None:
"""--confirm merges without prompting, even headless."""
with patch(_MERGE_PR, return_value=dict(_MERGE_OK)) as mock_merge:
with patch(f"{_GIT_MOD}.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = False
result = handle_command("merge", ["123", "--confirm"])
assert result["exit_code"] == 0
mock_merge.assert_called_once_with("123", "devpulse")
@patch(_AUTH, return_value="devpulse")
def test_confirm_flag_position_agnostic(self, _mock_auth: MagicMock) -> None:
"""--confirm before the PR number still resolves the right PR."""
with patch(_MERGE_PR, return_value=dict(_MERGE_OK)) as mock_merge:
with patch(f"{_GIT_MOD}.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = False
result = handle_command("merge", ["--confirm", "123"])
assert result["exit_code"] == 0
mock_merge.assert_called_once_with("123", "devpulse")
@patch(_AUTH, return_value="devpulse")
def test_tty_yes_proceeds(self, _mock_auth: MagicMock) -> None:
"""Interactive terminal answering y merges."""
with patch(_MERGE_PR, return_value=dict(_MERGE_OK)) as mock_merge:
with patch(f"{_GIT_MOD}.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = True
with patch("builtins.input", return_value="y"):
result = handle_command("merge", ["123"])
assert result["exit_code"] == 0
mock_merge.assert_called_once_with("123", "devpulse")
@patch(_AUTH, return_value="devpulse")
def test_tty_default_aborts(self, _mock_auth: MagicMock) -> None:
"""Interactive terminal hitting enter (default N) aborts."""
with patch(_MERGE_PR) as mock_merge:
with patch(f"{_GIT_MOD}.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = True
with patch("builtins.input", return_value=""):
result = handle_command("merge", ["123"])
assert result["exit_code"] == 1
assert "aborted" in result["stderr"]
mock_merge.assert_not_called()
@patch(_AUTH, return_value="devpulse")
def test_confirm_without_pr_number_is_usage_error(self, _mock_auth: MagicMock) -> None:
"""--confirm alone (no PR number) is a usage error, not a merge."""
with patch(_MERGE_PR) as mock_merge:
result = handle_command("merge", ["--confirm"])
assert result["exit_code"] == 1
assert "Usage" in result["stderr"]
mock_merge.assert_not_called()