Merge pull request #562 from AIOSAI/work/system-dplan-0173-git-workflow-redesign-tier-based-access

feat(system): DPLAN-0173: Git workflow redesign — tier-based access, new handlers, hook reverts
This commit is contained in:
AIPass
2026-05-12 20:19:03 -07:00
committed by GitHub
26 changed files with 1700 additions and 374 deletions
+19 -64
View File
@@ -49,46 +49,22 @@ Secrets live outside the repo at `~/.secrets/aipass/` — API keys, tokens, cred
- `drone systems` — list all registered branches
- `drone --help` — full drone reference
# Git — Always on Main
# Git — Zero Direct Access
**ONE rule: every agent works on `main`. No exceptions.**
**You have no git access.** All `git` and `gh` commands are blocked at the project level. Drone is the only git interface.
You do not create branches. You do not `git checkout -b`. You do not tell another agent to "create a branch first." Branches only exist during the atomic window inside `drone @git system-pr` which: commits → creates branch → pushes → opens PR → **returns HEAD to main**. That command owns the branch lifecycle end to end. You own nothing about branches.
Read-only awareness (available to all branches):
- `drone @git status` — what changed in your branch directory
- `drone @git diff` — see the actual changes
- `drone @git log` — recent commit history
Workflow:
1. You're on main. Always.
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. 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.
Everything else — commits, pushes, merges, branch switching — is handled by devpulse. You build code, you run tests, you report results. Devpulse reviews and commits.
Never merge. Ever. User-merges-only. Past PRs, your own PRs, closed PRs — none of them auto-qualify. You fix, you PR, you stop.
Drone runs git via Python subprocess, so its operations bypass the settings.json deny rules. This is by design — drone is the gate, not a workaround.
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.
Local files are source of truth. When you edit a file, the state on disk IS reality. If the truth is wrong, fix it locally.
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.
Rules exist to help, not to control. These rules came from fixing actual bugs. Trust them.
Allowed:
- `drone @git status` — what changed?
- `drone @git sync` — pull latest main
- `drone @git system-pr "msg"` — ship your work (devpulse only)
- `drone @git merge <PR#>` — squash-merge a reviewed PR (devpulse only)
- `drone @git smart-sync` — fetch + rebase (devpulse only)
- `drone @git fix` — repair broken git states (devpulse only)
- `git status`, `git diff`, `git log`, `git branch` (list), `git tag` (list), `git remote` (list/show) — read-only, always fine
Mechanically blocked by the `git_gate.py` PreToolUse hook (applies to ALL sessions including dispatched agents — bypassPermissions does not skip hooks):
- All raw `git` write verbs: `commit`, `push`, `pull`, `merge`, `rebase`, `reset`, `checkout`, `switch`, `cherry-pick`, `revert`, `rm`, `mv`, `restore`, `clean`, `config`, `stash drop|clear|pop|apply`
- Destructive `git branch` flags only (`-d`, `-D`, `-m`, `-M`, `--delete`, `--move`, `--set-upstream-to`, `--unset-upstream`). Read-only branch listing is allowed.
- Destructive `git tag` flags only (`-d`, `--delete`, `-f`, `--force`). Tag listing is allowed.
- Destructive `git remote` subcommands (`add`, `remove`, `rename`, `set-url`, `prune`). Remote listing/show is allowed.
- All raw `gh` write subcommands (`pr`, `issue`, `repo`, `release`, `workflow`, `run`, `cache`, `secret`, `variable`, `gist`) and any `gh api` call. Exception: project owners with `citizenship.owner: true` in their passport bypass gh blocking.
- Edits to `**/.claude/settings*.json`, `**/.claude/hooks/**`, `**/.git/hooks/**` (the enforcement layer itself)
- Use `drone @git pr "msg"` instead. Drone calls git via Python subprocess so its operations don't pass through this hook.
If `drone @git system-pr` fails to return HEAD to main, that's a drone bug — report it, don't work around it by staying on a branch.
The `git_gate.py` PreToolUse hook enforces this mechanically — it applies to ALL sessions including dispatched agents. bypassPermissions does not skip hooks.
# aipass init
@@ -186,44 +162,23 @@ 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.** All git/gh commands are denied at the project level. Drone handles everything via Python subprocess (bypasses settings.json deny rules by design).
**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.
Manual git is not a shortcut. It is a trap. Drone exists so you don't get stuck. Use it.
Always work on main. Edit files in your branch directory on the main branch. When ready to submit:
- `drone @git pr "description"` — full PR workflow (lock, branch, commit, push, PR, back to main)
You have read-only awareness via drone:
- `drone @git status` — what changed in your branch directory
- `drone @git sync` — pull latest main
- `drone @git lock` — check the PR lock state
- `drone @git --help` — full git reference
- `drone @git diff` — see the actual diff
- `drone @git log` — recent commits
`drone @git pr` does everything atomically: acquires a lock (so no other branch can PR simultaneously), creates a feature branch, stages only your files, commits with your Co-Authored-By signature, pushes, creates the PR on GitHub, returns to main, releases the lock.
All write operations (commit, push, merge, checkout) are restricted to devpulse via tier-based access control. Dispatched agents build code and run tests — devpulse reviews the diff and commits.
**Blocked system-wide via `.claude/settings.json` permission gate:** `git checkout*` (any form — switch, discard, new branch), `git add -f*`, `git add --force*`. These are denied for every agent including devpulse. Use `drone @git sync` to switch to main, `drone @git fix` to recover from broken states.
**Mechanically blocked via `.git/hooks/pre-commit`:** `git commit` is rejected on any branch except main (also catches detached HEAD). `git push`, `gh pr create` — go through drone.
**Allowed read-only:** `git status`, `git diff`, `git log`, `git branch` (list), `git tag` (list), `git remote` (list/show), `git stash` (safe transient save).
**If `drone @git pr` fails because the PR lock is held**, wait 30 seconds and retry. Keep retrying until the lock clears — do not skip the PR step, do not commit directly to main, do not give up. The lock means another agent is mid-PR; it will release shortly. `drone @git lock` shows the current lock state.
Never merge. Only devpulse or the user merges PRs. If your PR gets feedback, fix it and run `drone @git pr` again.
Local main is always ahead of origin — that's normal. `drone @git pr` commits on local main first, then pushes a feature branch for the PR. Don't `git pull` to fix it. The user merges and pulls when they choose.
Respect .gitignore — only commit what `git status` shows. Gitignored patterns like `.trinity/`, `.ai_mail.local/`, `DPLAN-*`, `*.local.*`, `logs/`, `.chroma/` are ignored for a reason. Don't go looking for files to commit. Changes drive commits, not file existence.
**Before you PR, run ruff on your diff.** Two commands, every time, no exceptions:
**Before submitting code, run ruff:**
```
ruff check --fix src/ tests/ # Auto-fix lint errors (unused imports, f-strings, etc.)
ruff format src/ tests/ # Auto-format (whitespace, line breaks, quote style)
ruff check --fix src/ tests/
ruff format src/ tests/
```
CI runs both as a gate — if you don't run them locally, CI catches it and your PR sits red until someone fixes it. Make this part of muscle memory: edit code → run ruff → `drone @git pr`. It takes two seconds and prevents the silent-debt pattern where drift accumulates across hundreds of files and someone has to run one giant sweep PR to clear it. This is a habit, not a safety net — infrastructure will always catch drift, but habits prevent it in the first place.
Respect .gitignore — only track what `git status` shows. Gitignored patterns like `.trinity/`, `.ai_mail.local/`, `DPLAN-*`, `*.local.*`, `logs/`, `.chroma/` are ignored for a reason.
# How to Work
+34
View File
@@ -137,7 +137,41 @@ claude --debug hooks --debug-file /tmp/debug.log
Type `/hooks` inside a Claude session — shows all hooks with source labels
(`[User]`, `[Project]`, `[Local]`).
## git_gate.py — Known Limitations
`git_gate.py` is the **only real enforcement layer** for blocking raw git/gh commands.
`Bash(git *)` deny rules in `settings.json` **do not work** — the permission gate
silently skips content-specific deny patterns. The hook is what actually blocks.
### What it blocks
- Bare `git`/`gh` commands (`git status`, `gh pr list`)
- Prefixed variants (`env git status`)
- Drone tier system enforces per-branch write restrictions on top
### Known bypass vectors (not caught by the hook)
These are inherent limitations of regex-based command scanning:
1. **Python subprocess** — `python3 -c 'import subprocess; subprocess.run(["git", "status"])'`
`git` never appears as a bare word in the scanned command
2. **Full binary path** — `/usr/bin/git status`
Lookbehind `(?<![@\w/.])` excludes `/` before `git`
3. **Nested bash with quotes** — `bash -c 'git log'`
Hook strips quoted strings before scanning, so `git` inside quotes is invisible
4. **Script file execution** — Write git commands to `/tmp/script.sh`, then run it
`git` is inside the file content, not the Bash command
5. **Subshell expansion** — `$(which git) status`
Hook sees `$(which git)` not bare `git`
### Why this is acceptable
These bypasses require deliberate circumvention — no agent will accidentally hit them.
The hook catches all natural/obvious git usage patterns. Combined with the drone tier
system (only devpulse has write-level git access), the defense is layered.
## Related
- **DPLAN-0173** — Git workflow redesign (whitelist-only drone git)
- **DPLAN-0167** — Hook testing framework
- **DPLAN-0166** — Hook audit + CI health
- **DPLAN-0139** — Hook overhaul + single-path enforcement
+2 -31
View File
@@ -10,37 +10,8 @@
],
"deny": [
"EnterPlanMode",
"Bash(git add*)",
"Bash(git commit*)",
"Bash(git push*)",
"Bash(git pull*)",
"Bash(git merge*)",
"Bash(git rebase*)",
"Bash(git reset*)",
"Bash(git checkout*)",
"Bash(git switch*)",
"Bash(git branch*)",
"Bash(git cherry-pick*)",
"Bash(git stash*)",
"Bash(git tag*)",
"Bash(git revert*)",
"Bash(git rm*)",
"Bash(git mv*)",
"Bash(git clean*)",
"Bash(git restore*)",
"Bash(git apply*)",
"Bash(gh pr create*)",
"Bash(gh pr merge*)",
"Bash(gh pr close*)",
"Bash(gh pr edit*)",
"Bash(gh pr comment*)",
"Bash(gh pr review*)",
"Bash(gh issue create*)",
"Bash(gh issue close*)",
"Bash(gh issue edit*)",
"Bash(gh issue comment*)",
"Bash(gh repo *)",
"Bash(gh api *)",
"Bash(git *)",
"Bash(gh *)",
"Read(/home/patrick/Patrick-Personal/**)",
"Edit(/home/patrick/Patrick-Personal/**)",
"Write(/home/patrick/Patrick-Personal/**)",
+2 -2
View File
@@ -2,9 +2,9 @@ name: CI
on:
push:
branches: [main]
branches: [main, dev]
pull_request:
branches: [main]
branches: [main, dev]
jobs:
lint:
+2 -2
View File
@@ -2,9 +2,9 @@ name: Security Scan
on:
push:
branches: [main]
branches: [main, dev]
pull_request:
branches: [main]
branches: [main, dev]
schedule:
- cron: "0 6 * * 1"
+1 -1
View File
@@ -3,7 +3,7 @@ name: Windows Test
on:
workflow_dispatch:
push:
branches: [main]
branches: [main, dev]
paths:
- 'setup.sh'
- 'src/aipass/*/apps/handlers/__init__.py'
@@ -81,12 +81,11 @@ drone @memory archive # Archive memories to vector store
drone @memory search <query> # Search archived memories
```
### Git Workflow
### Git
```
drone @git pr 'description' # Create a pull request
drone @git status # Git status (branch-scoped)
drone @git sync # Sync with main
drone @git lock / unlock # Lock/unlock the repo
drone @git diff # See changes in your branch
drone @git log # Recent commits
```
### Infrastructure
@@ -81,12 +81,11 @@ drone @memory archive # Archive memories to vector store
drone @memory search <query> # Search archived memories
```
### Git Workflow
### Git
```
drone @git pr 'description' # Create a pull request
drone @git status # Git status (branch-scoped)
drone @git sync # Sync with main
drone @git lock / unlock # Lock/unlock the repo
drone @git diff # See changes in your branch
drone @git log # Recent commits
```
### Infrastructure
@@ -15,7 +15,7 @@ You are DEVPULSE — Patrick's primary AI collaborator and orchestration hub for
- **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.
- **Start watchdog after any dispatch.** Use Monitor tool: `drone @devpulse watchdog agent @target` (timeout_ms=600000, persistent=false). This streams state changes live into the conversation. Never use run_in_background for watchdog — notifications get buried in task files.
## Branch Experts — Ask Before Rebuilding
@@ -32,36 +32,35 @@ 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, Never Merge
## Git Workflow — Dev Branch, Drone Only, You Are the Gatekeeper
**Three rules, in order:**
**You are the only branch with git write access.** All git/gh commands are blocked at the project level (`Bash(git *)`, `Bash(gh *)`). Drone bypasses this via subprocess — and drone's tier system only grants write access to devpulse.
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.
**Three rules:**
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.
1. **Work on dev, merge to main when satisfied.** All work happens on the `dev` branch. Stack changes until a feature is complete. Test in Docker against dev. When satisfied: `drone @git merge dev` squash-merges to main.
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.
2. **You commit, agents don't.** Dispatched agents build code and run tests. They report results. You review the diff and commit via `drone @git commit`. No agent PRs.
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.
3. **Local files are source of truth.** When you edit a file, the state on disk IS reality. If the truth is wrong, fix it locally first, then commit.
```
drone @git system-pr "description" # System-wide PR (devpulse only) — commit, branch, push, PR, back to main
drone @git pr "description" # Branch-scoped PR (any branch) — dispatched agents use this
drone @git status # What changed? (all branches can use)
drone @git diff # See the diff (all branches can use)
drone @git log # Recent commits (all branches can use)
drone @git commit "description" # Commit changes (devpulse only)
drone @git checkout dev # Switch to dev branch (devpulse only)
drone @git checkout main # Switch to main (devpulse only)
drone @git system-pr "description" # System-wide PR (devpulse only)
drone @git merge <PR#> # Merge a PR (devpulse only, user must request)
drone @git smart-sync # Fetch + rebase if behind (devpulse only)
drone @git sync # Pull latest (devpulse only)
drone @git smart-sync # Fetch + rebase (devpulse only)
drone @git fix # Fix broken git states (devpulse only)
drone @git status # What changed?
drone @git sync # Pull latest main
drone @git lock # Check PR lock status
```
**Dispatch briefs must say `drone @git pr`, not `drone @git system-pr`.** system-pr is devpulse-only. Agents dispatched to branches use `drone @git pr` for their own branch-scoped PRs.
**Dispatch briefs must NOT reference any git commands.** Agents have zero git access. They build, test, report. You handle git.
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 commands require `.trinity/passport.json` in the CWD hierarchy. Always run drone commands from this directory.
## Dispatch — Fresh vs Continue
@@ -107,7 +106,7 @@ Watchdog is a real devpulse module now (not a bash one-liner). After dispatching
**Pattern:**
```bash
drone @ai_mail dispatch @target "Subject" "Body"
drone @devpulse watchdog agent @target # run_in_background: true
drone @devpulse watchdog agent @target # use Monitor tool (not run_in_background)
```
The handler resolves `@target` → branch path → `.ai_mail.local/.dispatch.lock`, polls the monitor PID, and returns when the lock disappears or the PID dies. Crash vs success is distinguished by `last_bounce.json`. Default timeout 1800s — override with `--timeout SECONDS`.
@@ -0,0 +1,57 @@
# Session Wrap-Up
Purpose: Button up everything at the end of a session — or before a /compact. Memories, plans, git — all tidy. Works for both closing out a chat and preparing for compaction.
**Workflow:** `/prep` → review output → close chat or `/compact`
## Execution
1. Read `.trinity/passport.json` first — re-absorb your identity before writing anything
2. Do ALL of the following, then confirm what was updated
## 1. Memories
Each memory file plays a distinct role. Update based on what actually changed this session.
- **`.trinity/passport.json`** — IDENTITY. Who you are: role, capabilities, principles. Only update if identity genuinely evolved this session.
- **`.trinity/local.json`** — YOUR MEMORY. Add/update session entry with a summary of work done. Add key_learnings for anything learned. Trim oldest sessions if over 20.
- **`.trinity/observations.json`** — YOUR MEMORY OF THE USER. Collaboration insights, preferences, friction points. Skip if nothing new about the user this session.
- **`STATUS.local.md`** — PUBLIC STATUS BEACON. Current work, known issues, todos, notepad. Auto-synced to central STATUS.md on PR events — this is how other branches see you. Keep Current Work accurate.
## 2. Active Plans
- Check any DPLANs or FPLANs referenced in this session
- Update their execution logs, status, decision logs with current state
- If a plan was completed, note it (but don't close — the user does that)
## 3. Git State
- Run `git status` — report uncommitted changes
- If there's a logical commit waiting, suggest it (don't commit without asking)
- Note the current branch and any open PRs
## 4. Inbox
- Run `drone @ai_mail inbox 2>/dev/null` — report any unread emails
- Close any that were already processed but not formally closed
## 5. Loose Ends
- Flag anything in-flight: running background agents, dispatched branches waiting for replies, pending decisions
- If anything can't survive compaction (e.g., agent IDs needed for resume), write it to STATUS.local.md Notepad
## Confirm
List everything updated. Format:
```
Prep complete:
- local.json: [what was added]
- observations.json: [updated / skipped]
- STATUS.local.md: [updated / skipped]
- Plans: [which ones updated]
- Git: [branch, uncommitted count, suggestion]
- Inbox: [count, action taken]
- Loose ends: [any flagged]
Ready to close out or /compact.
```
@@ -2,38 +2,8 @@
"permissions": {
"allow": [],
"deny": [
"Bash(git reset*)",
"Bash(git rebase*)",
"Bash(git merge*)",
"Bash(git config*)",
"Bash(git push --force*)",
"Bash(git push -f *)",
"Bash(git checkout -- *)",
"Bash(git checkout .*)",
"Bash(git restore --staged*)",
"Bash(git restore .*)",
"Bash(git clean*)",
"Bash(git branch -D*)",
"Bash(git stash drop*)",
"Bash(git stash clear*)",
"Bash(rm -rf*)",
"Bash(rm -r *)",
"Bash(git checkout -b*)",
"Bash(git commit*)",
"Bash(git push*)",
"Bash(git add -f*)",
"Bash(gh pr close*)",
"Bash(gh pr create*)",
"Bash(*&& git commit*)",
"Bash(*&& git push*)",
"Bash(*&& git checkout -b*)",
"Bash(*&& git add -f*)",
"Bash(*&& git reset*)",
"Bash(*&& gh pr close*)",
"Bash(*&& gh pr create*)",
"Bash(*; git commit*)",
"Bash(*; git push*)",
"Bash(*; git checkout -b*)"
"Bash(rm -r *)"
],
"ask": []
},
@@ -0,0 +1,11 @@
{
"metadata": {
"id": "6e3e877e-56ed-4ec5-892f-81073257dc90",
"name": "DEVPULSE",
"version": "1.0.0",
"created": "2026-05-12",
"last_updated": "2026-05-12",
"total_branches": 0
},
"branches": []
}
+27
View File
@@ -154,6 +154,21 @@
"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_git_access.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_access.py",
"standard": "encapsulation",
"reason": "Test file imports handlers directly to test their public interface. Unit tests require direct access to implementation components."
},
{
"file": "tests/test_git_access.py",
"standard": "documentation",
"reason": "Test class docstrings describe intent; per-method docstrings would be noise for assertion-named test methods."
},
{
"file": "tests/test_system_pr.py",
"standard": "architecture",
@@ -179,6 +194,18 @@
"standard": "unused_function",
"lines": [232, 263],
"reason": "Public CRUD API surface (update_command, command_exists) — tested, available for programmatic use. No CLI subcommand wired yet."
},
{
"file": "apps/handlers/git/pr_handler.py",
"standard": "unused_function",
"lines": [108],
"reason": "create_pr() is deprecated per FPLAN-0210 — pr command blocked at auth tier. Handler kept for backwards compatibility; still tested in test_git_module.py."
},
{
"file": "apps/plugins/devpulse_ops/auth.py",
"standard": "unused_function",
"lines": [84],
"reason": "verify_caller() replaced by verify_git_access() for centralized auth. Still exported as public API and tested directly in test_system_pr.py."
}
],
"notes": {
+45 -13
View File
@@ -14,7 +14,7 @@
### What I Do
- Resolve `@branch` symbolic names to absolute paths via `AIPASS_REGISTRY.json`
- Route commands to registered branches and internal modules
- Manage git workflows: PR creation, branch sync, lock management, merge
- Manage git workflows: tier-based access (global read-only, owner write), commit, diff, log, sync, merge
- Discover and scan available commands across the system
- Provide `drone systems` introspection of all registered components
- Support external AIPass projects via dual registry lookup and module fallback
@@ -33,20 +33,34 @@ drone @seedgo audit aipass # Route "audit aipass" to seedgo
drone @module --help # Show help for any module
drone systems # List all registered modules and branches
# Git workflow
drone @git pr "description" # Create a PR from current branch
# Git workflow — global tier (all branches)
drone @git status # Git status scoped to branch directory
drone @git sync # Pull latest main with --rebase
drone @git sync --autostash # Sync with autostash for dirty trees
drone @git lock / unlock # Atomic branch lockfile
drone @git diff # Show git diff for your branch
drone @git diff --staged # Show staged changes
drone @git log # Show recent git log (default: 10)
drone @git log 20 # Show last 20 commits
drone @git lock # Check lock status
drone @git issue list # Passthrough to gh issue list
drone @git issue view 42 # Passthrough to gh issue view 42
drone @git run list # Passthrough to gh run list
drone @git workflow list # Passthrough to gh workflow list
# Git workflow (devpulse-authorized only)
# Git workflow — owner tier (devpulse only)
drone @git commit "message" # Commit staged changes
drone @git commit "msg" --all # Stage tracked files and commit
drone @git checkout main # Switch to main branch
drone @git sync # Checkout main and pull
drone @git sync --autostash # Sync with autostash for dirty trees
drone @git unlock --force # Force-release the PR lock
drone @git system-pr "desc" # System-wide PR across all tracked changes
drone @git merge <PR#> # Straight-merge a PR and sync local main
drone @git smart-sync # Fetch + detect divergence + rebase
drone @git fix # Auto-fix stuck rebase / detached HEAD
drone @git fix --dry-run # Detect issues without fixing
# Git workflow — deprecated
drone @git pr # DEPRECATED — returns error message
# Command discovery
drone scan @branch # Discover available commands in a branch
drone activate @branch # Scan + register all commands as shortcuts
@@ -125,7 +139,7 @@ drone/
│ │ ├── module_registry.py # Internal module routing
│ │ ├── registry.py # Registry query operations
│ │ ├── commands.py # Custom command shortcut orchestrator
│ │ ├── git_module.py # Git workflow (9 commands + plugin routing)
│ │ ├── git_module.py # Git workflow (tier-based access, 13 commands)
│ │ └── scan.py # Branch command scanning
│ ├── handlers/ # Implementation details
│ │ ├── executor.py # Safe subprocess execution (timeout, no shell)
@@ -146,8 +160,13 @@ drone/
│ │ │ ├── lookup.py # Greedy multi-word matching
│ │ │ └── formatters.py # Rich output for command lists
│ │ └── git/
│ │ ├── auth.py # Tier-based access (verify_git_access)
│ │ ├── lock_handler.py # Atomic lockfile (O_CREAT|O_EXCL)
│ │ ├── pr_handler.py # 10-step PR workflow with scoped staging
│ │ ├── pr_handler.py # DEPRECATED — returns error message
│ │ ├── diff_handler.py # Scoped git diff (--staged support)
│ │ ├── log_handler.py # Scoped git log (configurable count)
│ │ ├── commit_handler.py # Commit staged changes (--all support)
│ │ ├── checkout_handler.py # Branch switching (main/dev guard)
│ │ ├── status_handler.py # Scoped git status (subprocess)
│ │ ├── status_handler_gitpython.py # [prototype] DPLAN-0140 Phase 1, not wired in
│ │ └── sync_handler.py # Safe main sync (--autostash support)
@@ -162,7 +181,7 @@ drone/
│ └── hook_sounds_plugin.py # Toggle notification sounds on/off
├── docs/ # Public documentation
├── docs.local/ # Investigation reports and policies
└── tests/ # 530 tests across 20 test files
└── tests/ # 704 tests across 21 test files
```
### Routing Flow
@@ -185,6 +204,19 @@ Drone routes to two kinds of modules:
External modules are declared in `apps/handlers/routing_config.json` with entry points, descriptions, and versions.
### Git Access Tiers
Auth centralized via `verify_git_access()` in `apps/handlers/git/auth.py`. Two tiers:
| Tier | Who | Commands |
|------|-----|----------|
| **Global** | All branches | `status`, `diff`, `log`, `lock` |
| **Owner** | `devpulse` only | `commit`, `checkout`, `sync`, `unlock`, `system-pr`, `merge`, `smart-sync`, `fix` |
- `pr` is **deprecated** — returns an error message directing to devpulse
- Auth is checked once at the top of `git_module.handle_command()` before any handler is called
- Unauthorized commands return a clear "Access denied" message with the caller's tier
### Git Main-Only Enforcement
All agents work on `main`. Branch creation is only allowed inside `drone @git system-pr`, which:
@@ -280,12 +312,12 @@ Tip: set AIPASS_HOME=/path/to/AIPass to access all branches
## Testing
530 tests across 20 test files, covering all layers:
704 tests across 21 test files, covering all layers:
| Area | Files | Tests |
|------|-------|-------|
| Core routing | `test_resolver.py`, `test_router.py`, `test_activation.py` | ~128 |
| Git operations | `test_git_module.py`, `test_system_pr.py`, `test_devpulse_plugins.py` | ~95 |
| Git operations | `test_git_module.py`, `test_system_pr.py`, `test_devpulse_plugins.py`, `test_git_access.py` | ~150 |
| Handlers | `test_executor.py`, `test_registry_handler.py`, `test_discovery.py` | ~99 |
| Infrastructure | `test_generic_adapter.py`, `test_module_registry.py`, `test_config.py` | ~66 |
| Features | `test_commands.py`, `test_scan.py`, `test_hook_sounds.py`, `test_json_handler.py` | ~125 |
@@ -304,7 +336,7 @@ Run tests: `cd src/aipass/drone && python -m pytest tests/ -q`
---
**Seedgo:** 100% (34/34) | **Tests:** 530 pass, 4 skip | **Last Updated:** 2026-04-22
**Seedgo:** 99% | **Tests:** 704 pass, 4 skip | **Last Updated:** 2026-05-12
---
[← Back to AIPass](../../../README.md)
@@ -1,6 +1,10 @@
"""Git workflow handlers — lock, status, sync, PR."""
"""Git workflow handlers — lock, status, sync, PR, diff, log, commit, checkout."""
from . import lock_handler as lock_handler # explicit re-export for type checkers
from . import status_handler as status_handler
from . import sync_handler as sync_handler
from . import pr_handler as pr_handler
from . import diff_handler as diff_handler
from . import log_handler as log_handler
from . import commit_handler as commit_handler
from . import checkout_handler as checkout_handler
@@ -0,0 +1,84 @@
# =================== AIPass ====================
# Name: checkout_handler.py
# Description: Branch checkout handler with hard guard
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Branch checkout handler with hard guard."""
from __future__ import annotations
import subprocess
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
_ALLOWED_TARGETS = ("main", "dev")
def checkout_branch(target: str) -> dict:
"""Switch to target branch (main or dev only)."""
if target not in _ALLOWED_TARGETS:
allowed = ", ".join(_ALLOWED_TARGETS)
return {
"stdout": "",
"stderr": f"Checkout denied: only {allowed} branches are allowed. Got '{target}'.",
"exit_code": 1,
"current_branch": "",
}
repo_root = find_repo_root()
try:
status = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if status.stdout.strip():
return {
"stdout": "",
"stderr": "Cannot checkout: uncommitted changes in working tree. Commit or stash first.",
"exit_code": 1,
"current_branch": "",
}
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git status check failed: %s", exc)
return {
"stdout": "",
"stderr": f"Failed to check working tree status: {exc}",
"exit_code": 1,
"current_branch": "",
}
try:
result = subprocess.run(
["git", "checkout", target],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git checkout failed: %s", exc)
return {
"stdout": "",
"stderr": f"git checkout failed: {exc}",
"exit_code": 1,
"current_branch": "",
}
json_handler.log_operation(
"checkout_branch",
{"target": target, "exit_code": result.returncode},
)
return {
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"exit_code": result.returncode,
"current_branch": target if result.returncode == 0 else "",
}
@@ -0,0 +1,111 @@
# =================== AIPass ====================
# Name: commit_handler.py
# Description: Commit handler with scoped staging
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Commit handler with scoped staging."""
from __future__ import annotations
import subprocess
from pathlib import Path
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
def stage_branch_dir(branch_dir: Path, repo_root: Path | None = None) -> dict:
"""Stage all changes under branch_dir.
Shared utility used by both commit_handler and pr_handler.
"""
if repo_root is None:
repo_root = find_repo_root()
try:
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
except ValueError:
logger.warning(
"stage_branch_dir: branch_dir %s not relative to repo root %s",
branch_dir,
repo_root,
)
rel_dir = branch_dir
add_result = subprocess.run(
["git", "add", str(rel_dir) + "/"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if add_result.returncode != 0:
return {
"success": False,
"rel_dir": rel_dir,
"message": f"Failed to stage files: {add_result.stderr.strip()}",
}
return {"success": True, "rel_dir": rel_dir, "message": f"Staged files under {rel_dir}"}
def commit_changes(
message: str,
branch_dir: Path | None = None,
all_files: bool = False,
) -> dict:
"""Commit staged changes or all changes under branch_dir."""
repo_root = find_repo_root()
if all_files and branch_dir:
stage_result = stage_branch_dir(branch_dir, repo_root)
if not stage_result["success"]:
return {"stdout": "", "stderr": stage_result["message"], "exit_code": 1}
diff_check = subprocess.run(
["git", "diff", "--cached", "--quiet"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if diff_check.returncode == 0:
return {
"stdout": "",
"stderr": "Nothing to commit: no changes staged",
"exit_code": 1,
}
try:
cmd = ["git", "commit", "-m", message]
if branch_dir:
try:
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
except ValueError as exc:
logger.warning("commit_changes: branch_dir not relative to repo root: %s", exc)
rel_dir = branch_dir
cmd.extend(["--", str(rel_dir) + "/"])
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git commit failed: %s", exc)
return {"stdout": "", "stderr": f"git commit failed: {exc}", "exit_code": 1}
json_handler.log_operation(
"commit_changes",
{"message": message, "all_files": all_files, "exit_code": result.returncode},
)
return {
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"exit_code": result.returncode,
}
@@ -0,0 +1,79 @@
# =================== AIPass ====================
# Name: diff_handler.py
# Description: Scoped git diff for branch directories
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Scoped git diff for branch directories."""
from __future__ import annotations
import subprocess
from pathlib import Path
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
def get_branch_diff(branch_dir: Path, staged: bool = False) -> dict:
"""Get git diff filtered to files under branch_dir."""
repo_root = find_repo_root()
cmd = ["git", "diff"]
if staged:
cmd.append("--staged")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git diff failed: %s", exc)
return {"diff": "", "files_changed": 0, "message": f"git diff failed: {exc}"}
if result.returncode != 0:
return {
"diff": "",
"files_changed": 0,
"message": f"git diff error: {result.stderr.strip()}",
}
try:
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
except ValueError:
logger.warning(
"get_branch_diff: branch_dir %s not relative to repo root %s",
branch_dir,
repo_root,
)
rel_dir = branch_dir
rel_prefix = rel_dir.as_posix() + "/"
filtered_lines: list[str] = []
include_block = False
files_changed = 0
for line in result.stdout.splitlines():
if line.startswith("diff --git"):
include_block = rel_prefix in line
if include_block:
files_changed += 1
if include_block:
filtered_lines.append(line)
filtered_diff = "\n".join(filtered_lines)
message = f"{files_changed} file(s) changed under {rel_dir}"
json_handler.log_operation(
"get_branch_diff",
{"branch_dir": str(branch_dir), "files_changed": files_changed, "staged": staged},
)
logger.info(message)
return {"diff": filtered_diff, "files_changed": files_changed, "message": message}
@@ -0,0 +1,47 @@
# =================== AIPass ====================
# Name: log_handler.py
# Description: Git log handler
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Git log handler."""
from __future__ import annotations
import subprocess
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
def get_git_log(count: int = 10) -> dict:
"""Get recent git log entries."""
repo_root = find_repo_root()
try:
result = subprocess.run(
["git", "log", "--oneline", f"-{count}"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git log failed: %s", exc)
return {"entries": [], "count": 0, "message": f"git log failed: {exc}"}
if result.returncode != 0:
return {
"entries": [],
"count": 0,
"message": f"git log error: {result.stderr.strip()}",
}
entries = [line for line in result.stdout.splitlines() if line.strip()]
json_handler.log_operation("get_git_log", {"count": len(entries)})
logger.info("Retrieved %d log entries", len(entries))
return {"entries": entries, "count": len(entries), "message": f"{len(entries)} log entries"}
@@ -28,6 +28,7 @@ from aipass.drone.apps.handlers.git.lock_handler import (
find_repo_root,
release_lock,
)
from aipass.drone.apps.handlers.git.commit_handler import stage_branch_dir
def _has_credential_helper() -> bool:
@@ -166,24 +167,12 @@ def create_pr(branch_name: str, description: str, branch_dir: Path) -> dict:
lock_acquired = True
# Step 3: Stage only files under branch_dir (on main)
try:
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
except ValueError:
logger.warning(
"create_pr: branch_dir %s not relative to repo root %s, using absolute", branch_dir, repo_root
)
rel_dir = branch_dir
add_result = subprocess.run(
["git", "add", str(rel_dir) + "/"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if add_result.returncode != 0:
result["message"] = f"Failed to stage files: {add_result.stderr.strip()}"
stage_result = stage_branch_dir(branch_dir, repo_root)
if not stage_result["success"]:
result["message"] = stage_result["message"]
logger.error(result["message"])
return result
rel_dir = stage_result["rel_dir"]
# Step 4: Check if anything was staged
diff_check = subprocess.run(
+275 -157
View File
@@ -16,19 +16,47 @@ the module orchestrator, routing git commands to the appropriate handlers.
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git import lock_handler, status_handler, sync_handler, pr_handler
from aipass.drone.apps.handlers.git import (
lock_handler,
status_handler,
sync_handler,
diff_handler,
log_handler,
commit_handler,
checkout_handler,
)
DRONE_MODULE = {
"name": "git",
"version": "1.0.0",
"description": "Git workflow — PR, status, sync, lock management",
"version": "2.0.0",
"description": "Git workflow — tier-based access, status, diff, log, commit, checkout, sync, lock",
}
_COMMANDS = ("pr", "status", "sync", "lock", "unlock", "system-pr", "merge", "smart-sync", "fix")
_COMMANDS = (
"status",
"diff",
"log",
"lock",
"issue",
"run",
"workflow",
"commit",
"checkout",
"sync",
"unlock",
"system-pr",
"merge",
"smart-sync",
"fix",
"pr",
)
_GH_PASSTHROUGH_COMMANDS = ("issue", "run", "workflow")
def _detect_branch_dir() -> tuple[str, Path] | None:
@@ -63,42 +91,66 @@ def _detect_branch_dir() -> tuple[str, Path] | None:
def handle_command(command: str | None = None, args: list[str] | None = None) -> dict:
"""Route a git command to the appropriate handler.
Auth is centralized: verify_git_access() is called once at the top,
before any routing. Global-tier commands pass for all callers;
owner-tier commands require devpulse.
Args:
command: The subcommand (pr, status, sync, lock, unlock).
command: The subcommand (status, diff, log, commit, checkout, etc.).
args: Optional list of arguments.
Returns:
Dict with stdout, stderr, and exit_code.
"""
if not args:
if command is None:
print_introspection()
return {"stdout": "", "stderr": "", "exit_code": 0}
args = []
if command in ("--help", "-h") or (args and args[0] in ("--help", "-h")):
print_help()
return {"stdout": "", "stderr": "", "exit_code": 0}
json_handler.log_operation("git_handle_command", {"command": command, "args": args})
if command is None:
print_introspection()
return {"stdout": "", "stderr": "", "exit_code": 0}
if command == "system-pr":
return _handle_system_pr(args)
if command == "merge":
return _handle_merge(args)
if command == "smart-sync":
return _handle_smart_sync(args)
if command == "fix":
return _handle_fix(args)
if command == "pr":
return _handle_pr(args)
cmd: str = command
try:
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_git_access
caller = verify_git_access(cmd)
except PermissionError as exc:
logger.error("git access denied: %s", exc)
return {"stdout": "", "stderr": str(exc), "exit_code": 1}
json_handler.log_operation("git_handle_command", {"command": command, "args": args, "caller": caller})
if command in _GH_PASSTHROUGH_COMMANDS:
return _handle_gh_passthrough(command, args)
if command == "status":
return _handle_status()
if command == "sync":
return _handle_sync(args)
if command == "diff":
return _handle_diff(args)
if command == "log":
return _handle_log(args)
if command == "lock":
return _handle_lock()
if command == "commit":
return _handle_commit(args)
if command == "checkout":
return _handle_checkout(args)
if command == "sync":
return _handle_sync(args)
if command == "unlock":
return _handle_unlock(args)
if command == "system-pr":
return _handle_system_pr(args, caller)
if command == "merge":
return _handle_merge(args, caller)
if command == "smart-sync":
return _handle_smart_sync(caller)
if command == "fix":
return _handle_fix(args, caller)
if command == "pr":
return {"stdout": "", "stderr": "Agent PRs are deprecated.", "exit_code": 1}
available = ", ".join(_COMMANDS)
return {
@@ -108,8 +160,39 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
}
def _handle_system_pr(args: list[str]) -> dict:
"""Handle the system-pr subcommand (devpulse-only)."""
def _handle_gh_passthrough(subcommand: str, args: list[str]) -> dict:
"""Pass through to gh CLI for issue, run, and workflow subcommands."""
cmd = ["gh", subcommand] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60,
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.returncode,
}
except FileNotFoundError as exc:
logger.warning("gh CLI not found: %s", exc)
return {
"stdout": "",
"stderr": "gh CLI not found. Install: https://cli.github.com/",
"exit_code": 1,
}
except subprocess.TimeoutExpired as exc:
logger.warning("gh %s timed out: %s", subcommand, exc)
return {
"stdout": "",
"stderr": f"gh {subcommand} timed out after 60s",
"exit_code": 1,
}
def _handle_system_pr(args: list[str], caller: str) -> dict:
"""Handle the system-pr subcommand (owner-tier, auth pre-checked)."""
if not args:
return {
"stdout": "",
@@ -120,7 +203,6 @@ def _handle_system_pr(args: list[str]) -> dict:
description = " ".join(args)
try:
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
from aipass.drone.apps.plugins.devpulse_ops.pr_plugin import create_system_pr
except ImportError as exc:
logger.error("Failed to import devpulse_ops plugin: %s", exc)
@@ -130,16 +212,6 @@ def _handle_system_pr(args: list[str]) -> dict:
"exit_code": 1,
}
try:
caller = verify_caller()
except PermissionError as exc:
logger.error("system-pr authorization failed: %s", exc)
return {
"stdout": "",
"stderr": str(exc),
"exit_code": 1,
}
result = create_system_pr(description, caller)
if result["success"]:
@@ -155,8 +227,8 @@ def _handle_system_pr(args: list[str]) -> dict:
}
def _handle_merge(args: list[str]) -> dict:
"""Handle the merge subcommand (devpulse-only)."""
def _handle_merge(args: list[str], caller: str) -> dict:
"""Handle the merge subcommand (owner-tier, auth pre-checked)."""
if not args:
return {
"stdout": "",
@@ -167,7 +239,6 @@ def _handle_merge(args: list[str]) -> dict:
pr_number = args[0]
try:
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
except ImportError as exc:
logger.error("Failed to import devpulse_ops merge plugin: %s", exc)
@@ -177,16 +248,6 @@ def _handle_merge(args: list[str]) -> dict:
"exit_code": 1,
}
try:
caller = verify_caller()
except PermissionError as exc:
logger.error("merge authorization failed: %s", exc)
return {
"stdout": "",
"stderr": str(exc),
"exit_code": 1,
}
result = merge_pr(pr_number, caller)
if result["success"]:
@@ -202,10 +263,9 @@ def _handle_merge(args: list[str]) -> dict:
}
def _handle_smart_sync(args: list[str]) -> dict:
"""Handle the smart-sync subcommand (devpulse-only)."""
def _handle_smart_sync(caller: str) -> dict:
"""Handle the smart-sync subcommand (owner-tier, auth pre-checked)."""
try:
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
from aipass.drone.apps.plugins.devpulse_ops.sync_plugin import smart_sync
except ImportError as exc:
logger.error("Failed to import devpulse_ops sync plugin: %s", exc)
@@ -215,16 +275,6 @@ def _handle_smart_sync(args: list[str]) -> dict:
"exit_code": 1,
}
try:
caller = verify_caller()
except PermissionError as exc:
logger.error("smart-sync authorization failed: %s", exc)
return {
"stdout": "",
"stderr": str(exc),
"exit_code": 1,
}
result = smart_sync(caller)
if result["success"]:
@@ -240,10 +290,9 @@ def _handle_smart_sync(args: list[str]) -> dict:
}
def _handle_fix(args: list[str]) -> dict:
"""Handle the fix subcommand (devpulse-only)."""
def _handle_fix(args: list[str], caller: str) -> dict:
"""Handle the fix subcommand (owner-tier, auth pre-checked)."""
try:
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
from aipass.drone.apps.plugins.devpulse_ops.fix_plugin import fix_git_state
except ImportError as exc:
logger.error("Failed to import devpulse_ops fix plugin: %s", exc)
@@ -253,16 +302,6 @@ def _handle_fix(args: list[str]) -> dict:
"exit_code": 1,
}
try:
caller = verify_caller()
except PermissionError as exc:
logger.error("fix authorization failed: %s", exc)
return {
"stdout": "",
"stderr": str(exc),
"exit_code": 1,
}
dry_run = "--dry-run" in (args or [])
result = fix_git_state(caller, dry_run=dry_run)
@@ -279,42 +318,8 @@ def _handle_fix(args: list[str]) -> dict:
}
def _handle_pr(args: list[str]) -> dict:
"""Handle the PR subcommand."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git pr <description>",
"exit_code": 1,
}
description = " ".join(args)
detected = _detect_branch_dir()
if detected is None:
return {
"stdout": "",
"stderr": "Cannot detect branch directory from CWD. Run from within src/aipass/<branch>/",
"exit_code": 1,
}
branch_name, branch_dir = detected
result = pr_handler.create_pr(branch_name, description, branch_dir)
if result["success"]:
return {
"stdout": f"PR created: {result['pr_url']}\nBranch: {result['feature_branch']}",
"stderr": "",
"exit_code": 0,
}
return {
"stdout": "",
"stderr": result["message"],
"exit_code": 1,
}
def _handle_status() -> dict:
"""Handle the status subcommand."""
"""Handle the status subcommand (global tier)."""
detected = _detect_branch_dir()
if detected is None:
return {
@@ -337,8 +342,96 @@ def _handle_status() -> dict:
}
def _handle_diff(args: list[str]) -> dict:
"""Handle the diff subcommand (global tier)."""
detected = _detect_branch_dir()
if detected is None:
return {
"stdout": "",
"stderr": "Cannot detect branch directory from CWD. Run from within src/aipass/<branch>/",
"exit_code": 1,
}
_, branch_dir = detected
staged = "--staged" in args
result = diff_handler.get_branch_diff(branch_dir, staged=staged)
return {
"stdout": result["diff"] if result["diff"] else result["message"],
"stderr": "",
"exit_code": 0,
}
def _handle_log(args: list[str]) -> dict:
"""Handle the log subcommand (global tier)."""
count = 10
for arg in args:
try:
count = int(arg)
break
except ValueError as exc:
logger.warning("Invalid log count argument '%s': %s", arg, exc)
continue
result = log_handler.get_git_log(count=count)
if result["entries"]:
return {
"stdout": "\n".join(result["entries"]),
"stderr": "",
"exit_code": 0,
}
return {
"stdout": result["message"],
"stderr": "",
"exit_code": 0,
}
def _handle_commit(args: list[str]) -> dict:
"""Handle the commit subcommand (owner tier)."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git commit <message> [--all]",
"exit_code": 1,
}
all_files = "--all" in args
msg_parts = [a for a in args if a != "--all"]
message = " ".join(msg_parts)
if not message:
return {
"stdout": "",
"stderr": "Commit message cannot be empty",
"exit_code": 1,
}
branch_dir = None
if all_files:
detected = _detect_branch_dir()
if detected:
_, branch_dir = detected
return commit_handler.commit_changes(message, branch_dir=branch_dir, all_files=all_files)
def _handle_checkout(args: list[str]) -> dict:
"""Handle the checkout subcommand (owner tier)."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git checkout <main|dev>",
"exit_code": 1,
}
return checkout_handler.checkout_branch(args[0])
def _handle_sync(args: list[str]) -> dict:
"""Handle the sync subcommand."""
"""Handle the sync subcommand (owner tier)."""
autostash = "--autostash" in args
result = sync_handler.sync_main(autostash=autostash)
@@ -356,7 +449,7 @@ def _handle_sync(args: list[str]) -> dict:
def _handle_lock() -> dict:
"""Handle the lock subcommand (check status)."""
"""Handle the lock subcommand (global tier)."""
result = lock_handler.check_lock_status()
return {
"stdout": json.dumps(result, indent=2),
@@ -366,7 +459,7 @@ def _handle_lock() -> dict:
def _handle_unlock(args: list[str]) -> dict:
"""Handle the unlock subcommand (force only)."""
"""Handle the unlock subcommand (owner tier)."""
if "--force" not in args:
return {
"stdout": "",
@@ -397,101 +490,126 @@ def get_help(command: str | None = None) -> str:
Returns:
Help text string.
"""
if command == "pr":
if command == "issue":
return (
"git pr <description> — Create a PR with scoped changes\n"
" Stages only files under the caller's branch directory,\n"
" creates a feature branch, commits, pushes, and opens a PR.\n"
"git issue [args] — Passthrough to gh issue CLI [global]\n Examples: list, create, view <#>, close <#>\n"
)
if command == "run":
return "git run [args] — Passthrough to gh run CLI [global]\n Examples: list, view <id>, watch <id>\n"
if command == "workflow":
return (
"git workflow [args] — Passthrough to gh workflow CLI [global]\n Examples: list, view <name>, run <name>\n"
)
if command == "pr":
return "git pr — DEPRECATED. Agent PRs are no longer supported. Devpulse handles git.\n"
if command == "status":
return "git status — Show git status filtered to your branch directory\n"
return "git status — Show git status filtered to your branch directory [global]\n"
if command == "diff":
return (
"git diff [--staged] — Show git diff filtered to your branch directory [global]\n"
" Options:\n"
" --staged Show staged changes only.\n"
)
if command == "log":
return "git log [count] — Show recent git log entries (default: 10) [global]\n"
if command == "lock":
return "git lock — Check current lock status [global]\n Shows lock holder, age, stale/orphan detection.\n"
if command == "commit":
return (
"git commit <message> [--all] — Commit staged changes [owner]\n"
" Options:\n"
" --all Stage all changes under your branch directory first.\n"
)
if command == "checkout":
return "git checkout <main|dev> — Switch branches (main or dev only) [owner]\n"
if command == "sync":
return (
"git sync [--autostash] — Checkout main and pull latest changes\n"
"git sync [--autostash] — Checkout main and pull latest changes [owner]\n"
" Options:\n"
" --autostash Stash local changes before pull and restore after.\n"
" Use when sync fails with 'unstaged changes' error.\n"
)
if command == "lock":
return "git lock — Check current lock status\n Shows lock holder, age, stale/orphan detection.\n"
if command == "unlock":
return "git unlock --force — Force-release the PR lock\n Removes .git_pr.lock regardless of holder.\n"
return "git unlock --force — Force-release the PR lock [owner]\n"
if command == "system-pr":
return (
"git system-pr <description> — Create a system-wide PR (devpulse only)\n"
" Stages all tracked changes, creates a disposable feature branch,\n"
" and opens a PR. Requires devpulse passport authorization.\n"
"git system-pr <description> — Create a system-wide PR [owner]\n"
" Stages all tracked changes, creates a feature branch, and opens a PR.\n"
)
if command == "merge":
return (
"git merge <PR#> — Merge a PR and sync local main (devpulse only)\n"
"git merge <PR#> — Merge a PR and sync local main [owner]\n"
" Runs gh pr merge --merge --delete-branch, then git pull --rebase.\n"
)
if command == "smart-sync":
return (
"git smart-sync — Fetch origin and rebase if behind (devpulse only)\n"
" Detects divergence and rebases safely; aborts on conflict.\n"
)
return "git smart-sync — Fetch origin and rebase if behind [owner]\n"
if command == "fix":
return (
"git fix [--dry-run] — Detect and fix common broken git states (devpulse only)\n"
"git fix [--dry-run] — Detect and fix broken git states [owner]\n"
"\n"
"Detected states and actions:\n"
" Stuck rebase .git/rebase-merge or rebase-apply exists\n"
" → git rebase --abort\n"
" Detached HEAD HEAD is not on a named branch\n"
" → git checkout main\n"
" Diverged local main and origin/main have diverged\n"
" → git fetch + git merge origin/main --no-edit\n"
" Dirty index files are staged but not committed\n"
" → git reset HEAD (unstages all)\n"
" Stuck rebase → git rebase --abort\n"
" Detached HEAD → git checkout main\n"
" Diverged → git fetch + git merge origin/main\n"
" Dirty index → git reset HEAD\n"
"\n"
"Options:\n"
" --dry-run Report detected states and proposed actions without\n"
" executing any fixes. Safe to run anytime.\n"
" --dry-run Report without executing fixes.\n"
)
return (
"git — Git workflow: PR, status, sync, lock management\n"
"git — Tier-based git workflow\n"
"\n"
"Commands:\n"
" pr <description> Create a PR with scoped changes\n"
" system-pr <desc> Create a system-wide PR (devpulse only)\n"
" merge <PR#> Merge a PR (devpulse only)\n"
" smart-sync Fetch + rebase if behind (devpulse only)\n"
" fix Fix broken git states (devpulse only)\n"
"Global (all branches):\n"
" status Show git status for your branch\n"
" sync Checkout main and pull\n"
" diff [--staged] Show git diff for your branch\n"
" log [count] Show recent git log (default: 10)\n"
" lock Check lock status\n"
" unlock --force Force-release the PR lock\n"
" issue [args] Passthrough to gh issue\n"
" run [args] Passthrough to gh run\n"
" workflow [args] Passthrough to gh workflow\n"
"\n"
"Policy: raw git commands are blocked for agents via .claude/settings.json:\n"
" - git checkout* (any form) — use `drone @git sync` instead\n"
" - git add -f* / --force* — use scoped `drone @git pr` instead\n"
"Culturally also avoid: raw git commit/push, gh pr create. Go through drone.\n"
"Owner (devpulse only):\n"
" commit <msg> [--all] Commit staged changes\n"
" checkout <main|dev> Switch branches\n"
" sync [--autostash] Checkout main and pull\n"
" unlock --force Force-release the PR lock\n"
" system-pr <desc> Create a system-wide PR\n"
" merge <PR#> Merge a PR\n"
" smart-sync Fetch + rebase if behind\n"
" fix [--dry-run] Fix broken git states\n"
"\n"
"Deprecated:\n"
" pr Agent PRs removed — devpulse handles git\n"
)
def get_introspective() -> str:
"""Return introspection text showing connected handlers."""
return (
"@git — Git workflow: PR, status, sync, lock management\n"
"@git — Tier-based git workflow (v2.0.0)\n"
"\n"
"Connected Handlers:\n"
" handlers/git/\n"
" - lock_handler.py (acquire_lock, release_lock, check_lock_status, force_unlock)\n"
" - status_handler.py (get_branch_status — scoped git status)\n"
" - diff_handler.py (get_branch_diff — scoped git diff)\n"
" - log_handler.py (get_git_log — recent log entries)\n"
" - commit_handler.py (commit_changes, stage_branch_dir)\n"
" - checkout_handler.py (checkout_branch — main/dev only)\n"
" - sync_handler.py (sync_main — safe main synchronization)\n"
" - pr_handler.py (create_pr — full PR workflow with lockfile)\n"
" - pr_handler.py (create_pr — DEPRECATED)\n"
"\n"
" plugins/devpulse_ops/\n"
" - auth.py (verify_caller — passport-based authorization)\n"
" - auth.py (verify_git_access — tier-based authorization)\n"
" - pr_plugin.py (create_system_pr — system-wide PR workflow)\n"
" - merge_plugin.py (merge_pr — merge PR + sync)\n"
" - sync_plugin.py (smart_sync — fetch + rebase if behind)\n"
" - fix_plugin.py (fix_git_state — detect/fix broken states)\n"
"\n"
"Commands: pr, system-pr, merge, smart-sync, fix, status, sync, lock, unlock\n"
" gh passthrough:\n"
" - issue, run, workflow → subprocess gh <cmd> [args]\n"
"\n"
"Access Tiers: global (status, diff, log, lock, issue, run, workflow) | owner (commit, checkout, sync, unlock, system-pr, merge, smart-sync, fix)\n"
)
@@ -24,6 +24,27 @@ from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
GIT_ACCESS_TIERS: dict[str, dict] = {
"global": {
"commands": ["status", "diff", "log", "lock", "issue", "run", "workflow"],
"description": "Read-only — available to all branches",
},
"owner": {
"commands": [
"commit",
"checkout",
"sync",
"unlock",
"system-pr",
"merge",
"smart-sync",
"fix",
],
"allowed_callers": ["devpulse"],
"description": "Write operations — project owner only",
},
}
def _find_caller() -> str:
"""Walk up from CWD to find passport.json and return branch name.
@@ -83,3 +104,46 @@ def verify_caller() -> str:
)
logger.info("Caller '%s' authorized for devpulse operations", name)
return name
def verify_git_access(command: str) -> str:
"""Check if the calling branch is authorized for this git command.
Uses GIT_ACCESS_TIERS to determine access level. Global-tier commands
are available to all branches; owner-tier commands require the caller
to be in the allowed_callers list.
Returns:
The caller's branch name if authorized.
Raises:
PermissionError: If the caller is not authorized for this command.
"""
if command == "pr":
raise PermissionError("Agent PRs are deprecated. Build code, run tests, report results. Devpulse handles git.")
global_cmds = GIT_ACCESS_TIERS["global"]["commands"]
owner_tier = GIT_ACCESS_TIERS["owner"]
if command in global_cmds:
caller = _find_caller()
json_handler.log_operation(
"git_access_verify",
{"caller": caller, "command": command, "tier": "global"},
)
return caller
if command in owner_tier["commands"]:
caller = _find_caller()
allowed = owner_tier["allowed_callers"]
if caller not in allowed:
msg = f"Branch '{caller}' is not authorized for '{command}'. Only {allowed} can use owner-tier commands."
logger.error(msg)
raise PermissionError(msg)
json_handler.log_operation(
"git_access_verify",
{"caller": caller, "command": command, "tier": "owner"},
)
return caller
raise PermissionError(f"Unknown git command: '{command}'")
+20 -10
View File
@@ -518,26 +518,32 @@ class TestGitModuleRouting:
assert "sync_plugin" in intro
assert "fix_plugin" in intro
def test_handle_merge_no_args(self) -> None:
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
return_value="devpulse",
)
def test_handle_merge_no_args(self, _mock_access: MagicMock) -> None:
from aipass.drone.apps.modules.git_module import handle_command
result = handle_command("merge", [])
assert result["exit_code"] == 1
assert "Usage" in result["stderr"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
return_value="devpulse",
)
@patch("aipass.drone.apps.plugins.devpulse_ops.merge_plugin.find_repo_root")
@patch("aipass.drone.apps.plugins.devpulse_ops.merge_plugin.subprocess.run")
def test_handle_merge_routes_correctly(
self,
mock_run: MagicMock,
mock_root: MagicMock,
mock_verify: MagicMock,
_mock_access: MagicMock,
tmp_path: Path,
) -> None:
from aipass.drone.apps.modules.git_module import handle_command
mock_verify.return_value = "devpulse"
mock_root.return_value = tmp_path
proc = MagicMock()
@@ -549,19 +555,21 @@ class TestGitModuleRouting:
result = handle_command("merge", ["42"])
assert result["exit_code"] == 0
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
return_value="devpulse",
)
@patch("aipass.drone.apps.plugins.devpulse_ops.sync_plugin.find_repo_root")
@patch("aipass.drone.apps.plugins.devpulse_ops.sync_plugin.subprocess.run")
def test_handle_smart_sync_routes_correctly(
self,
mock_run: MagicMock,
mock_root: MagicMock,
mock_verify: MagicMock,
_mock_access: MagicMock,
tmp_path: Path,
) -> None:
from aipass.drone.apps.modules.git_module import handle_command
mock_verify.return_value = "devpulse"
mock_root.return_value = tmp_path
def side_effect(cmd: list[str], **kwargs: object) -> MagicMock:
@@ -579,19 +587,21 @@ class TestGitModuleRouting:
result = handle_command("smart-sync", [])
assert result["exit_code"] == 0
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
return_value="devpulse",
)
@patch("aipass.drone.apps.plugins.devpulse_ops.fix_plugin.find_repo_root")
@patch("aipass.drone.apps.plugins.devpulse_ops.fix_plugin.subprocess.run")
def test_handle_fix_routes_correctly(
self,
mock_run: MagicMock,
mock_root: MagicMock,
mock_verify: MagicMock,
_mock_access: MagicMock,
tmp_path: Path,
) -> None:
from aipass.drone.apps.modules.git_module import handle_command
mock_verify.return_value = "devpulse"
mock_root.return_value = tmp_path
git_dir = tmp_path / ".git"
git_dir.mkdir()
+754
View File
@@ -0,0 +1,754 @@
# =================== AIPass ====================
# Name: test_git_access.py
# Description: Tests for tier-based git access, new handlers, and PR deprecation
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Tests for tier-based git access, new handlers (diff, log, commit, checkout), and PR deprecation."""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from aipass.drone.apps.plugins.devpulse_ops.auth import (
GIT_ACCESS_TIERS,
verify_git_access,
)
from aipass.drone.apps.handlers.git.diff_handler import get_branch_diff
from aipass.drone.apps.handlers.git.log_handler import get_git_log
from aipass.drone.apps.handlers.git.commit_handler import commit_changes, stage_branch_dir
from aipass.drone.apps.handlers.git.checkout_handler import checkout_branch
from aipass.drone.apps.modules.git_module import handle_command
# ===========================================================================
# Fixtures
# ===========================================================================
@pytest.fixture()
def devpulse_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Create a temp directory with a devpulse passport."""
trinity = tmp_path / ".trinity"
trinity.mkdir()
passport = trinity / "passport.json"
passport.write_text(
json.dumps({"branch_info": {"branch_name": "devpulse"}}),
encoding="utf-8",
)
monkeypatch.chdir(tmp_path)
return tmp_path
@pytest.fixture()
def seedgo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Create a temp directory with a non-owner passport."""
trinity = tmp_path / ".trinity"
trinity.mkdir()
passport = trinity / "passport.json"
passport.write_text(
json.dumps({"branch_info": {"branch_name": "seedgo"}}),
encoding="utf-8",
)
monkeypatch.chdir(tmp_path)
return tmp_path
@pytest.fixture()
def repo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Create a temp repo root with AIPASS_REGISTRY.json."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
return tmp_path
# ===========================================================================
# 1. GIT_ACCESS_TIERS config structure
# ===========================================================================
class TestGitAccessTiers:
"""Verify the tier config is correctly structured."""
def test_tiers_has_global_and_owner(self) -> None:
assert "global" in GIT_ACCESS_TIERS
assert "owner" in GIT_ACCESS_TIERS
def test_global_commands(self) -> None:
cmds = GIT_ACCESS_TIERS["global"]["commands"]
assert "status" in cmds
assert "diff" in cmds
assert "log" in cmds
assert "lock" in cmds
def test_owner_commands(self) -> None:
cmds = GIT_ACCESS_TIERS["owner"]["commands"]
assert "commit" in cmds
assert "checkout" in cmds
assert "sync" in cmds
assert "unlock" in cmds
assert "system-pr" in cmds
assert "merge" in cmds
assert "smart-sync" in cmds
assert "fix" in cmds
def test_owner_allowed_callers(self) -> None:
allowed = GIT_ACCESS_TIERS["owner"]["allowed_callers"]
assert allowed == ["devpulse"]
def test_pr_not_in_any_tier(self) -> None:
all_cmds = GIT_ACCESS_TIERS["global"]["commands"] + GIT_ACCESS_TIERS["owner"]["commands"]
assert "pr" not in all_cmds
# ===========================================================================
# 2. verify_git_access — tier enforcement
# ===========================================================================
class TestVerifyGitAccessGlobal:
"""Global-tier commands should pass for any caller."""
def test_status_allowed_for_any_branch(self, devpulse_dir: Path) -> None:
assert verify_git_access("status") == "devpulse"
def test_diff_allowed_for_seedgo(self, seedgo_dir: Path) -> None:
assert verify_git_access("diff") == "seedgo"
def test_log_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
assert verify_git_access("log") == "seedgo"
def test_lock_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
assert verify_git_access("lock") == "seedgo"
class TestVerifyGitAccessOwner:
"""Owner-tier commands should only pass for devpulse."""
def test_commit_allowed_for_devpulse(self, devpulse_dir: Path) -> None:
assert verify_git_access("commit") == "devpulse"
def test_commit_denied_for_seedgo(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="not authorized"):
verify_git_access("commit")
def test_checkout_denied_for_seedgo(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="not authorized"):
verify_git_access("checkout")
def test_sync_denied_for_seedgo(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="not authorized"):
verify_git_access("sync")
def test_unlock_denied_for_seedgo(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="not authorized"):
verify_git_access("unlock")
def test_system_pr_denied_for_seedgo(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="not authorized"):
verify_git_access("system-pr")
class TestVerifyGitAccessPrDeprecated:
"""PR command should be denied with deprecation message."""
def test_pr_deprecated_for_devpulse(self, devpulse_dir: Path) -> None:
with pytest.raises(PermissionError, match="deprecated"):
verify_git_access("pr")
def test_pr_deprecated_for_any_branch(self, seedgo_dir: Path) -> None:
with pytest.raises(PermissionError, match="deprecated"):
verify_git_access("pr")
class TestVerifyGitAccessUnknown:
"""Unknown commands should be denied."""
def test_unknown_command_denied(self, devpulse_dir: Path) -> None:
with pytest.raises(PermissionError, match="Unknown git command"):
verify_git_access("nonexistent")
# ===========================================================================
# 3. diff_handler
# ===========================================================================
class TestDiffHandler:
"""Scoped git diff tests."""
def test_basic_diff(self, repo_dir: Path) -> None:
diff_output = (
"diff --git a/src/aipass/api/foo.py b/src/aipass/api/foo.py\n"
"--- a/src/aipass/api/foo.py\n"
"+++ b/src/aipass/api/foo.py\n"
"@@ -1,3 +1,4 @@\n"
"+new line\n"
"diff --git a/src/aipass/drone/bar.py b/src/aipass/drone/bar.py\n"
"--- a/src/aipass/drone/bar.py\n"
"+++ b/src/aipass/drone/bar.py\n"
)
mock_result = MagicMock(returncode=0, stdout=diff_output, stderr="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
result = get_branch_diff(branch_dir)
assert result["files_changed"] == 1
assert "src/aipass/api/foo.py" in result["diff"]
assert "src/aipass/drone/bar.py" not in result["diff"]
def test_staged_diff(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=0, stdout="", stderr="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result) as mock_run:
get_branch_diff(branch_dir, staged=True)
cmd = mock_run.call_args[0][0]
assert "--staged" in cmd
def test_empty_diff(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=0, stdout="", stderr="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
result = get_branch_diff(branch_dir)
assert result["files_changed"] == 0
assert result["diff"] == ""
def test_git_failure(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=128, stderr="fatal: not a git repo", stdout="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
result = get_branch_diff(branch_dir)
assert result["files_changed"] == 0
assert "error" in result["message"].lower()
def test_os_error(self, repo_dir: Path) -> None:
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", side_effect=OSError("git not found")):
result = get_branch_diff(branch_dir)
assert result["files_changed"] == 0
assert "failed" in result["message"].lower()
# ===========================================================================
# 4. log_handler
# ===========================================================================
class TestLogHandler:
"""Git log tests."""
def test_basic_log(self, repo_dir: Path) -> None:
log_output = "abc1234 feat: first\ndef5678 fix: second\n"
mock_result = MagicMock(returncode=0, stdout=log_output, stderr="")
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
result = get_git_log(count=5)
assert result["count"] == 2
assert len(result["entries"]) == 2
def test_custom_count_passed_to_git(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=0, stdout="", stderr="")
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result) as mock_run:
get_git_log(count=25)
cmd = mock_run.call_args[0][0]
assert "-25" in cmd
def test_git_failure(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=128, stderr="fatal: bad default", stdout="")
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
result = get_git_log()
assert result["count"] == 0
assert "error" in result["message"].lower()
def test_os_error(self, repo_dir: Path) -> None:
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", side_effect=OSError("git not found")):
result = get_git_log()
assert result["count"] == 0
assert "failed" in result["message"].lower()
# ===========================================================================
# 5. commit_handler
# ===========================================================================
class TestStageBranchDir:
"""Shared staging utility tests."""
def test_stage_success(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=0, stderr="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_result):
result = stage_branch_dir(branch_dir, repo_dir)
assert result["success"] is True
def test_stage_failure(self, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=1, stderr="fatal: pathspec error")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_result):
result = stage_branch_dir(branch_dir, repo_dir)
assert result["success"] is False
assert "failed" in result["message"].lower()
class TestCommitChanges:
"""Commit handler tests."""
def test_commit_staged(self, repo_dir: Path) -> None:
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
mock_commit = MagicMock(returncode=0, stdout="[main abc123] test commit", stderr="")
with patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[mock_diff, mock_commit],
):
result = commit_changes("test commit")
assert result["exit_code"] == 0
assert "abc123" in result["stdout"]
def test_commit_nothing_staged(self, repo_dir: Path) -> None:
mock_diff = MagicMock(returncode=0, stdout="", stderr="")
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_diff):
result = commit_changes("test commit")
assert result["exit_code"] == 1
assert "nothing to commit" in result["stderr"].lower()
def test_commit_all_stages_first(self, repo_dir: Path) -> None:
mock_add = MagicMock(returncode=0, stderr="")
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
mock_commit = MagicMock(returncode=0, stdout="[main def456] all commit", stderr="")
branch_dir = repo_dir / "src" / "aipass" / "api"
with patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[mock_add, mock_diff, mock_commit],
):
result = commit_changes("all commit", branch_dir=branch_dir, all_files=True)
assert result["exit_code"] == 0
def test_commit_os_error(self, repo_dir: Path) -> None:
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
with patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[mock_diff, OSError("git not found")],
):
result = commit_changes("test commit")
assert result["exit_code"] == 1
assert "failed" in result["stderr"].lower()
# ===========================================================================
# 6. checkout_handler
# ===========================================================================
class TestCheckoutHandler:
"""Branch checkout with hard guard tests."""
def test_checkout_main_allowed(self, repo_dir: Path) -> None:
mock_status = MagicMock(returncode=0, stdout="", stderr="")
mock_checkout = MagicMock(returncode=0, stdout="", stderr="Switched to branch 'main'")
with patch(
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
side_effect=[mock_status, mock_checkout],
):
result = checkout_branch("main")
assert result["exit_code"] == 0
assert result["current_branch"] == "main"
def test_checkout_dev_allowed(self, repo_dir: Path) -> None:
mock_status = MagicMock(returncode=0, stdout="", stderr="")
mock_checkout = MagicMock(returncode=0, stdout="", stderr="Switched to branch 'dev'")
with patch(
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
side_effect=[mock_status, mock_checkout],
):
result = checkout_branch("dev")
assert result["exit_code"] == 0
assert result["current_branch"] == "dev"
def test_checkout_feature_branch_denied(self) -> None:
result = checkout_branch("feat/my-feature")
assert result["exit_code"] == 1
assert "denied" in result["stderr"].lower()
assert result["current_branch"] == ""
def test_checkout_arbitrary_branch_denied(self) -> None:
result = checkout_branch("release/v2")
assert result["exit_code"] == 1
assert "denied" in result["stderr"].lower()
def test_checkout_dirty_tree_aborts(self, repo_dir: Path) -> None:
mock_status = MagicMock(returncode=0, stdout=" M some/file.py\n", stderr="")
with patch("aipass.drone.apps.handlers.git.checkout_handler.subprocess.run", return_value=mock_status):
result = checkout_branch("main")
assert result["exit_code"] == 1
assert "uncommitted" in result["stderr"].lower()
def test_checkout_git_failure(self, repo_dir: Path) -> None:
mock_status = MagicMock(returncode=0, stdout="", stderr="")
mock_checkout = MagicMock(returncode=1, stdout="", stderr="error: pathspec 'main' did not match")
with patch(
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
side_effect=[mock_status, mock_checkout],
):
result = checkout_branch("main")
assert result["exit_code"] == 1
assert result["current_branch"] == ""
# ===========================================================================
# 7. PR deprecation through handle_command
# ===========================================================================
class TestPrDeprecation:
"""PR command returns deprecation message via centralized auth."""
def test_pr_returns_deprecation(self, devpulse_dir: Path) -> None:
result = handle_command("pr", ["some description"])
assert result["exit_code"] == 1
assert "deprecated" in result["stderr"].lower()
def test_pr_no_args_also_deprecated(self, devpulse_dir: Path) -> None:
result = handle_command("pr")
assert result["exit_code"] == 1
assert "deprecated" in result["stderr"].lower()
# ===========================================================================
# 8. New commands via handle_command routing
# ===========================================================================
class TestNewCommandRouting:
"""Verify new commands route through handle_command correctly."""
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
def test_diff_routes(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
trinity = repo_dir / ".trinity"
trinity.mkdir()
passport = trinity / "passport.json"
passport.write_text(json.dumps({"branch_info": {"branch_name": "test_branch"}}))
mock_result = MagicMock(returncode=0, stdout="", stderr="")
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
result = handle_command("diff")
assert result["exit_code"] == 0
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
def test_log_routes(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
mock_result = MagicMock(returncode=0, stdout="abc123 test\n", stderr="")
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
result = handle_command("log")
assert result["exit_code"] == 0
assert "abc123" in result["stdout"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_commit_no_args_error(self, _mock_auth: MagicMock) -> None:
result = handle_command("commit")
assert result["exit_code"] == 1
assert "usage" in result["stderr"].lower()
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_checkout_no_args_error(self, _mock_auth: MagicMock) -> None:
result = handle_command("checkout")
assert result["exit_code"] == 1
assert "usage" in result["stderr"].lower()
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_checkout_routes_to_handler(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
mock_status = MagicMock(returncode=0, stdout="", stderr="")
mock_checkout = MagicMock(returncode=0, stdout="", stderr="")
with patch(
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
side_effect=[mock_status, mock_checkout],
):
result = handle_command("checkout", ["main"])
assert result["exit_code"] == 0
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_checkout_guard_rejects_feature(self, _mock_auth: MagicMock) -> None:
result = handle_command("checkout", ["feat/bad"])
assert result["exit_code"] == 1
assert "denied" in result["stderr"].lower()
# ===========================================================================
# 9. Help text includes new commands and tiers
# ===========================================================================
class TestUpdatedHelp:
"""Help and introspection reflect new commands and tiers."""
def test_help_includes_diff(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "diff" in text
def test_help_includes_log(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "log" in text
def test_help_includes_commit(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "commit" in text
def test_help_includes_checkout(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "checkout" in text
def test_help_shows_tier_sections(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "global" in text.lower()
assert "owner" in text.lower()
def test_help_marks_pr_deprecated(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help()
assert "deprecated" in text.lower()
def test_introspection_includes_new_handlers(self) -> None:
from aipass.drone.apps.modules.git_module import get_introspective
text = get_introspective()
assert "diff_handler" in text
assert "log_handler" in text
assert "commit_handler" in text
assert "checkout_handler" in text
def test_introspection_shows_tiers(self) -> None:
from aipass.drone.apps.modules.git_module import get_introspective
text = get_introspective()
assert "global" in text.lower()
assert "owner" in text.lower()
# ===========================================================================
# 10. gh passthrough commands (issue, run, workflow)
# ===========================================================================
class TestGhPassthroughTierConfig:
"""Passthrough commands are in the global tier."""
def test_issue_in_global_tier(self) -> None:
assert "issue" in GIT_ACCESS_TIERS["global"]["commands"]
def test_run_in_global_tier(self) -> None:
assert "run" in GIT_ACCESS_TIERS["global"]["commands"]
def test_workflow_in_global_tier(self) -> None:
assert "workflow" in GIT_ACCESS_TIERS["global"]["commands"]
def test_passthrough_not_in_owner_tier(self) -> None:
owner_cmds = GIT_ACCESS_TIERS["owner"]["commands"]
assert "issue" not in owner_cmds
assert "run" not in owner_cmds
assert "workflow" not in owner_cmds
class TestGhPassthroughAccess:
"""Global-tier access for passthrough commands."""
def test_issue_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
assert verify_git_access("issue") == "seedgo"
def test_run_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
assert verify_git_access("run") == "seedgo"
def test_workflow_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
assert verify_git_access("workflow") == "seedgo"
class TestGhPassthroughRouting:
"""handle_command routes passthrough to subprocess."""
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_issue_list(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="Issue #1\nIssue #2\n", stderr="")
result = handle_command("issue", ["list"])
assert result["exit_code"] == 0
assert "Issue #1" in result["stdout"]
mock_run.assert_called_once_with(
["gh", "issue", "list"],
capture_output=True,
text=True,
timeout=60,
)
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_run_list(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="run 123\n", stderr="")
result = handle_command("run", ["list"])
assert result["exit_code"] == 0
mock_run.assert_called_once_with(
["gh", "run", "list"],
capture_output=True,
text=True,
timeout=60,
)
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_workflow_list(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="CI workflow\n", stderr="")
result = handle_command("workflow", ["list"])
assert result["exit_code"] == 0
mock_run.assert_called_once_with(
["gh", "workflow", "list"],
capture_output=True,
text=True,
timeout=60,
)
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_passthrough_no_args(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="usage info\n", stderr="")
result = handle_command("issue")
assert result["exit_code"] == 0
mock_run.assert_called_once_with(
["gh", "issue"],
capture_output=True,
text=True,
timeout=60,
)
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_passthrough_returns_stderr(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="not authenticated")
result = handle_command("issue", ["list"])
assert result["exit_code"] == 1
assert "not authenticated" in result["stderr"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run", side_effect=FileNotFoundError("gh"))
def test_passthrough_gh_not_found(self, _mock_run: MagicMock, _mock_auth: MagicMock) -> None:
result = handle_command("issue", ["list"])
assert result["exit_code"] == 1
assert "gh CLI not found" in result["stderr"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch(
"aipass.drone.apps.modules.git_module.subprocess.run",
side_effect=__import__("subprocess").TimeoutExpired(["gh", "issue"], 60),
)
def test_passthrough_timeout(self, _mock_run: MagicMock, _mock_auth: MagicMock) -> None:
result = handle_command("issue", ["list"])
assert result["exit_code"] == 1
assert "timed out" in result["stderr"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
@patch("aipass.drone.apps.modules.git_module.subprocess.run")
def test_passthrough_multiple_args(self, mock_run: MagicMock, _mock_auth: MagicMock) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
handle_command("issue", ["create", "--title", "Bug", "--body", "Details"])
mock_run.assert_called_once_with(
["gh", "issue", "create", "--title", "Bug", "--body", "Details"],
capture_output=True,
text=True,
timeout=60,
)
class TestGhPassthroughHelp:
"""Help text includes passthrough commands."""
def test_help_includes_issue(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
assert "issue" in get_help()
def test_help_includes_run(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
assert "run" in get_help()
def test_help_includes_workflow(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
assert "workflow" in get_help()
def test_per_command_help_issue(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help("issue")
assert "gh issue" in text
assert "global" in text.lower()
def test_per_command_help_run(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help("run")
assert "gh run" in text
def test_per_command_help_workflow(self) -> None:
from aipass.drone.apps.modules.git_module import get_help
text = get_help("workflow")
assert "gh workflow" in text
def test_introspection_includes_passthrough(self) -> None:
from aipass.drone.apps.modules.git_module import get_introspective
text = get_introspective()
assert "issue" in text
assert "run" in text
assert "workflow" in text
+23 -13
View File
@@ -614,9 +614,14 @@ class TestGitModuleRouting:
result = handle_command("bogus")
assert result["exit_code"] == 1
assert "unknown" in result["stderr"].lower()
assert "pr" in result["stderr"]
def test_lock_routes_to_handler(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
return_value="test_branch",
)
def test_lock_routes_to_handler(
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""lock command routes to check_lock_status."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
@@ -627,13 +632,15 @@ class TestGitModuleRouting:
data = json.loads(result["stdout"])
assert data["locked"] is False
def test_unlock_requires_force(self) -> None:
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_unlock_requires_force(self, _mock_auth: MagicMock) -> None:
"""unlock without --force returns error."""
result = handle_command("unlock")
assert result["exit_code"] == 1
assert "--force" in result["stderr"]
def test_unlock_with_force(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_unlock_with_force(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""unlock --force routes to force_unlock."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
@@ -642,7 +649,10 @@ class TestGitModuleRouting:
result = handle_command("unlock", ["--force"])
assert result["exit_code"] == 0
def test_sync_routes_to_handler(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
def test_sync_routes_to_handler(
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""sync command routes to sync_main."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
@@ -661,7 +671,8 @@ class TestGitModuleRouting:
assert result["exit_code"] == 0
def test_status_no_branch_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
def test_status_no_branch_dir(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""status outside a branch directory returns error."""
monkeypatch.chdir(tmp_path)
result = handle_command("status")
@@ -669,17 +680,16 @@ class TestGitModuleRouting:
assert "cannot detect" in result["stderr"].lower()
def test_pr_no_args(self) -> None:
"""pr with no arguments returns usage error."""
"""pr command is deprecated."""
result = handle_command("pr")
assert result["exit_code"] == 1
assert "usage" in result["stderr"].lower()
assert "deprecated" in result["stderr"].lower()
def test_pr_no_branch_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""pr outside a branch directory returns error."""
monkeypatch.chdir(tmp_path)
def test_pr_no_branch_dir(self) -> None:
"""pr command is deprecated regardless of context."""
result = handle_command("pr", ["some description"])
assert result["exit_code"] == 1
assert "cannot detect" in result["stderr"].lower()
assert "deprecated" in result["stderr"].lower()
class TestDetectBranchDir:
@@ -756,7 +766,7 @@ class TestGitModuleHelp:
"""Command-specific help returns relevant text."""
text = get_help("pr")
assert "pr" in text.lower()
assert "description" in text.lower()
assert "deprecated" in text.lower()
def test_introspective(self) -> None:
"""Introspection lists connected handlers."""
+7 -5
View File
@@ -319,11 +319,11 @@ class TestGitModuleSystemPrRouting:
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."""
"""get_help('system-pr') output mentions owner as the tier label."""
from aipass.drone.apps.modules.git_module import get_help
help_text = get_help("system-pr")
assert "devpulse" in help_text
assert "owner" in help_text.lower()
def test_get_introspective_includes_plugin(self) -> None:
"""get_introspective() output mentions the devpulse_ops plugin."""
@@ -332,7 +332,7 @@ class TestGitModuleSystemPrRouting:
intro = get_introspective()
assert "devpulse_ops" in intro
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
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
@@ -341,12 +341,14 @@ class TestGitModuleSystemPrRouting:
assert result["exit_code"] == 1
assert "Usage" in result["stderr"]
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
@patch(
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
side_effect=PermissionError("not authorized"),
)
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")
result = handle_command("system-pr", ["test"])
assert result["exit_code"] == 1
assert "not authorized" in result["stderr"]