Skip to content

Commit phase

commit_phase

Commit-phase hook action.

Stages all changes and commits with [axm] {phase_name}. Supports a from_outputs mode that reads commit_spec from context for targeted file staging.

CommitPhaseHook dataclass

Commit all changes with message [axm] {phase_name}.

Reads phase_name from context. An optional message_format can be provided via params (default "[axm] {phase}"). Skips gracefully when there is nothing to commit or no git repository.

When from_outputs=True is passed in params, reads commit_spec from context instead: {message, body?, files}. Only the listed files are staged (no git add -A).

Source code in packages/axm-git/src/axm_git/hooks/commit_phase.py
@dataclass
class CommitPhaseHook:
    """Commit all changes with message ``[axm] {phase_name}``.

    Reads ``phase_name`` from *context*.  An optional
    ``message_format`` can be provided via *params*
    (default ``"[axm] {phase}"``).  Skips gracefully when
    there is nothing to commit or no git repository.

    When ``from_outputs=True`` is passed in *params*, reads
    ``commit_spec`` from *context* instead: ``{message, body?, files}``.
    Only the listed files are staged (no ``git add -A``).
    """

    def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
        """Execute the hook action.

        Args:
            context: Session context dictionary.
            **params: Optional ``message_format``, ``from_outputs``,
                ``working_dir``.

        Returns:
            HookResult with ``commit`` hash and ``message`` in metadata.
        """
        working_dir = Path(params.get("working_dir", context.get("working_dir", ".")))

        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        if find_git_root(working_dir) is None:
            return HookResult.ok(skipped=True, reason="not a git repo")

        if params.get("from_outputs"):
            return self._commit_from_outputs(context, working_dir)

        return self._commit_legacy(context, working_dir, **params)

    def _commit_legacy(
        self,
        context: dict[str, Any],
        working_dir: Path,
        **params: Any,
    ) -> HookResult:
        """Legacy mode: stage all and commit with format string."""
        phase_name: str = context["phase_name"]

        # Stage all changes (scoped to working_dir for workspace layouts)
        run_git(["add", "-A", "."], working_dir)

        # Check if there's anything to commit
        status = run_git(["status", "--porcelain", "--", "."], working_dir)
        if not status.stdout.strip():
            return HookResult.ok(skipped=True, reason="nothing to commit")

        # Commit
        msg = params.get("message_format", "[axm] {phase}").format(
            phase=phase_name,
        )
        result = run_git(["commit", "-m", msg], working_dir)
        if result.returncode != 0:
            return HookResult.fail(f"git commit failed: {result.stderr}")

        # Get commit hash
        hash_result = run_git(["rev-parse", "--short", "HEAD"], working_dir)
        return HookResult.ok(commit=hash_result.stdout.strip(), message=msg)

    def _commit_from_outputs(
        self,
        context: dict[str, Any],
        working_dir: Path,
    ) -> HookResult:
        """Outputs mode: read commit_spec from context, stage listed files.

        If the commit fails because pre-commit hooks auto-fixed files
        (stderr contains ``"files were modified"``), the listed files are
        re-staged and the commit is retried once.
        """
        spec, err = _validate_commit_spec(context.get("commit_spec"))
        if err:
            return HookResult.fail(err)
        assert spec is not None  # guaranteed by _validate_commit_spec

        files: list[str] = spec["files"]
        message: str = spec["message"]
        body: str | None = spec.get("body")

        git_root = find_git_root(working_dir) or working_dir

        stage_err = _stage_spec_files(files, git_root)
        if stage_err:
            return HookResult.fail(stage_err)

        # Check if there's anything to commit
        status = run_git(["diff", "--cached", "--name-only"], git_root)
        if not status.stdout.strip():
            return HookResult.ok(skipped=True, reason="nothing to commit")

        # Build commit command
        commit_cmd = ["commit", "-m", message]
        if body:
            commit_cmd.extend(["-m", body])

        result = run_git(commit_cmd, git_root)
        if result.returncode != 0:
            # Pre-commit hooks may auto-fix files; re-stage and retry once
            if "files were modified" in result.stderr:
                restage_err = _stage_spec_files(files, git_root)
                if restage_err:
                    return HookResult.fail(restage_err)
                result = run_git(commit_cmd, git_root)
            if result.returncode != 0:
                return HookResult.fail(f"git commit failed: {result.stderr}")

        # Get commit hash
        hash_result = run_git(["rev-parse", "--short", "HEAD"], git_root)
        return HookResult.ok(commit=hash_result.stdout.strip(), message=message)
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

required
**params Any

Optional message_format, from_outputs, working_dir.

{}

Returns:

Type Description
HookResult

HookResult with commit hash and message in metadata.

Source code in packages/axm-git/src/axm_git/hooks/commit_phase.py
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``message_format``, ``from_outputs``,
            ``working_dir``.

    Returns:
        HookResult with ``commit`` hash and ``message`` in metadata.
    """
    working_dir = Path(params.get("working_dir", context.get("working_dir", ".")))

    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    if find_git_root(working_dir) is None:
        return HookResult.ok(skipped=True, reason="not a git repo")

    if params.get("from_outputs"):
        return self._commit_from_outputs(context, working_dir)

    return self._commit_legacy(context, working_dir, **params)