Implement observe/reconcile/update-task-status CLI path: workplan parse, JSON projection index, git-backed task status writeback with correlation events, and E2E pytest plus evidence artifacts. Finish foundation workplan.
66 lines
1.7 KiB
Python
66 lines
1.7 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
|