Merge pull request #382 from AIOSAI/system/devpulse-docs-lowercase-shouting-imperatives-in-localglobal

feat(system): docs: lowercase shouting imperatives in local+global prompts (volume isn't the fix)

.aipass/aipass_global_prompt.md

@@ -57,7 +57,11 @@ Workflow:
 2. Make edits directly on main.
 3. When the work is ready to ship: `drone @git system-pr "description"`.
 4. That command commits + branches + pushes + PRs + returns you to main. One action.
-5. Devpulse reviews + merges with `drone @git merge <PR#>`.
+5. STOP. The user merges. Do not run `drone @git merge` unless the user explicitly tells you to merge a specific PR number in this session.
+
+Never merge. Ever. User-merges-only. Past PRs, your own PRs, closed PRs — none of them auto-qualify. You fix, you PR, you stop.
+
+Local files are source of truth. When you edit a file, the state on disk IS reality — you don't wait for a merge to act on what you see locally. This also means: if the truth is wrong, fix it locally, then PR.
 
 Why this matters: the AIPass repo has ONE shared HEAD across all branches. If any agent lingers on a non-main HEAD, every other agent's next edit lands on the wrong branch. Files get stranded. Work gets lost. Conflicts pile up. We've lived this pain — don't repeat it.
 
@@ -175,7 +179,7 @@ Archive commands:
 
 # Git Workflow
 
-**Drone is the ONLY git interface. Period.** All PR workflow goes through drone. Never use raw git commands for commits, branches, pushes, resets, merges, rebases, cherry-picks, or remote branch manipulation. Drone handles everything atomically with a lockfile that prevents concurrent PR collisions.
+**Drone is the only git interface. Period.** All PR workflow goes through drone. Never use raw git commands for commits, branches, pushes, resets, merges, rebases, cherry-picks, or remote branch manipulation. Drone handles everything atomically with a lockfile that prevents concurrent PR collisions.
 
 **If you think you need a raw git command to fix a git problem, STOP. You don't.** Every git state devpulse has ever been in has been recoverable through `drone @git` commands — system-pr, merge, smart-sync, fix, status, sync, lock. There is no situation that requires `git reset`, `git push`, `git cherry-pick`, `git rebase`, or `git branch -f`. Reaching for them has always made things worse. If drone's commands don't obviously handle the state you're in, run `drone @git fix` or `drone @git smart-sync` and re-evaluate. If still stuck, ASK THE USER — do not improvise with raw git.
 

.claude/CLAUDE.md

@@ -36,7 +36,9 @@ You do not create branches. You do not checkout other branches. You do not instr
 
 Branches only exist inside the atomic `drone @git system-pr` command which commits → creates branch → pushes → opens PR → **returns HEAD to main**. That one command owns the entire branch lifecycle. Agents own nothing about branches.
 
-Workflow: edit on main → `drone @git system-pr "msg"` → back on main. Devpulse merges reviewed PRs with `drone @git merge <PR#>`.
+Workflow: edit on main → `drone @git system-pr "msg"` → back on main → STOP. The user merges. Never run `drone @git merge` without an explicit user instruction for that specific PR number — not even for your own PRs, not even for PRs that look ready. User-merges-only.
+
+Local files are source of truth. A file edit IS reality on disk; you don't wait for a merge to act on it. If the truth is wrong, fix locally first, then PR.
 
 `git checkout*` and `git add -f*` are denied system-wide in `.claude/settings.json`. These aren't arbitrary rules — they came from fixing actual bugs caused by agents staying on branches. Trust them.
 

.claude/settings.json

@@ -13,6 +13,12 @@
       "Bash(git add -f*)",
       "Bash(git add --force*)",
       "Bash(git checkout*)",
+      "Bash(git switch -c*)",
+      "Bash(git switch --create*)",
+      "Bash(git branch -c*)",
+      "Bash(git branch --copy*)",
+      "Bash(git branch -m*)",
+      "Bash(git branch --move*)",
       "Read(/home/patrick/Patrick-Personal/**)",
       "Edit(/home/patrick/Patrick-Personal/**)",
       "Write(/home/patrick/Patrick-Personal/**)",

src/aipass/devpulse/.aipass/aipass_local_prompt.md

@@ -14,8 +14,8 @@ You are DEVPULSE — Patrick's primary AI collaborator and orchestration hub for
 - **Don't solo-rebuild other branches.** Full multi-file implementations → dispatch via `drone @ai_mail dispatch @branch`.
 - **Delegate heavy code to sub-agents** (`run_in_background: true`). Fire and forget, move on immediately. Launch → continue → get notified → report results. Never block waiting on agents.
 - Use `drone @branch --help` for command syntax. Use `drone systems` for branch list.
-- **ALWAYS WAKE after sending dispatch emails.** Send email → wake. Every time. No asking.
-- **START WATCHDOG after any dispatch.** Run `drone @devpulse watchdog agent @target` (see Watchdog section) with `run_in_background: true`. Don't wait for the user to ask.
+- **Always wake after sending dispatch emails.** Send email → wake. Every time. No asking.
+- **Start watchdog after any dispatch.** Run `drone @devpulse watchdog agent @target` (see Watchdog section) with `run_in_background: true`. Don't wait for the user to ask.
 
 ## Branch Experts — Ask Before Rebuilding
 
@@ -32,11 +32,17 @@ When a task belongs to a specialist's DOMAIN, ask them. You can still investigat
 | Command routing | @drone | @branch resolution, subprocess |
 | Memory, vectors | @memory | ChromaDB, search, archival |
 
-## Git Workflow — Always on Main, Drone Only
+## Git Workflow — Always on Main, Drone Only, Never Merge
 
-**One rule: always on main. No exceptions.** You don't create branches. You don't tell other agents to create branches. You don't stay on someone else's branch while they're mid-work. Branches exist ONLY inside the atomic `drone @git system-pr` window which commits → creates branch → pushes → PRs → returns HEAD to main. Every other moment: you're on main.
+**Three rules, in order:**
 
-Why: AIPass repo has ONE shared HEAD. Linger on a non-main HEAD and every agent's next edit lands on the wrong branch. Work gets stranded. Dispatch briefs must NEVER say "create a branch as step 1" — that's what caused the S101 merge mess.
+1. **Always on main. No exceptions.** You don't create branches. You don't tell other agents to create branches. Branches exist only inside the atomic `drone @git system-pr` window which commits → creates branch → pushes → PRs → returns HEAD to main. Every other moment: you're on main.
+
+2. **Never merge PRs.** That's the user's role. You fix, you PR, you stop. The user says "merge X" or merges themselves. Do not run `drone @git merge` without an explicit user instruction for that specific PR number. Past PRs, closed PRs, your own PRs — none of them auto-qualify. User-merges-only is the rule.
+
+3. **Local files are source of truth.** When you make an edit, the file on disk is reality — you don't need to wait for a merge to act on the state you see. But that also means: if the truth is wrong, fix it locally first, then PR. Don't assume remote state matches.
+
+Why main-only: AIPass repo has one shared HEAD. Linger on a non-main HEAD and every agent's next edit lands on the wrong branch. Work gets stranded. Dispatch briefs must never say "create a branch as step 1" — that's what caused the S101 merge mess.
 
 Never use raw git commands (git commit, git push, git checkout anything, gh pr create). `Bash(git checkout*)` and `Bash(git add -f*)` are denied system-wide in `.claude/settings.json`. Drone handles everything correctly.
 
@@ -52,7 +58,7 @@ drone @git lock                      # Check PR lock status
 
 Read-only git commands are fine: `git status`, `git diff`, `git log`.
 
-**NEVER cd to repo root.** `drone @git system-pr` requires `.trinity/passport.json` in the CWD hierarchy. If you cd to the repo root, it fails. Stage files with relative paths from devpulse: `git add ../../../HERALD.md`. Always run drone commands from this directory.
+**Never cd to repo root.** `drone @git system-pr` requires `.trinity/passport.json` in the CWD hierarchy. If you cd to the repo root, it fails. Stage files with relative paths from devpulse: `git add ../../../HERALD.md`. Always run drone commands from this directory.
 
 ## Key Commands
 

src/aipass/drone/.seedgo/bypass.json

@@ -142,6 +142,22 @@
       "file": "tests/test_devpulse_plugins.py",
       "standard": "documentation",
       "reason": "Test methods and mock helpers in this file are pre-existing without docstrings. Class docstrings describe intent; per-method docs would be noise."
+    },
+    {
+      "file": "tests/test_git_module.py",
+      "standard": "architecture",
+      "reason": "Test file — intentionally outside 3-layer structure. tests/ is a peer of apps/, not part of it."
+    },
+    {
+      "file": "tests/test_git_module.py",
+      "standard": "encapsulation",
+      "lines": [20],
+      "reason": "Test file imports lock_handler directly to test its public interface. Unit tests require direct access to implementation components."
+    },
+    {
+      "file": "tests/test_system_pr.py",
+      "standard": "architecture",
+      "reason": "Test file — intentionally outside 3-layer structure. tests/ is a peer of apps/, not part of it."
     }
   ],
   "notes": {

src/aipass/drone/README.md

@@ -117,14 +117,15 @@ drone/
 │   │   ├── scanning/      # Scan result formatting + discovery
 │   │   ├── command_registry/  # Command shortcut CRUD + lookup
 │   │   └── git/           # Git workflow handlers
-│   │       ├── lock_handler.py    # Atomic lockfile (O_CREAT|O_EXCL)
-│   │       ├── pr_handler.py      # 10-step PR workflow
-│   │       ├── status_handler.py  # Scoped git status
-│   │       └── sync_handler.py    # Safe main sync
+│   │       ├── lock_handler.py              # Atomic lockfile (O_CREAT|O_EXCL)
+│   │       ├── pr_handler.py               # 10-step PR workflow
+│   │       ├── status_handler.py           # Scoped git status (subprocess)
+│   │       ├── status_handler_gitpython.py # [prototype] GitPython status — DPLAN-0140 Phase 1, not wired in
+│   │       └── sync_handler.py             # Safe main sync
 │   └── plugins/           # Extensions beyond core routing
 │       └── devpulse_ops/  # System-wide PR, merge, smart-sync, fix
 ├── docs/                  # Documentation
-└── tests/                 # 513 tests, 19 test files
+└── tests/                 # 529+ tests, 19 test files
 ```
 
 ---
@@ -172,7 +173,7 @@ Infrastructure modules (seedgo, cli, git) work from external AIPass projects wit
 
 ---
 
-**Last Updated:** 2026-04-07
+**Last Updated:** 2026-04-21
 
 ---
 [← Back to AIPass](../../../README.md)

src/aipass/drone/apps/handlers/git/status_handler_gitpython.py

@@ -38,15 +38,22 @@ Design note (two-library split):
 from __future__ import annotations
 
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from aipass.prax import logger
 from aipass.drone.apps.handlers.json import json_handler
 from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
 
+if TYPE_CHECKING:
+    from git import Repo as GitRepo
+
 try:
     import git as _git_module
+
     _GITPYTHON_AVAILABLE = True
-except ImportError:
+except ImportError as exc:
+    logger.info("status_handler_gitpython: GitPython not installed (%s) — handler disabled", exc)
+    _git_module = None  # type: ignore[assignment]
     _GITPYTHON_AVAILABLE = False
 
 
@@ -69,13 +76,13 @@ _UNSTAGED_STATUS_MAP: dict[str, str] = {
 }
 
 
-def _collect_staged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
+def _collect_staged(repo: "GitRepo", rel_prefix: str, rel_dir: str) -> list[dict]:
     """Return staged changes that fall under the branch directory."""
     files: list[dict] = []
     try:
         staged_diffs = repo.head.commit.diff()
     except Exception as exc:  # empty repo or detached HEAD
-        logger.debug("status_handler_gitpython: could not get staged diffs: %s", exc)
+        logger.info("status_handler_gitpython: could not get staged diffs: %s", exc)
         return files
 
     for diff in staged_diffs:
@@ -84,12 +91,13 @@ def _collect_staged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) ->
             continue
         if not (path.startswith(rel_prefix) or path == rel_dir):
             continue
-        code = _STAGED_STATUS_MAP.get(diff.change_type, diff.change_type)
+        change_key = diff.change_type or ""
+        code = _STAGED_STATUS_MAP.get(change_key, change_key)
         files.append({"status": code, "path": path})
     return files
 
 
-def _collect_unstaged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
+def _collect_unstaged(repo: "GitRepo", rel_prefix: str, rel_dir: str) -> list[dict]:
     """Return unstaged working-tree changes that fall under the branch directory."""
     files: list[dict] = []
     for diff in repo.index.diff(None):
@@ -98,12 +106,13 @@ def _collect_unstaged(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -
             continue
         if not (path.startswith(rel_prefix) or path == rel_dir):
             continue
-        code = _UNSTAGED_STATUS_MAP.get(diff.change_type, diff.change_type)
+        change_key = diff.change_type or ""
+        code = _UNSTAGED_STATUS_MAP.get(change_key, change_key)
         files.append({"status": code, "path": path})
     return files
 
 
-def _collect_untracked(repo: "_git_module.Repo", rel_prefix: str, rel_dir: str) -> list[dict]:
+def _collect_untracked(repo: "GitRepo", rel_prefix: str, rel_dir: str) -> list[dict]:
     """Return untracked files that fall under the branch directory."""
     files: list[dict] = []
     for upath in repo.untracked_files:
@@ -127,11 +136,8 @@ def get_branch_status(branch_dir: Path) -> dict:
             total  -- int count of changed files
             message -- human-readable summary string
     """
-    if not _GITPYTHON_AVAILABLE:
-        logger.error(
-            "status_handler_gitpython: GitPython is not installed. "
-            "Run: pip install gitpython"
-        )
+    if not _GITPYTHON_AVAILABLE or _git_module is None:
+        logger.error("status_handler_gitpython: GitPython is not installed. Run: pip install gitpython")
         return {
             "files": [],
             "total": 0,

src/aipass/drone/tests/test_git_module.py

@@ -1,3 +1,11 @@
+# =================== AIPass ====================
+# Name: test_git_module.py
+# Description: Tests for the @git module — lock, status, sync, PR, and routing
+# Version: 1.0.0
+# Created: 2026-04-21
+# Modified: 2026-04-21
+# =============================================
+
 """Tests for the @git module — lock, status, sync, PR, and routing."""
 
 from __future__ import annotations
@@ -358,6 +366,41 @@ class TestSyncHandler:
 # ===========================================================================
 
 
+def _run_nothing_staged(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock: on main, nothing staged (diff --cached returns 0)."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = "main\n" if cmd[1:3] == ["rev-parse", "--abbrev-ref"] else ""
+    return r
+
+
+def _run_cleanup_early_exit(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock: on main, nothing staged — triggers early exit path."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = "main\n" if cmd[1:3] == ["rev-parse", "--abbrev-ref"] else ""
+    return r
+
+
+def _run_with_staged(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock: on main, staged changes, successful commit/push/PR."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = ""
+    if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
+        r.stdout = "main\n"
+    elif cmd[1:3] == ["diff", "--cached"]:
+        r.returncode = 1  # 1 = something staged
+    elif cmd[0] == "git" and cmd[1] == "commit":
+        r.stdout = "[main abc1234] feat(api): test"
+    elif cmd[0] == "gh":
+        r.stdout = "https://github.com/test/repo/pull/1"
+    return r
+
+
 class TestPRHandler:
     """PR workflow error path tests."""
 
@@ -404,34 +447,7 @@ class TestPRHandler:
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        call_count = 0
-
-        def mock_subprocess_run(cmd, **kwargs):
-            """Simulate git subprocess returning main branch and staged-nothing."""
-            nonlocal call_count
-            call_count += 1
-            result = MagicMock()
-            result.stderr = ""
-            result.stdout = ""
-
-            if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
-                result.returncode = 0
-                result.stdout = "main\n"
-            elif cmd[1:3] == ["checkout", "-b"]:
-                result.returncode = 0
-            elif cmd[0] == "git" and cmd[1] == "add":
-                result.returncode = 0
-            elif cmd[1:3] == ["diff", "--cached"]:
-                result.returncode = 0  # 0 means nothing staged
-            elif cmd[1] == "checkout" and cmd[2] == "main":
-                result.returncode = 0
-            elif cmd[1] == "pull":
-                result.returncode = 0
-            else:
-                result.returncode = 0
-            return result
-
-        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=mock_subprocess_run):
+        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=_run_nothing_staged):
             with patch(
                 "aipass.drone.apps.handlers.git.pr_handler.acquire_lock",
                 return_value={"success": True, "message": "ok"},
@@ -448,27 +464,9 @@ class TestPRHandler:
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        def mock_subprocess_run(cmd, **kwargs):
-            """Simulate git subprocess returning main branch, then early exit on no staged files."""
-            result = MagicMock()
-            result.stderr = ""
-            result.stdout = ""
-
-            if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
-                result.returncode = 0
-                result.stdout = "main\n"
-            elif cmd[0] == "git" and cmd[1] == "add":
-                result.returncode = 0
-            elif cmd[1:3] == ["diff", "--cached"]:
-                # Nothing staged — triggers early exit
-                result.returncode = 0
-            else:
-                result.returncode = 0
-            return result
-
         release_mock = MagicMock(return_value={"success": True, "message": "ok"})
 
-        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=mock_subprocess_run):
+        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=_run_cleanup_early_exit):
             with patch(
                 "aipass.drone.apps.handlers.git.pr_handler.acquire_lock",
                 return_value={"success": True, "message": "ok"},
@@ -491,36 +489,9 @@ class TestPRHandler:
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        commit_cmd_seen: list[list[str]] = []
-
-        def mock_subprocess_run(cmd, **kwargs):
-            """Simulate git/gh subprocess calls, recording commit invocations."""
-            r = MagicMock()
-            r.stderr = ""
-            r.stdout = ""
-            if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
-                r.returncode = 0
-                r.stdout = "main\n"
-            elif cmd[0] == "git" and cmd[1] == "add":
-                r.returncode = 0
-            elif cmd[1:3] == ["diff", "--cached"]:
-                r.returncode = 1  # 1 means something is staged
-            elif cmd[0] == "git" and cmd[1] == "commit":
-                commit_cmd_seen.append(list(cmd))
-                r.returncode = 0
-                r.stdout = "[main abc1234] feat(api): test"
-            elif cmd[0] == "git" and cmd[1] == "branch":
-                r.returncode = 0
-            elif cmd[0] == "git" and cmd[1] == "push":
-                r.returncode = 0
-            elif cmd[0] == "gh":
-                r.returncode = 0
-                r.stdout = "https://github.com/test/repo/pull/1"
-            else:
-                r.returncode = 0
-            return r
-
-        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=mock_subprocess_run):
+        with patch(
+            "aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=_run_with_staged
+        ) as mock_run:
             with patch(
                 "aipass.drone.apps.handlers.git.pr_handler.acquire_lock",
                 return_value={"success": True, "message": "ok"},
@@ -529,8 +500,11 @@ class TestPRHandler:
                     create_pr("api", "test desc", tmp_path / "src" / "aipass" / "api")
 
         # The commit command must include '--' separator + pathspec to scope to branch_dir
-        assert commit_cmd_seen, "commit was never called"
-        commit_cmd = commit_cmd_seen[0]
+        commit_calls = [
+            c.args[0] for c in mock_run.call_args_list if c.args and c.args[0][0] == "git" and c.args[0][1] == "commit"
+        ]
+        assert commit_calls, "commit was never called"
+        commit_cmd = commit_calls[0]
         assert "--" in commit_cmd, "commit missing '--' pathspec separator"
         pathspec_idx = commit_cmd.index("--")
         pathspec = commit_cmd[pathspec_idx + 1]
@@ -744,6 +718,49 @@ class TestModuleRegistration:
 # ===========================================================================
 
 
+def _run_pr_created_success(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock for a full successful pr_handler run (fires pr_created)."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = ""
+    if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
+        r.stdout = "main\n"
+    elif cmd[1:3] == ["diff", "--cached"]:
+        r.returncode = 1
+    elif cmd[0] == "gh" and cmd[1] == "pr":
+        r.stdout = "https://github.com/org/repo/pull/99\n"
+    return r
+
+
+def _run_pr_trigger_resilience(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock for pr_handler run where trigger.fire raises."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = ""
+    if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
+        r.stdout = "main\n"
+    elif cmd[1:3] == ["diff", "--cached"]:
+        r.returncode = 1
+    elif cmd[0] == "gh" and cmd[1] == "pr":
+        r.stdout = "https://github.com/org/repo/pull/100\n"
+    return r
+
+
+def _run_merge_success(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Subprocess mock for a successful merge_plugin run (fires pr_merged)."""
+    r = MagicMock()
+    r.returncode = 0
+    r.stderr = ""
+    r.stdout = ""
+    if cmd[0] == "gh" and cmd[1] == "pr" and cmd[2] == "view":
+        r.stdout = "Fix the thing\n"
+    elif cmd[1:3] == ["rev-parse", "HEAD"]:
+        r.stdout = "abc123def456\n"
+    return r
+
+
 class TestTriggerFireIntegration:
     """Verify trigger.fire() is called after successful PR/merge operations."""
 
@@ -753,43 +770,9 @@ class TestTriggerFireIntegration:
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        call_count = 0
-
-        def mock_run(cmd, **kwargs):
-            """Simulate git subprocess calls and count invocations."""
-            nonlocal call_count
-            call_count += 1
-            r = MagicMock()
-            r.stderr = ""
-            if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
-                r.returncode = 0
-                r.stdout = "main\n"
-            elif cmd[0] == "git" and cmd[1] == "add":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[1:3] == ["diff", "--cached"]:
-                r.returncode = 1  # 1 = something staged
-                r.stdout = ""
-            elif cmd[0] == "git" and cmd[1] == "commit":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[0] == "git" and cmd[1] == "branch":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[0] == "git" and cmd[1] == "push":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[0] == "gh" and cmd[1] == "pr":
-                r.returncode = 0
-                r.stdout = "https://github.com/org/repo/pull/99\n"
-            else:
-                r.returncode = 0
-                r.stdout = ""
-            return r
-
         mock_trigger = MagicMock()
 
-        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=mock_run):
+        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=_run_pr_created_success):
             with patch(
                 "aipass.drone.apps.handlers.git.pr_handler.acquire_lock",
                 return_value={"success": True, "message": "ok"},
@@ -807,28 +790,10 @@ class TestTriggerFireIntegration:
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        def mock_run(cmd, **kwargs):
-            """Simulate git/gh subprocess calls for trigger-failure resilience test."""
-            r = MagicMock()
-            r.stderr = ""
-            if cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
-                r.returncode = 0
-                r.stdout = "main\n"
-            elif cmd[1:3] == ["diff", "--cached"]:
-                r.returncode = 1
-                r.stdout = ""
-            elif cmd[0] == "gh" and cmd[1] == "pr":
-                r.returncode = 0
-                r.stdout = "https://github.com/org/repo/pull/100\n"
-            else:
-                r.returncode = 0
-                r.stdout = ""
-            return r
-
         mock_trigger = MagicMock()
         mock_trigger.fire.side_effect = RuntimeError("trigger broken")
 
-        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=mock_run):
+        with patch("aipass.drone.apps.handlers.git.pr_handler.subprocess.run", side_effect=_run_pr_trigger_resilience):
             with patch(
                 "aipass.drone.apps.handlers.git.pr_handler.acquire_lock",
                 return_value={"success": True, "message": "ok"},
@@ -841,40 +806,17 @@ class TestTriggerFireIntegration:
 
     def test_merge_plugin_fires_pr_merged(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
         """merge_plugin.merge_pr fires pr_merged event on success."""
+        from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
+
         registry = tmp_path / "AIPASS_REGISTRY.json"
         registry.write_text("{}", encoding="utf-8")
         monkeypatch.chdir(tmp_path)
 
-        from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
-
-        call_idx = 0
-
-        def mock_run(cmd, **kwargs):
-            """Simulate gh pr merge and git pull subprocess calls."""
-            nonlocal call_idx
-            call_idx += 1
-            r = MagicMock()
-            r.stderr = ""
-            if cmd[0] == "gh" and cmd[1] == "pr" and cmd[2] == "merge":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[0] == "git" and cmd[1] == "pull":
-                r.returncode = 0
-                r.stdout = ""
-            elif cmd[1:3] == ["rev-parse", "HEAD"]:
-                r.returncode = 0
-                r.stdout = "abc123def456\n"
-            elif cmd[0] == "gh" and cmd[1] == "pr" and cmd[2] == "view":
-                r.returncode = 0
-                r.stdout = "Fix the thing\n"
-            else:
-                r.returncode = 0
-                r.stdout = ""
-            return r
-
         mock_trigger = MagicMock()
 
-        with patch("aipass.drone.apps.plugins.devpulse_ops.merge_plugin.subprocess.run", side_effect=mock_run):
+        with patch(
+            "aipass.drone.apps.plugins.devpulse_ops.merge_plugin.subprocess.run", side_effect=_run_merge_success
+        ):
             with patch("aipass.trigger.apps.modules.core.trigger", mock_trigger):
                 result = merge_pr("42", "devpulse")
 

src/aipass/drone/tests/test_system_pr.py

@@ -3,7 +3,7 @@
 # Description: Tests for devpulse_ops plugin — auth and system PR workflow
 # Version: 1.0.0
 # Created: 2026-03-30
-# Modified: 2026-03-30
+# Modified: 2026-04-21
 # =============================================
 
 """Tests for devpulse_ops plugin — auth and system PR workflow."""
@@ -52,15 +52,15 @@ def devpulse_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
 
 @pytest.fixture()
 def seedgo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
-    """Create a temp directory with a seedgo passport (unauthorized)."""
+    """Create a temp directory with an unauthorized branch passport."""
     trinity = tmp_path / ".trinity"
     trinity.mkdir()
     passport = trinity / "passport.json"
     passport.write_text(
         json.dumps(
             {
-                "branch_info": {"branch_name": "seedgo"},
-                "identity": {"name": "seedgo"},
+                "branch_info": {"branch_name": "citizen/unauthorized"},
+                "identity": {"name": "citizen/unauthorized"},
             }
         ),
         encoding="utf-8",
@@ -85,6 +85,54 @@ def repo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     return tmp_path
 
 
+# ===========================================================================
+# Module-level subprocess side-effect helpers
+# ===========================================================================
+
+# Responses for "nothing staged" scenario: (stdout, returncode)
+_NOTHING_TO_COMMIT_RESPONSES: dict[tuple[str, ...], tuple[str, int]] = {
+    ("git", "rev-parse", "--abbrev-ref"): ("main\n", 0),
+    ("git", "add", "-u"): ("", 0),
+    ("git", "diff", "--cached"): ("", 0),  # exit 0 = nothing staged
+    ("git", "fetch", "origin"): ("", 0),
+    ("git", "rev-list", "--count"): ("0\n", 0),
+}
+
+# Responses for a complete successful PR flow: (stdout, returncode)
+_PR_FLOW_RESPONSES: dict[tuple[str, ...], tuple[str, int]] = {
+    ("git", "rev-parse", "--abbrev-ref"): ("main\n", 0),
+    ("git", "add", "-u"): ("", 0),
+    ("git", "diff", "--cached"): ("diff --git a/foo.py b/foo.py\n", 1),  # 1 = staged
+    ("git", "fetch", "origin"): ("", 0),
+    ("git", "rev-list", "--count"): ("1\n", 0),
+    ("git", "commit"): ("[main abc1234] test description\n", 0),
+    ("git", "branch"): ("", 0),
+    ("git", "push"): ("", 0),
+    ("gh", "pr"): ("https://github.com/org/repo/pull/42\n", 0),
+}
+
+
+def _make_proc(stdout: str, returncode: int) -> MagicMock:
+    """Build a minimal subprocess mock with stdout, returncode, and empty stderr."""
+    proc = MagicMock()
+    proc.stdout = stdout
+    proc.returncode = returncode
+    proc.stderr = ""
+    return proc
+
+
+def _nothing_to_commit_side_effect(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Return a mock proc for the 'nothing staged' subprocess sequence."""
+    stdout, rc = _NOTHING_TO_COMMIT_RESPONSES.get(tuple(cmd[:3]), ("", 0))
+    return _make_proc(stdout, rc)
+
+
+def _pr_flow_run_side_effect(cmd: list[str], **kwargs: object) -> MagicMock:
+    """Return a mock proc for a complete successful PR-flow subprocess sequence."""
+    stdout, rc = _PR_FLOW_RESPONSES.get(tuple(cmd[:2]), ("", 0))
+    return _make_proc(stdout, rc)
+
+
 # ===========================================================================
 # 1. auth.verify_caller tests
 # ===========================================================================
@@ -94,6 +142,7 @@ class TestVerifyCallerAuthorized:
     """verify_caller should return the branch name for devpulse."""
 
     def test_verify_caller_with_devpulse_passport(self, devpulse_dir: Path) -> None:
+        """Devpulse passport resolves to the 'devpulse' branch name."""
         result = verify_caller()
         assert result == "devpulse"
         assert result in ALLOWED_CALLERS
@@ -103,11 +152,13 @@ class TestVerifyCallerUnauthorized:
     """verify_caller should raise PermissionError for non-devpulse branches."""
 
     def test_verify_caller_unauthorized(self, seedgo_dir: Path) -> None:
+        """An unauthorized branch raises PermissionError with 'not authorized'."""
         with pytest.raises(PermissionError, match="not authorized"):
             verify_caller()
 
     def test_error_message_includes_branch_name(self, seedgo_dir: Path) -> None:
-        with pytest.raises(PermissionError, match="seedgo"):
+        """The PermissionError message includes the actual branch name."""
+        with pytest.raises(PermissionError, match="citizen/unauthorized"):
             verify_caller()
 
 
@@ -115,6 +166,7 @@ class TestVerifyCallerNoPassport:
     """verify_caller should raise PermissionError when no passport exists."""
 
     def test_verify_caller_no_passport(self, no_passport_dir: Path) -> None:
+        """Missing passport raises PermissionError mentioning passport path."""
         with pytest.raises(PermissionError, match="No .trinity/passport.json"):
             verify_caller()
 
@@ -128,29 +180,37 @@ class TestSlugify:
     """Test the slugify function with various inputs."""
 
     def test_basic_slugify(self) -> None:
+        """Space-separated words become hyphen-separated lowercase slugs."""
         assert slugify("Update all configs") == "update-all-configs"
 
     def test_special_characters_removed(self) -> None:
+        """Punctuation and special characters are stripped from the output."""
         assert slugify("fix: broken pipe!") == "fix-broken-pipe"
 
     def test_multiple_spaces_collapse(self) -> None:
+        """Consecutive spaces collapse into a single hyphen."""
         assert slugify("too   many   spaces") == "too-many-spaces"
 
     def test_max_length_truncation(self) -> None:
+        """Slugs longer than 50 characters are truncated."""
         long_desc = "a" * 100
         result = slugify(long_desc)
         assert len(result) <= 50
 
     def test_leading_trailing_hyphens_stripped(self) -> None:
+        """Leading and trailing hyphens are stripped from the result."""
         assert slugify("  --hello world--  ") == "hello-world"
 
     def test_empty_string(self) -> None:
+        """An empty input returns an empty string."""
         assert slugify("") == ""
 
     def test_all_special_chars(self) -> None:
+        """A string of only special characters produces an empty slug."""
         assert slugify("!!!@@@###") == ""
 
     def test_mixed_case(self) -> None:
+        """Mixed-case input is normalized to lowercase."""
         assert slugify("Hello World FOO") == "hello-world-foo"
 
 
@@ -165,13 +225,9 @@ class TestSystemPrNotOnMain:
     @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.find_repo_root")
     @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.subprocess.run")
     def test_system_pr_not_on_main(self, mock_run: MagicMock, mock_root: MagicMock, tmp_path: Path) -> None:
+        """Returns failure dict when HEAD is on a feature branch instead of main."""
         mock_root.return_value = tmp_path
-
-        # Simulate being on a feature branch
-        proc = MagicMock()
-        proc.stdout = "feature/something\n"
-        proc.returncode = 0
-        mock_run.return_value = proc
+        mock_run.return_value = _make_proc("feature/something\n", 0)
 
         result = create_system_pr("test description", "devpulse")
 
@@ -199,39 +255,10 @@ class TestSystemPrNothingToCommit:
         mock_release: MagicMock,
         tmp_path: Path,
     ) -> None:
+        """Returns failure dict with 'Nothing to PR' when no staged changes exist."""
         mock_root.return_value = tmp_path
         mock_acquire.return_value = {"success": True, "message": "Lock acquired"}
-
-        def side_effect(cmd: list[str], **kwargs: object) -> MagicMock:
-            proc = MagicMock()
-            if cmd[:3] == ["git", "rev-parse", "--abbrev-ref"]:
-                proc.stdout = "main\n"
-                proc.returncode = 0
-            elif cmd[:3] == ["git", "add", "-u"]:
-                proc.stdout = ""
-                proc.stderr = ""
-                proc.returncode = 0
-            elif cmd[:3] == ["git", "diff", "--cached"]:
-                # returncode 0 means nothing staged
-                proc.stdout = ""
-                proc.stderr = ""
-                proc.returncode = 0
-            elif cmd[:3] == ["git", "fetch", "origin"]:
-                proc.stdout = ""
-                proc.stderr = ""
-                proc.returncode = 0
-            elif cmd[:3] == ["git", "rev-list", "--count"]:
-                # 0 commits ahead
-                proc.stdout = "0\n"
-                proc.stderr = ""
-                proc.returncode = 0
-            else:
-                proc.stdout = ""
-                proc.stderr = ""
-                proc.returncode = 0
-            return proc
-
-        mock_run.side_effect = side_effect
+        mock_run.side_effect = _nothing_to_commit_side_effect
 
         result = create_system_pr("test description", "devpulse")
 
@@ -240,7 +267,38 @@ class TestSystemPrNothingToCommit:
 
 
 # ===========================================================================
-# 5. git_module routing for system-pr
+# 5. create_system_pr — HEAD stays on main (no git checkout)
+# ===========================================================================
+
+
+class TestSystemPrHeadStaysOnMain:
+    """system-pr must never call 'git checkout' — HEAD stays on main throughout."""
+
+    @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.release_lock")
+    @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.acquire_lock")
+    @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.find_repo_root")
+    @patch("aipass.drone.apps.plugins.devpulse_ops.pr_plugin.subprocess.run")
+    def test_no_git_checkout_during_successful_system_pr(
+        self,
+        mock_run: MagicMock,
+        mock_root: MagicMock,
+        mock_acquire: MagicMock,
+        mock_release: MagicMock,
+        tmp_path: Path,
+    ) -> None:
+        """No subprocess call contains 'git checkout' during a complete PR flow."""
+        mock_root.return_value = tmp_path
+        mock_acquire.return_value = {"success": True, "message": "Lock acquired"}
+        mock_run.side_effect = _pr_flow_run_side_effect
+
+        create_system_pr("test description", "devpulse")
+
+        all_cmds = [c[0][0] if c[0] else c.args[0] for c in mock_run.call_args_list]
+        assert not any("checkout" in cmd for cmd in all_cmds)
+
+
+# ===========================================================================
+# 6. git_module routing for system-pr
 # ===========================================================================
 
 
@@ -248,23 +306,27 @@ class TestGitModuleSystemPrRouting:
     """Test that git_module routes system-pr correctly."""
 
     def test_system_pr_in_commands(self) -> None:
+        """The _COMMANDS registry includes the 'system-pr' verb."""
         from aipass.drone.apps.modules.git_module import _COMMANDS
 
         assert "system-pr" in _COMMANDS
 
     def test_get_help_includes_system_pr(self) -> None:
+        """Generic get_help() output mentions 'system-pr'."""
         from aipass.drone.apps.modules.git_module import get_help
 
         help_text = get_help()
         assert "system-pr" in help_text
 
     def test_get_help_system_pr_specific(self) -> None:
+        """get_help('system-pr') output mentions devpulse as the authorized caller."""
         from aipass.drone.apps.modules.git_module import get_help
 
         help_text = get_help("system-pr")
         assert "devpulse" in help_text
 
     def test_get_introspective_includes_plugin(self) -> None:
+        """get_introspective() output mentions the devpulse_ops plugin."""
         from aipass.drone.apps.modules.git_module import get_introspective
 
         intro = get_introspective()
@@ -272,6 +334,7 @@ class TestGitModuleSystemPrRouting:
 
     @patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
     def test_handle_system_pr_no_args(self, mock_verify: MagicMock) -> None:
+        """handle_command('system-pr', []) exits with code 1 and a Usage message."""
         from aipass.drone.apps.modules.git_module import handle_command
 
         result = handle_command("system-pr", [])
@@ -280,6 +343,7 @@ class TestGitModuleSystemPrRouting:
 
     @patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
     def test_handle_system_pr_unauthorized(self, mock_verify: MagicMock) -> None:
+        """handle_command propagates PermissionError as exit_code 1 with the message."""
         from aipass.drone.apps.modules.git_module import handle_command
 
         mock_verify.side_effect = PermissionError("not authorized")

src/aipass/seedgo/apps/handlers/file/__init__.py

@@ -33,3 +33,12 @@ def read_lines_safe(path: Path, n: int = 0, encoding: str = "utf-8") -> list[str
             return fh.readlines()
     except OSError:
         return []
+
+
+def write_text_safe(path: Path, text: str, encoding: str = "utf-8") -> bool:
+    """Write text to a file. Returns True on success, False on OSError."""
+    try:
+        path.write_text(text, encoding=encoding)
+        return True
+    except OSError:
+        return False

src/aipass/seedgo/apps/handlers/hooks/__init__.py

@@ -0,0 +1,47 @@
+# =================== AIPass ====================
+# Name: hooks/__init__.py
+# Description: Hook test runner — subprocess execution for hooks_ext module
+# Version: 1.0.0
+# Created: 2026-04-21
+# Modified: 2026-04-21
+# =============================================
+
+"""Hook test runner handler.
+
+Encapsulates subprocess execution so hooks_ext module stays
+at the display/coordination layer.
+"""
+
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+
+def run_pytest_file(test_file: Path, repo_root: Path, timeout: int = 60) -> tuple[int, int, float]:
+    """Run pytest on a single test file. Returns (passed, failed, duration_seconds)."""
+    t0 = time.monotonic()
+    proc = subprocess.run(
+        [sys.executable, "-m", "pytest", str(test_file), "--tb=no", "-q", "--no-header"],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        cwd=str(repo_root),
+    )
+    duration = time.monotonic() - t0
+    passed = 0
+    failed = 0
+    for line in proc.stdout.splitlines():
+        line = line.strip()
+        if "passed" in line or "failed" in line or "error" in line.lower():
+            m_passed = re.search(r"(\d+) passed", line)
+            m_failed = re.search(r"(\d+) failed", line)
+            m_error = re.search(r"(\d+) error", line)
+            if m_passed:
+                passed = int(m_passed.group(1))
+            if m_failed:
+                failed = int(m_failed.group(1))
+            if m_error:
+                failed += int(m_error.group(1))
+    return passed, failed, duration

src/aipass/seedgo/apps/modules/hooks.py

@@ -48,6 +48,9 @@ from aipass.cli.apps.modules import warning
 # JSON handler for tracking
 from aipass.seedgo.apps.handlers.json import json_handler
 
+# File handler — modules must not call open()/write_text() directly
+from aipass.seedgo.apps.handlers.file import write_text_safe
+
 # Extended subcommands (test + list)
 from aipass.seedgo.apps.modules.hooks_ext import cmd_hooks_list, cmd_hooks_test
 
@@ -206,10 +209,8 @@ def _cmd_probe_display(log_path: Path | None = None) -> None:
 def _run_headless_claude() -> int:
     """Spawn headless claude with a Read tool call. Returns exit code."""
     canary = Path("/tmp/probe_canary.txt")
-    try:
-        canary.write_text("probe canary 2026-04-20\n", encoding="utf-8")
-    except OSError as exc:
-        logger.info("hooks.py: could not write canary: %s", exc)
+    if not write_text_safe(canary, "probe canary 2026-04-20\n"):
+        logger.info("hooks.py: could not write canary")
 
     console.print("[dim]Spawning headless claude...[/dim]")
     # --permission-mode bypassPermissions is the AIPass-approved bypass flag
@@ -545,8 +546,10 @@ def handle_command(command: str, args: List[str]) -> bool:
     if command != "hooks":
         return False
 
-    # No args or help -> introspection
-    if not args or args[0] in ("--help", "-h", "help"):
+    if not args:
+        print_introspection()
+        return True
+    if args[0] in ("--help", "-h", "help"):
         print_introspection()
         return True
 

src/aipass/seedgo/apps/modules/hooks_ext.py

@@ -17,16 +17,13 @@ Extended subcommands for hooks.py — split out to keep hooks.py under 700 lines
 
 import glob as _glob
 import json
-import re
-import subprocess
-import sys
-import time
 from pathlib import Path
 
 from aipass.prax import logger
 from aipass.cli import console
 from aipass.cli.apps.modules import warning
-from aipass.seedgo.apps.handlers.file import read_lines_safe
+from aipass.seedgo.apps.handlers.file import read_lines_safe, read_text_safe
+from aipass.seedgo.apps.handlers.hooks import run_pytest_file
 from aipass.seedgo.apps.handlers.json import json_handler
 from rich.table import Table
 
@@ -36,25 +33,6 @@ from rich.table import Table
 # =============================================================================
 
 
-def _parse_pytest_counts(stdout: str) -> tuple[int, int]:
-    """Parse passed/failed counts from pytest -q output. Returns (passed, failed)."""
-    passed = 0
-    failed = 0
-    for line in stdout.splitlines():
-        line = line.strip()
-        if "passed" in line or "failed" in line or "error" in line.lower():
-            m_passed = re.search(r"(\d+) passed", line)
-            m_failed = re.search(r"(\d+) failed", line)
-            m_error = re.search(r"(\d+) error", line)
-            if m_passed:
-                passed = int(m_passed.group(1))
-            if m_failed:
-                failed = int(m_failed.group(1))
-            if m_error:
-                failed += int(m_error.group(1))
-    return passed, failed
-
-
 def cmd_hooks_test(repo_root: Path) -> None:
     """Run hook test suite, display per-file pass/fail table."""
     pattern = str(repo_root / "src" / "aipass" / "seedgo" / "tests" / "test_hooks*.py")
@@ -82,16 +60,7 @@ def cmd_hooks_test(repo_root: Path) -> None:
 
     for tf in test_files:
         stem = Path(tf).stem
-        t0 = time.monotonic()
-        proc = subprocess.run(
-            [sys.executable, "-m", "pytest", tf, "--tb=no", "-q", "--no-header"],
-            capture_output=True,
-            text=True,
-            timeout=60,
-            cwd=str(repo_root),
-        )
-        duration = time.monotonic() - t0
-        passed, failed = _parse_pytest_counts(proc.stdout)
+        passed, failed, duration = run_pytest_file(Path(tf), repo_root)
         total_passed += passed
         total_failed += failed
         status = (
@@ -126,10 +95,14 @@ def read_settings_file(path: Path) -> dict:
     """Read and parse a settings.json file. Returns {} on failure."""
     if not path.exists():
         return {}
+    text = read_text_safe(path)
+    if text is None:
+        logger.info("hooks_ext.py: could not read %s", path)
+        return {}
     try:
-        return json.loads(path.read_text(encoding="utf-8"))
+        return json.loads(text)
     except Exception as exc:
-        logger.info("hooks_ext.py: could not read %s: %s", path, exc)
+        logger.info("hooks_ext.py: could not parse %s: %s", path, exc)
         return {}