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

feat(system): DPLAN-0173: Git workflow redesign — tier-based access, new handlers, hook reverts

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-05-12 18:50:22 -07:00
co-authored by @devpulse
parent 3b8fa1fa5a
commit d19a840253
12 changed files with 338 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/hooks/README.md
+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
src/aipass/devpulse/.claude/hooks/auto_fix_diagnostics.py
+1 -25
View File
@@ -299,30 +299,9 @@ def is_same_file_as_last(file_path: str) -> bool:
return False
def _project_has_own_posttooluse_hooks() -> bool:
"""Check if CWD is inside a project with its own PostToolUse hooks."""
search = Path.cwd()
home = Path.home()
while search != home and search.parent != search:
settings = search / ".claude" / "settings.json"
if settings.exists():
try:
data = json.loads(settings.read_text(encoding="utf-8"))
ptu = data.get("hooks", {}).get("PostToolUse", [])
if ptu:
return True
except (json.JSONDecodeError, OSError):
pass
search = search.parent
return False
def main():
"""Main hook entry point."""
try:
if _project_has_own_posttooluse_hooks():
return
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
@@ -384,7 +363,4 @@ Fix these errors in {Path(file_path).name} now. Do not skip or defer."""
if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("PostToolUse", "provider", __file__, main)
main()
src/aipass/devpulse/.claude/hooks/branch_prompt_loader.py
+2 -33
View File
@@ -6,35 +6,12 @@ Injects branch-specific prompts based on CWD. When working in a branch
directory, loads .aipass/aipass_local_prompt.md and outputs it so the
AI sees branch-specific context.
When CWD is inside a project that has its own UserPromptSubmit hooks
(e.g. a standalone aipass-init project), this provider-level hook exits
silently to avoid double-firing.
Version: 1.1.0
Version: 1.0.0
"""
import json
from pathlib import Path
def _project_has_own_hooks() -> bool:
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
search = Path.cwd()
home = Path.home()
while search != home and search.parent != search:
settings = search / ".claude" / "settings.json"
if settings.exists():
try:
data = json.loads(settings.read_text(encoding="utf-8"))
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
if ups:
return True
except (json.JSONDecodeError, OSError):
pass
search = search.parent
return False
def find_branch_root() -> Path | None:
"""
Find the branch root directory.
@@ -62,9 +39,6 @@ def find_branch_root() -> Path | None:
def main():
if _project_has_own_hooks():
return
branch_root = find_branch_root()
if branch_root:
@@ -76,9 +50,4 @@ def main():
if __name__ == "__main__":
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("UserPromptSubmit", "provider", __file__, main)
main()
src/aipass/devpulse/.claude/hooks/email_notification.py
+2 -31
View File
@@ -5,34 +5,13 @@ Email Notification Hook - Notifies of new emails on prompt submit.
Checks the current branch's inbox for unread emails and displays
a notification if any exist.
When CWD is inside a project that has its own UserPromptSubmit hooks,
this provider-level hook exits silently to avoid double-firing.
Version: 1.1.0
Version: 1.0.0
"""
import json
from pathlib import Path
def _project_has_own_hooks() -> bool:
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
search = Path.cwd()
home = Path.home()
while search != home and search.parent != search:
settings = search / ".claude" / "settings.json"
if settings.exists():
try:
data = json.loads(settings.read_text(encoding="utf-8"))
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
if ups:
return True
except (json.JSONDecodeError, OSError):
pass
search = search.parent
return False
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
@@ -101,9 +80,6 @@ def count_new_emails(branch_root: Path) -> int:
def main():
if _project_has_own_hooks():
return
branch_root = find_branch_root()
if not branch_root:
return
@@ -117,9 +93,4 @@ def main():
if __name__ == "__main__":
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("UserPromptSubmit", "provider", __file__, main)
main()
src/aipass/devpulse/.claude/hooks/identity_injector.py
+2 -31
View File
@@ -5,34 +5,13 @@ Identity Injector - Injects branch identity on every prompt.
Reads from [BRANCH].id.json and outputs core identity fields.
Finds the branch root by walking up from CWD looking for apps/ or *.id.json.
When CWD is inside a project that has its own UserPromptSubmit hooks,
this provider-level hook exits silently to avoid double-firing.
Version: 1.1.0
Version: 1.0.0
"""
import json
from pathlib import Path
def _project_has_own_hooks() -> bool:
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
search = Path.cwd()
home = Path.home()
while search != home and search.parent != search:
settings = search / ".claude" / "settings.json"
if settings.exists():
try:
data = json.loads(settings.read_text(encoding="utf-8"))
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
if ups:
return True
except (json.JSONDecodeError, OSError):
pass
search = search.parent
return False
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
@@ -118,9 +97,6 @@ def format_identity(data: dict) -> str:
def main():
if _project_has_own_hooks():
return
branch_root = find_branch_root()
if not branch_root:
return
@@ -139,9 +115,4 @@ def main():
if __name__ == "__main__":
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("UserPromptSubmit", "provider", __file__, main)
main()
src/aipass/devpulse/.claude/hooks/pre_compact.py
+2 -5
View File
@@ -85,7 +85,7 @@ def _get_git_info():
text=True,
timeout=5,
)
status = subprocess.run(
subprocess.run(
["git", "diff", "--stat", "--cached", "HEAD"],
capture_output=True,
text=True,
@@ -165,7 +165,4 @@ Context just compacted. Below is your live state. Use it to continue seamlessly.
if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("PreCompact", "provider", __file__, main)
main()
src/aipass/devpulse/.claude/hooks/subagent_stop_gate.py
+13 -68
View File
@@ -29,41 +29,18 @@ def _find_repo_root() -> Path | None:
AIPASS_ROOT = _find_repo_root()
def _get_cwd_branch() -> str | None:
"""Detect which branch directory (src/aipass/<name>) the CWD is in."""
cwd = Path.cwd().resolve()
if AIPASS_ROOT is None:
return None
src = AIPASS_ROOT / "src" / "aipass"
try:
rel = cwd.relative_to(src)
return rel.parts[0] if rel.parts else None
except ValueError:
return None
def get_modified_py_files() -> list[str]:
"""Get Python files modified in the working tree, scoped to the CWD branch.
Only returns files inside the current branch's directory (or repo-root files).
This prevents dispatched agents' changes from triggering violations on the
orchestrator or other agents sharing the worktree.
"""
"""Get Python files modified in the working tree (unstaged + staged)."""
if AIPASS_ROOT is None:
return []
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
)
cwd_branch = _get_cwd_branch()
files = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if line.endswith(".py") and not line.startswith(".claude/"):
if cwd_branch and line.startswith("src/aipass/"):
file_branch = line.split("/")[2] if len(line.split("/")) > 2 else None
if file_branch and file_branch != cwd_branch:
continue
full = AIPASS_ROOT / line
if full.exists():
files.append(str(full))
@@ -100,29 +77,6 @@ def run_seedgo_checklist(file_path: str) -> list[str]:
return []
def check_hook_readme_accountability() -> str | None:
"""Check if hook files changed but README wasn't updated. Returns reminder or None."""
if AIPASS_ROOT is None:
return None
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
)
changed = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
hook_files_changed = any(f.startswith(".claude/hooks/") and f.endswith(".py") for f in changed)
readme_changed = ".claude/hooks/README.md" in changed
if hook_files_changed and not readme_changed:
return (
"Hook files were modified but .claude/hooks/README.md was not updated. "
"Consider updating the README to reflect your changes."
)
except Exception:
pass
return None
def main():
try:
json.load(sys.stdin)
@@ -131,8 +85,6 @@ def main():
if not modified:
return # Nothing to check
readme_reminder = check_hook_readme_accountability()
all_violations = {}
for f in modified:
vs = run_seedgo_checklist(f)
@@ -140,30 +92,23 @@ def main():
name = Path(f).name
all_violations[name] = vs
if all_violations:
# Build the block reason
lines = ["Standards violations found in files you modified:\n"]
for fname, vs in all_violations.items():
lines.append(f" {fname}:")
for v in vs:
lines.append(f" - {v}")
lines.append("\nFix these violations before finishing.")
if not all_violations:
return # All clear
if readme_reminder:
lines.append(f"\n⚠️ {readme_reminder}")
# Build the block reason
lines = ["Standards violations found in files you modified:\n"]
for fname, vs in all_violations.items():
lines.append(f" {fname}:")
for v in vs:
lines.append(f" - {v}")
lines.append("\nFix these violations before finishing.")
output = {"decision": "block", "reason": "\n".join(lines)}
print(json.dumps(output))
elif readme_reminder:
output = {"decision": "allow", "reason": f"⚠️ {readme_reminder}"}
print(json.dumps(output))
output = {"decision": "block", "reason": "\n".join(lines)}
print(json.dumps(output))
except Exception:
pass # Silent fail — don't block on errors
if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).resolve().parent))
from hook_log import run_and_log
run_and_log("SubagentStop", "provider", __file__, main)
main()
src/aipass/drone/README.md
+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)
src/aipass/drone/apps/modules/git_module.py
+56 -1
View File
@@ -16,6 +16,7 @@ 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
@@ -41,6 +42,9 @@ _COMMANDS = (
"diff",
"log",
"lock",
"issue",
"run",
"workflow",
"commit",
"checkout",
"sync",
@@ -52,6 +56,8 @@ _COMMANDS = (
"pr",
)
_GH_PASSTHROUGH_COMMANDS = ("issue", "run", "workflow")
def _detect_branch_dir() -> tuple[str, Path] | None:
"""Detect caller's branch from CWD via passport lookup.
@@ -117,6 +123,8 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
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 == "diff":
@@ -152,6 +160,37 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
}
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:
@@ -451,6 +490,16 @@ def get_help(command: str | None = None) -> str:
Returns:
Help text string.
"""
if command == "issue":
return (
"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":
@@ -515,6 +564,9 @@ def get_help(command: str | None = None) -> str:
" diff [--staged] Show git diff for your branch\n"
" log [count] Show recent git log (default: 10)\n"
" lock Check lock status\n"
" issue [args] Passthrough to gh issue\n"
" run [args] Passthrough to gh run\n"
" workflow [args] Passthrough to gh workflow\n"
"\n"
"Owner (devpulse only):\n"
" commit <msg> [--all] Commit staged changes\n"
@@ -554,7 +606,10 @@ def get_introspective() -> str:
" - sync_plugin.py (smart_sync — fetch + rebase if behind)\n"
" - fix_plugin.py (fix_git_state — detect/fix broken states)\n"
"\n"
"Access Tiers: global (status, diff, log, lock) | owner (commit, checkout, sync, unlock, system-pr, merge, smart-sync, fix)\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"
)
src/aipass/drone/apps/plugins/devpulse_ops/auth.py
+1 -1
View File
@@ -26,7 +26,7 @@ ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
GIT_ACCESS_TIERS: dict[str, dict] = {
"global": {
"commands": ["status", "diff", "log", "lock"],
"commands": ["status", "diff", "log", "lock", "issue", "run", "workflow"],
"description": "Read-only — available to all branches",
},
"owner": {
src/aipass/drone/tests/test_git_access.py
+177
View File
@@ -575,3 +575,180 @@ class TestUpdatedHelp:
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
src/aipass/spawn/templates/builder/.spawn/.template_registry.json
+3 -3
View File
@@ -1,7 +1,7 @@
{
"metadata": {
"version": "1.0.0",
"last_updated": "2026-05-12",
"last_updated": "2026-05-10",
"description": "Template file tracking registry for ID-based updates"
},
"files": {
@@ -155,7 +155,7 @@
"content_hash": "a4cf0a8e3b4f",
"has_branch_placeholder": false
},
"f015": {
"f026": {
"path": "apps/modules/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",
@@ -263,7 +263,7 @@
"content_hash": "28e9ae373563",
"has_branch_placeholder": false
},
"f026": {
"f015": {
"path": "apps/plugins/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",