feat(system): add dev-pr, branches, delete-branch handlers + auth tier updates

Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-05-12 21:01:47 -07:00
co-authored by @devpulse
parent 004d05508f
commit 4fe7c15cd3
7 changed files with 248 additions and 1 deletions
+1
View File
@@ -0,0 +1 @@
{"file": "/home/patrick/Projects/AIPass/src/aipass/drone/apps/modules/git_module.py", "errors": [{"line": 665, "message": "E501: Line too long (156 > 120)"}]}
@@ -8,3 +8,6 @@ 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
from . import dev_pr_handler as dev_pr_handler
from . import branches_handler as branches_handler
from . import delete_branch_handler as delete_branch_handler
@@ -0,0 +1,54 @@
# =================== AIPass ====================
# Name: branches_handler.py
# Description: Remote branches handler — list remote branches
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Remote branches handler — list remote branches."""
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 list_remote_branches() -> dict:
"""List remote branches with origin/ prefix stripped.
Returns:
Dict with branches list, count, and message.
"""
repo_root = find_repo_root()
try:
result = subprocess.run(
["git", "branch", "-r"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git branch -r failed: %s", exc)
return {"branches": [], "count": 0, "message": f"Failed to list branches: {exc}"}
if result.returncode != 0:
return {"branches": [], "count": 0, "message": f"git branch -r error: {result.stderr.strip()}"}
branches = []
for line in result.stdout.splitlines():
name = line.strip()
if not name or " -> " in name:
continue
if name.startswith("origin/"):
name = name[len("origin/") :]
branches.append(name)
json_handler.log_operation("list_remote_branches", {"count": len(branches)})
logger.info("Listed %d remote branches", len(branches))
return {"branches": branches, "count": len(branches), "message": f"{len(branches)} remote branches"}
@@ -0,0 +1,50 @@
# =================== AIPass ====================
# Name: delete_branch_handler.py
# Description: Delete branch handler — remove remote branches with safety guards
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Delete branch handler — remove remote branches with safety guards."""
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
_PROTECTED_BRANCHES = ("main", "dev")
def delete_remote_branch(branch_name: str) -> dict:
"""Delete a remote branch with hard guards on protected branches.
Returns:
Dict with success and message keys.
"""
if branch_name in _PROTECTED_BRANCHES:
return {"success": False, "message": f"Refusing to delete protected branch '{branch_name}'."}
repo_root = find_repo_root()
try:
result = subprocess.run(
["git", "push", "origin", "--delete", branch_name],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git push origin --delete %s failed: %s", branch_name, exc)
return {"success": False, "message": f"Delete failed: {exc}"}
if result.returncode != 0:
return {"success": False, "message": f"Delete failed: {result.stderr.strip()}"}
json_handler.log_operation("delete_remote_branch", {"branch": branch_name})
logger.info("Deleted remote branch: %s", branch_name)
return {"success": True, "message": f"Deleted remote branch '{branch_name}'."}
@@ -0,0 +1,84 @@
# =================== AIPass ====================
# Name: dev_pr_handler.py
# Description: Dev branch PR handler — push dev and create PR to main
# Version: 1.0.0
# Created: 2026-05-12
# Modified: 2026-05-12
# =============================================
"""Dev branch PR handler — push dev and create PR to main."""
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 create_dev_pr(description: str) -> dict:
"""Push dev branch and create a PR to main.
Verifies HEAD is on dev, pushes to origin, then creates a PR via gh CLI.
Returns:
Dict with success, message, and pr_url keys.
"""
repo_root = find_repo_root()
try:
head = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("Failed to detect current branch: %s", exc)
return {"success": False, "message": f"Failed to detect current branch: {exc}", "pr_url": ""}
current_branch = head.stdout.strip()
if current_branch != "dev":
return {
"success": False,
"message": f"Not on dev branch (current: {current_branch}). Switch to dev first.",
"pr_url": "",
}
try:
push = subprocess.run(
["git", "push", "origin", "dev"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git push origin dev failed: %s", exc)
return {"success": False, "message": f"Push failed: {exc}", "pr_url": ""}
if push.returncode != 0:
return {"success": False, "message": f"Push failed: {push.stderr.strip()}", "pr_url": ""}
try:
pr = subprocess.run(
["gh", "pr", "create", "--head", "dev", "--base", "main", "--title", description],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except FileNotFoundError as exc:
logger.warning("gh CLI not found: %s", exc)
return {"success": False, "message": "gh CLI not found. Install: https://cli.github.com/", "pr_url": ""}
except (OSError, subprocess.SubprocessError) as exc:
logger.error("gh pr create failed: %s", exc)
return {"success": False, "message": f"PR creation failed: {exc}", "pr_url": ""}
if pr.returncode != 0:
return {"success": False, "message": f"PR creation failed: {pr.stderr.strip()}", "pr_url": ""}
pr_url = pr.stdout.strip()
json_handler.log_operation("create_dev_pr", {"pr_url": pr_url, "description": description})
logger.info("Dev PR created: %s", pr_url)
return {"success": True, "message": f"PR created: {pr_url}", "pr_url": pr_url}
@@ -29,6 +29,9 @@ from aipass.drone.apps.handlers.git import (
log_handler,
commit_handler,
checkout_handler,
dev_pr_handler,
branches_handler,
delete_branch_handler,
)
DRONE_MODULE = {
@@ -42,6 +45,7 @@ _COMMANDS = (
"diff",
"log",
"lock",
"branches",
"issue",
"run",
"workflow",
@@ -49,6 +53,8 @@ _COMMANDS = (
"checkout",
"sync",
"unlock",
"dev-pr",
"delete-branch",
"system-pr",
"merge",
"smart-sync",
@@ -133,6 +139,12 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
return _handle_log(args)
if command == "lock":
return _handle_lock()
if command == "branches":
return _handle_branches()
if command == "dev-pr":
return _handle_dev_pr(args)
if command == "delete-branch":
return _handle_delete_branch(args)
if command == "commit":
return _handle_commit(args)
if command == "checkout":
@@ -191,6 +203,47 @@ def _handle_gh_passthrough(subcommand: str, args: list[str]) -> dict:
}
def _handle_branches() -> dict:
"""Handle the branches subcommand (global tier)."""
result = branches_handler.list_remote_branches()
if result["branches"]:
return {
"stdout": "\n".join(result["branches"]),
"stderr": "",
"exit_code": 0,
}
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
def _handle_dev_pr(args: list[str]) -> dict:
"""Handle the dev-pr subcommand (owner tier)."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git dev-pr <description>",
"exit_code": 1,
}
description = " ".join(args)
result = dev_pr_handler.create_dev_pr(description)
if result["success"]:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _handle_delete_branch(args: list[str]) -> dict:
"""Handle the delete-branch subcommand (owner tier)."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git delete-branch <name>",
"exit_code": 1,
}
result = delete_branch_handler.delete_remote_branch(args[0])
if result["success"]:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
return {"stdout": "", "stderr": result["message"], "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:
@@ -26,7 +26,7 @@ ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
GIT_ACCESS_TIERS: dict[str, dict] = {
"global": {
"commands": ["status", "diff", "log", "lock", "issue", "run", "workflow"],
"commands": ["status", "diff", "log", "lock", "issue", "run", "workflow", "branches"],
"description": "Read-only — available to all branches",
},
"owner": {
@@ -39,6 +39,8 @@ GIT_ACCESS_TIERS: dict[str, dict] = {
"merge",
"smart-sync",
"fix",
"dev-pr",
"delete-branch",
],
"allowed_callers": ["devpulse"],
"description": "Write operations — project owner only",