diff --git a/.claude/hooks/README.md b/.claude/hooks/README.md index d997a04c..94e0cfe3 100644 --- a/.claude/hooks/README.md +++ b/.claude/hooks/README.md @@ -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 `(? 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() diff --git a/src/aipass/devpulse/.claude/hooks/branch_prompt_loader.py b/src/aipass/devpulse/.claude/hooks/branch_prompt_loader.py index c3d2e94e..a38d03f9 100644 --- a/src/aipass/devpulse/.claude/hooks/branch_prompt_loader.py +++ b/src/aipass/devpulse/.claude/hooks/branch_prompt_loader.py @@ -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() diff --git a/src/aipass/devpulse/.claude/hooks/email_notification.py b/src/aipass/devpulse/.claude/hooks/email_notification.py index 70f6df76..401663cd 100644 --- a/src/aipass/devpulse/.claude/hooks/email_notification.py +++ b/src/aipass/devpulse/.claude/hooks/email_notification.py @@ -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() diff --git a/src/aipass/devpulse/.claude/hooks/identity_injector.py b/src/aipass/devpulse/.claude/hooks/identity_injector.py index 69f25ff3..4eaa4463 100644 --- a/src/aipass/devpulse/.claude/hooks/identity_injector.py +++ b/src/aipass/devpulse/.claude/hooks/identity_injector.py @@ -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() diff --git a/src/aipass/devpulse/.claude/hooks/pre_compact.py b/src/aipass/devpulse/.claude/hooks/pre_compact.py index 2e2c7e40..717ebb09 100644 --- a/src/aipass/devpulse/.claude/hooks/pre_compact.py +++ b/src/aipass/devpulse/.claude/hooks/pre_compact.py @@ -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() diff --git a/src/aipass/devpulse/.claude/hooks/subagent_stop_gate.py b/src/aipass/devpulse/.claude/hooks/subagent_stop_gate.py index edbef876..b903b885 100644 --- a/src/aipass/devpulse/.claude/hooks/subagent_stop_gate.py +++ b/src/aipass/devpulse/.claude/hooks/subagent_stop_gate.py @@ -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/) 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() diff --git a/src/aipass/drone/README.md b/src/aipass/drone/README.md index 8060b3cc..85e6d12f 100644 --- a/src/aipass/drone/README.md +++ b/src/aipass/drone/README.md @@ -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 # 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) diff --git a/src/aipass/drone/apps/modules/git_module.py b/src/aipass/drone/apps/modules/git_module.py index 89b91e43..eaea5c3a 100644 --- a/src/aipass/drone/apps/modules/git_module.py +++ b/src/aipass/drone/apps/modules/git_module.py @@ -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 , watch \n" + if command == "workflow": + return ( + "git workflow [args] — Passthrough to gh workflow CLI [global]\n Examples: list, view , run \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 [--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 [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" ) diff --git a/src/aipass/drone/apps/plugins/devpulse_ops/auth.py b/src/aipass/drone/apps/plugins/devpulse_ops/auth.py index 1d0ae07b..4c5a92f7 100644 --- a/src/aipass/drone/apps/plugins/devpulse_ops/auth.py +++ b/src/aipass/drone/apps/plugins/devpulse_ops/auth.py @@ -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": { diff --git a/src/aipass/drone/tests/test_git_access.py b/src/aipass/drone/tests/test_git_access.py index dce63094..957e5082 100644 --- a/src/aipass/drone/tests/test_git_access.py +++ b/src/aipass/drone/tests/test_git_access.py @@ -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 diff --git a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json b/src/aipass/spawn/templates/builder/.spawn/.template_registry.json index 79d7b496..e5e17e8e 100644 --- a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json +++ b/src/aipass/spawn/templates/builder/.spawn/.template_registry.json @@ -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",