Runner

runner

Subprocess runners for git, gh, and uv commands.

detect_package_name(project_path)

Read the package name from pyproject.toml.

Parameters:

Name	Type	Description	Default
project_path	Path	Project root containing pyproject.toml.	required

Returns:

Type	Description
str | None	Package name or None if not found.

Source code in packages/axm-git/src/axm_git/core/runner.py

def detect_package_name(project_path: Path) -> str | None:
    """Read the package name from ``pyproject.toml``.

    Args:
        project_path: Project root containing ``pyproject.toml``.

    Returns:
        Package name or ``None`` if not found.
    """
    pyproject = project_path / "pyproject.toml"
    if not pyproject.exists():
        return None

    try:
        import tomllib

        with open(pyproject, "rb") as f:
            data = tomllib.load(f)
        return data.get("project", {}).get("name")  # type: ignore[no-any-return]
    except (OSError, KeyError, ValueError):
        return None

find_git_root(path)

Find the git repository root containing path.

Uses git rev-parse --show-toplevel which walks up the directory tree, supporting mono-repo and workspace layouts where .git lives above the package directory.

Parameters:

Name	Type	Description	Default
path	Path	Any directory that may be inside a git repository.	required

Returns:

Type	Description
Path | None	Repository root as a Path, or None if path is not
Path | None	inside a git repository.

Source code in packages/axm-git/src/axm_git/core/runner.py

