tegwick
310b43079d
Add dual-run flags/meter, harden task-status (idempotency, UUID, head, push), State Hub adapter for PATCH /tasks and C-15/reconcile proxy, pilot evidence, and finish RMGR-WP-0002.
84 lines
2.2 KiB
Python
84 lines
2.2 KiB
Python
"""Git helpers via subprocess (ADR-001)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
class GitError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _run(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
|
|
proc = subprocess.run(
|
|
["git", *args],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if check and proc.returncode != 0:
|
|
raise GitError(proc.stderr.strip() or proc.stdout.strip() or f"git {' '.join(args)} failed")
|
|
return proc
|
|
|
|
|
|
def head_sha(repo: Path) -> str | None:
|
|
proc = _run(repo, "rev-parse", "HEAD", check=False)
|
|
if proc.returncode != 0:
|
|
return None
|
|
return proc.stdout.strip() or None
|
|
|
|
|
|
def is_git_repo(repo: Path) -> bool:
|
|
return (repo / ".git").exists() or _run(repo, "rev-parse", "--git-dir", check=False).returncode == 0
|
|
|
|
|
|
def commit_paths(
|
|
repo: Path,
|
|
paths: list[str],
|
|
message: str,
|
|
*,
|
|
author_name: str = "repo-manager",
|
|
author_email: str = "repo-manager@local",
|
|
) -> str:
|
|
"""Stage paths and commit. Returns new HEAD sha."""
|
|
env_author = [
|
|
"-c",
|
|
f"user.name={author_name}",
|
|
"-c",
|
|
f"user.email={author_email}",
|
|
]
|
|
for path in paths:
|
|
_run(repo, "add", "--", path)
|
|
proc = subprocess.run(
|
|
["git", *env_author, "commit", "-m", message],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if proc.returncode != 0:
|
|
raise GitError(proc.stderr.strip() or proc.stdout.strip() or "git commit failed")
|
|
sha = head_sha(repo)
|
|
if not sha:
|
|
raise GitError("commit succeeded but HEAD missing")
|
|
return sha
|
|
|
|
|
|
def push_ff(repo: Path) -> tuple[bool, str]:
|
|
"""Best-effort push (push-seal compatible). Never force-pushes."""
|
|
try:
|
|
proc = subprocess.run(
|
|
["git", "push"],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
if proc.returncode == 0:
|
|
return True, (proc.stdout.strip() or "pushed")
|
|
return False, (proc.stderr.strip() or proc.stdout.strip() or "push failed")
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, str(exc)
|