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

feat(drone): add generic drone @git pr command — works from any branch including main

Browse Source
This commit is contained in:
AIOSAI
2026-05-15 13:05:53 -07:00
parent 974d697563
commit 7b5a073f5b
3 changed files with 116 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/drone/apps/handlers/git/dev_pr_handler.py
+97 -1
View File
@@ -6,10 +6,11 @@
# Modified: 2026-05-12
# =============================================
"""Dev branch PR handler — push dev and create PR to main."""
"""Branch PR handlers — generic and dev-specific."""
from __future__ import annotations
import re
import subprocess
from aipass.prax import logger
@@ -17,6 +18,101 @@ from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
def _slugify(text: str) -> str:
"""Convert text to a branch-safe slug."""
slug = text.lower().strip().replace(" ", "-")
slug = re.sub(r"[^a-z0-9-]", "", slug)
slug = re.sub(r"-{2,}", "-", slug).strip("-")
return slug[:40]
def create_branch_pr(description: str, target_branch: str = "main") -> dict:
"""Push current branch and create a PR to target.
Works from any branch. Detects HEAD, pushes to origin, creates 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": ""}
pr_branch = head.stdout.strip()
if pr_branch == target_branch:
pr_branch = _slugify(description)
if not pr_branch:
return {"success": False, "message": "Description required to generate branch name.", "pr_url": ""}
push_refspec = f"{target_branch}:{pr_branch}"
else:
push_refspec = pr_branch
try:
push = subprocess.run(
["git", "push", "-u", "origin", push_refspec],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git push 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", pr_branch, "--base", target_branch, "--title", description, "--body", ""],
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:
stderr = pr.stderr.strip()
if "already exists" in stderr:
existing_url = ""
for line in stderr.splitlines():
if "github.com" in line:
existing_url = line.strip()
break
msg = (
f"Pushed to {pr_branch}. PR already open: {existing_url}"
if existing_url
else f"Pushed to {pr_branch}. PR already open."
)
json_handler.log_operation(
"branch_pr_push_existing", {"pr_url": existing_url, "branch": pr_branch, "description": description}
)
return {"success": True, "message": msg, "pr_url": existing_url}
return {"success": False, "message": f"PR creation failed: {stderr}", "pr_url": ""}
pr_url = pr.stdout.strip()
json_handler.log_operation(
"create_branch_pr", {"pr_url": pr_url, "branch": pr_branch, "target": target_branch, "description": description}
)
logger.info("PR created from %s: %s", pr_branch, pr_url)
return {"success": True, "message": f"PR created: {pr_url}", "pr_url": pr_url}
def create_dev_pr(description: str) -> dict:
"""Push dev branch and create a PR to main.
src/aipass/drone/apps/modules/git_module.py
+18 -17
View File
@@ -55,7 +55,6 @@ _COMMANDS = (
"unlock",
"dev-pr",
"delete-branch",
"system-pr",
"merge",
"smart-sync",
"fix",
@@ -153,8 +152,6 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
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":
@@ -162,7 +159,7 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
if command == "fix":
return _handle_fix(args, caller)
if command == "pr":
return {"stdout": "", "stderr": "Agent PRs are deprecated.", "exit_code": 1}
return _handle_pr(args)
available = ", ".join(_COMMANDS)
return {
@@ -215,6 +212,21 @@ def _handle_branches() -> dict:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
def _handle_pr(args: list[str]) -> dict:
"""Handle the pr subcommand — push current branch and create PR to main."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git pr <description>",
"exit_code": 1,
}
description = " ".join(args)
result = dev_pr_handler.create_branch_pr(description)
if result["success"]:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _handle_dev_pr(args: list[str]) -> dict:
"""Handle the dev-pr subcommand (owner tier)."""
if not args:
@@ -244,15 +256,6 @@ def _handle_delete_branch(args: list[str]) -> dict:
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _handle_system_pr(_args: list[str], _caller: str) -> dict:
"""Handle the system-pr subcommand — DEPRECATED."""
return {
"stdout": "",
"stderr": "system-pr is deprecated. Use: drone @git dev-pr <description>",
"exit_code": 1,
}
def _handle_merge(args: list[str], caller: str) -> dict:
"""Handle the merge subcommand (owner-tier, auth pre-checked)."""
if not args:
@@ -542,7 +545,7 @@ def get_help(command: str | None = None) -> str:
"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"
return "git pr <description> — Push current branch and create PR to main [owner]\n"
if command == "status":
return "git status [--all] — Show git status filtered to your branch (--all for repo-wide) [global]\n"
if command == "diff":
@@ -585,8 +588,6 @@ def get_help(command: str | None = None) -> str:
)
if command == "unlock":
return "git unlock --force — Force-release the PR lock [owner]\n"
if command == "system-pr":
return "git system-pr — DEPRECATED. Use: drone @git dev-pr <description>\n"
if command == "merge":
return (
"git merge <PR#> — Merge a PR and sync local main [owner]\n"
@@ -624,13 +625,13 @@ def get_help(command: str | None = None) -> str:
"Owner (devpulse only):\n"
" commit <msg> [--all | files] Commit changes (selective or --all)\n"
" checkout <main|dev> Switch branches\n"
" pr <desc> Push current branch and create PR to main\n"
" dev-pr <desc> Push dev and create PR to main\n"
" delete-branch <name> Delete a remote branch\n"
" merge <PR#> Merge a PR\n"
" sync [--autostash] Checkout main and pull\n"
" smart-sync Fetch + rebase if behind\n"
" unlock --force Force-release the PR lock\n"
" system-pr DEPRECATED (use dev-pr)\n"
" fix [--dry-run] Fix broken git states\n"
)
src/aipass/drone/apps/plugins/devpulse_ops/auth.py
+1 -4
View File
@@ -35,11 +35,11 @@ GIT_ACCESS_TIERS: dict[str, dict] = {
"checkout",
"sync",
"unlock",
"system-pr",
"merge",
"smart-sync",
"fix",
"dev-pr",
"pr",
"delete-branch",
],
"allowed_callers": ["devpulse"],
@@ -121,9 +121,6 @@ def verify_git_access(command: str) -> str:
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"]