def find_git_root(path: Path) -> Path | None:
    """Find the git repository root containing *path*.

    Uses ``git rev-parse --show-toplevel`` which walks up the directory
    tree, supporting mono-repo and workspace layouts where ``.git``
    lives above the package directory.

    Args:
        path: Any directory that may be inside a git repository.

    Returns:
        Repository root as a ``Path``, or ``None`` if *path* is not
        inside a git repository.
    """
    try:
        result = subprocess.run(
            ["git", "-C", str(path), "rev-parse", "--show-toplevel"],
            capture_output=True,
            text=True,
            check=False,
            timeout=DEFAULT_GIT_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        logger.warning("git rev-parse timed out after %ss", DEFAULT_GIT_TIMEOUT)
        return None
    if result.returncode != 0:
        return None
    return Path(result.stdout.strip())

gh_available()

Check whether the GitHub CLI is installed and authenticated.

Source code in packages/axm-git/src/axm_git/core/runner.py

def gh_available() -> bool:
    """Check whether the GitHub CLI is installed and authenticated."""
    if not shutil.which("gh"):
        return False
    try:
        result = subprocess.run(
            ["gh", "auth", "status"],
            capture_output=True,
            text=True,
            check=False,
            timeout=DEFAULT_GIT_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        logger.warning("gh auth status timed out after %ss", DEFAULT_GIT_TIMEOUT)
        return False
    return result.returncode == 0

not_a_repo_error(stderr, path)

Build a ToolResult for a failed git command.

If stderr contains "not a git repository" and path has child directories that are git repos, the error message is enriched with suggestions. Otherwise a standard error is returned.

Parameters:

Name	Type	Description	Default
stderr	str	Stderr output from the failed git command.	required
path	Path	Directory that was used as cwd.	required

Returns:

Type	Description
ToolResult	ToolResult(success=False, ...) with optional suggestions.

Source code in packages/axm-git/src/axm_git/core/runner.py

def not_a_repo_error(stderr: str, path: Path) -> ToolResult:
    """Build a ``ToolResult`` for a failed git command.

    If *stderr* contains ``"not a git repository"`` and *path* has
    child directories that are git repos, the error message is enriched
    with suggestions.  Otherwise a standard error is returned.

    Args:
        stderr: Stderr output from the failed git command.
        path: Directory that was used as ``cwd``.

    Returns:
        ``ToolResult(success=False, ...)`` with optional suggestions.
    """
    msg = stderr.strip()

    if "not a git repository" not in msg:
        return ToolResult(success=False, error=msg)

    suggestions = suggest_git_repos(path)
    if suggestions:
        hint = ", ".join(suggestions)
        return ToolResult(
            success=False,
            error=(
                f"{msg}. This directory contains git repos: {hint}. "
                f"Pass one of these as the path instead."
            ),
            data={"suggestions": suggestions},
        )

    return ToolResult(success=False, error=msg)

parse_porcelain_z(status_stdout)

Parse git status --porcelain -z output into {path, status} rows.

Records are NUL-terminated rather than newline-terminated, so paths with spaces are emitted verbatim (unquoted, unescaped). Rename/copy entries (R/C) span two NUL-separated fields — XY <space>dest followed by the original source path — so the destination is kept as path and the trailing source field is consumed and discarded.

Parameters:

Name	Type	Description	Default
status_stdout	str	Raw stdout from git status --porcelain -z.	required

Returns:

Type	Description
list[dict[str, str]]	List of {"path", "status"} dicts in encounter order.

Source code in packages/axm-git/src/axm_git/core/runner.py

def parse_porcelain_z(status_stdout: str) -> list[dict[str, str]]:
    """Parse ``git status --porcelain -z`` output into ``{path, status}`` rows.

    Records are NUL-terminated rather than newline-terminated, so paths with
    spaces are emitted verbatim (unquoted, unescaped). Rename/copy entries
    (``R``/``C``) span two NUL-separated fields — ``XY <space>dest`` followed
    by the original source path — so the destination is kept as ``path`` and
    the trailing source field is consumed and discarded.

    Args:
        status_stdout: Raw stdout from ``git status --porcelain -z``.

    Returns:
        List of ``{"path", "status"}`` dicts in encounter order.
    """
    records = [rec for rec in status_stdout.split("\x00") if rec]
    files: list[dict[str, str]] = []
    index = 0
    while index < len(records):
        record = records[index]
        index += 1
        if len(record) < _MIN_PORCELAIN_RECORD_LEN:
            continue
        status = record[:2].strip()
        files.append({"path": record[3:], "status": status})
        # Rename/copy entries carry a trailing source-path field; skip it.
        if record[:2].strip(" ?")[:1] in {"R", "C"}:
            index += 1
    return files

resolve_default_branch(working_dir)

Resolve the repository's default branch.

Reads git symbolic-ref refs/remotes/origin/HEAD (e.g. refs/remotes/origin/master) and strips the refs/remotes/origin/ prefix. Falls back to "main" when the command fails or returns an empty/unexpected value (for instance a repo with no origin/HEAD ref).

Parameters:

Name	Type	Description	Default
working_dir	Path	A directory inside the git repository.	required

Returns:

Type	Description
str	The default branch name, or "main" as a fallback.

Source code in packages/axm-git/src/axm_git/core/runner.py

def resolve_default_branch(working_dir: Path) -> str:
    """Resolve the repository's default branch.

    Reads ``git symbolic-ref refs/remotes/origin/HEAD`` (e.g.
    ``refs/remotes/origin/master``) and strips the
    ``refs/remotes/origin/`` prefix. Falls back to ``"main"`` when the
    command fails or returns an empty/unexpected value (for instance a
    repo with no ``origin/HEAD`` ref).

    Args:
        working_dir: A directory inside the git repository.

    Returns:
        The default branch name, or ``"main"`` as a fallback.
    """
    result = run_git(["symbolic-ref", "refs/remotes/origin/HEAD"], working_dir)
    ref = result.stdout.strip()
    if result.returncode != 0 or not ref.startswith(_ORIGIN_HEAD_PREFIX):
        return "main"
    branch = ref.removeprefix(_ORIGIN_HEAD_PREFIX)
    return branch or "main"

run_gh(args, cwd, *, timeout=DEFAULT_GH_TIMEOUT)

Run a GitHub CLI command.

Parameters:

Name	Type	Description	Default
args	list[str]	gh subcommand and arguments.	required
cwd	Path	Working directory (project root).	required
timeout	float | None	Subprocess timeout in seconds (default 120.0). Use None to disable.	DEFAULT_GH_TIMEOUT

Returns:

Type	Description
CompletedProcess[str]	Completed process result with capture_output=True and text=True.

Raises:

Type	Description
FileNotFoundError	If gh is not installed.
TimeoutExpired	If the command exceeds timeout. Callers should catch and convert via :func:timeout_error_result.

Source code in packages/axm-git/src/axm_git/core/runner.py

def run_gh(
    args: list[str],
    cwd: Path,
    *,
    timeout: float | None = DEFAULT_GH_TIMEOUT,
) -> subprocess.CompletedProcess[str]:
    """Run a GitHub CLI command.

    Args:
        args: gh subcommand and arguments.
        cwd: Working directory (project root).
        timeout: Subprocess timeout in seconds (default 120.0). Use
            ``None`` to disable.

    Returns:
        Completed process result with ``capture_output=True`` and ``text=True``.

    Raises:
        FileNotFoundError: If ``gh`` is not installed.
        subprocess.TimeoutExpired: If the command exceeds *timeout*.
            Callers should catch and convert via :func:`timeout_error_result`.
    """
    try:
        return subprocess.run(
            ["gh", *args],
            cwd=str(cwd),
            timeout=timeout,
            capture_output=True,
            text=True,
            check=False,
        )
    except subprocess.TimeoutExpired:
        logger.warning("gh %s timed out after %ss", args[0] if args else "", timeout)
        raise

run_git(args, cwd, *, timeout=DEFAULT_GIT_TIMEOUT)

Run a git command in the given directory.

Parameters:

Name	Type	Description	Default
args	list[str]	Git subcommand and arguments (e.g. ["status", "--short"]).	required
cwd	Path	Working directory (project root).	required
timeout	float | None	Subprocess timeout in seconds (default 30.0). Use None to disable.	DEFAULT_GIT_TIMEOUT

Returns:

Type	Description
CompletedProcess[str]	Completed process result with capture_output=True and text=True.

Raises:

Type	Description
TimeoutExpired	If the command exceeds timeout. Callers should catch and convert via :func:timeout_error_result.

Source code in packages/axm-git/src/axm_git/core/runner.py

def run_git(
    args: list[str],
    cwd: Path,
    *,
    timeout: float | None = DEFAULT_GIT_TIMEOUT,
) -> subprocess.CompletedProcess[str]:
    """Run a git command in the given directory.

    Args:
        args: Git subcommand and arguments (e.g. ``["status", "--short"]``).
        cwd: Working directory (project root).
        timeout: Subprocess timeout in seconds (default 30.0). Use
            ``None`` to disable.

    Returns:
        Completed process result with ``capture_output=True`` and ``text=True``.

    Raises:
        subprocess.TimeoutExpired: If the command exceeds *timeout*.
            Callers should catch and convert via :func:`timeout_error_result`.
    """
    try:
        return subprocess.run(
            ["git", *args],
            cwd=str(cwd),
            timeout=timeout,
            capture_output=True,
            text=True,
            check=False,
        )
    except subprocess.TimeoutExpired:
        logger.warning("git %s timed out after %ss", args[0] if args else "", timeout)
        raise

stage_spec_files(files, git_root, *, working_dir=None, warnings=None)

Stage each file in files, returning an error message on failure.

Paths in files are resolved against git_root first, then against working_dir (if provided and distinct), so both git-root-relative and package-relative inputs work transparently. Absolute inputs are accepted when they point inside git_root.

Tracked-but-deleted files (git status D) are staged as deletions. Gitignored files are skipped with a warning appended to warnings. Truly missing files (never tracked) produce a clear diagnostic error listing every absolute path that was attempted.

Source code in packages/axm-git/src/axm_git/core/runner.py

def stage_spec_files(
    files: list[str],
    git_root: Path,
    *,
    working_dir: Path | None = None,
    warnings: list[str] | None = None,
) -> str | None:
    """Stage each file in *files*, returning an error message on failure.

    Paths in *files* are resolved against *git_root* first, then against
    *working_dir* (if provided and distinct), so both git-root-relative
    and package-relative inputs work transparently. Absolute inputs are
    accepted when they point inside *git_root*.

    Tracked-but-deleted files (git status ``D``) are staged as deletions.
    Gitignored files are skipped with a warning appended to *warnings*.
    Truly missing files (never tracked) produce a clear diagnostic error
    listing every absolute path that was attempted.
    """
    for filepath in files:
        err = _stage_single_file(filepath, git_root, working_dir, warnings)
        if err:
            return err
    return None

suggest_git_repos(path)

Find immediate child directories that are git repositories.

Scans one level deep for subdirectories containing a .git dir. Returns a sorted list of directory names. If path is itself a git repository (has .git/ at root), returns an empty list.

Parameters:

Name	Type	Description	Default
path	Path	Directory to scan.	required

Returns:

Type	Description
list[str]	Sorted list of child directory names that are git repos.

Source code in packages/axm-git/src/axm_git/core/runner.py

def suggest_git_repos(path: Path) -> list[str]:
    """Find immediate child directories that are git repositories.

    Scans one level deep for subdirectories containing a ``.git`` dir.
    Returns a sorted list of directory names.  If *path* is itself a
    git repository (has ``.git/`` at root), returns an empty list.

    Args:
        path: Directory to scan.

    Returns:
        Sorted list of child directory names that are git repos.
    """
    if (path / ".git").is_dir():
        return []

    repos: list[str] = []
    try:
        children = sorted(path.iterdir())
    except (PermissionError, FileNotFoundError):
        return []

    for child in children:
        if not child.is_dir():
            continue
        try:
            if (child / ".git").is_dir():
                repos.append(child.name)
        except PermissionError:
            continue

    return repos

timeout_error_result(exc)

Build a ToolResult for a subprocess.TimeoutExpired.

Source code in packages/axm-git/src/axm_git/core/runner.py

def timeout_error_result(exc: subprocess.TimeoutExpired) -> ToolResult:
    """Build a ``ToolResult`` for a ``subprocess.TimeoutExpired``."""
    cmd = exc.cmd
    if isinstance(cmd, (list, tuple)) and cmd:
        cmd_str = str(cmd[0])
    else:
        cmd_str = str(cmd)
    return ToolResult(
        success=False,
        error=f"{cmd_str} timed out after {exc.timeout}s",
    )