Skip to content

Index

hooks

Git hook actions for AXM lifecycle hooks.

Provides BranchDeleteHook, CreateBranchHook, CommitPhaseHook, MergeSquashHook, PreflightHook, WorktreeAddHook, WorktreeRemoveHook, PushHook, CreatePRHook, and AwaitMergeHook, auto-discovered by HookRegistry via the axm.hooks entry-point group.

AwaitMergeHook dataclass

Poll a PR until merged or timeout.

Reads pr_number (or pr_url) from context and polls gh pr view --json state every 30 seconds. Times out after 10 minutes by default.

Source code in packages/axm-git/src/axm_git/hooks/await_merge.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass
class AwaitMergeHook:
    """Poll a PR until merged or timeout.

    Reads ``pr_number`` (or ``pr_url``) from *context* and polls
    ``gh pr view --json state`` every 30 seconds.  Times out after
    10 minutes by default.
    """

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

        Args:
            context: Session context dictionary (must contain
                ``pr_number`` or ``pr_url``).
            **params: Optional ``enabled``, ``timeout`` (seconds),
                ``interval`` (seconds).

        Returns:
            HookResult with ``merged=True`` on success.
        """
        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        if not gh_available():
            return HookResult.ok(skipped=True, reason="gh not available")

        working_dir = _resolve_working_dir(params, context)

        pr_ref = (
            params.get("pr_number")
            or params.get("pr_url")
            or context.get("pr_number")
            or context.get("pr_url")
        )
        if not pr_ref:
            return HookResult.fail("no pr_number or pr_url in params or context")

        timeout = int(params.get("timeout", _DEFAULT_TIMEOUT))
        interval = int(params.get("interval", _DEFAULT_INTERVAL))

        elapsed = 0
        while elapsed < timeout:
            state = _poll_pr_state(str(pr_ref), working_dir)
            if state is None:
                return HookResult.fail(f"failed to query PR {pr_ref} state")
            if state == "MERGED":
                return HookResult.ok(merged=True, pr_ref=str(pr_ref))
            if state == "CLOSED":
                return HookResult.fail(f"PR {pr_ref} was closed without merging")

            time.sleep(interval)
            elapsed += interval

        return HookResult.fail(f"PR {pr_ref} not merged after {timeout}s timeout")
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary (must contain pr_number or pr_url).

 required
**params Any

Optional enabled, timeout (seconds), interval (seconds).

 {}

Returns:

Type Description
HookResult

HookResult with merged=True on success.
Source code in packages/axm-git/src/axm_git/hooks/await_merge.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary (must contain
            ``pr_number`` or ``pr_url``).
        **params: Optional ``enabled``, ``timeout`` (seconds),
            ``interval`` (seconds).

    Returns:
        HookResult with ``merged=True`` on success.
    """
    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    if not gh_available():
        return HookResult.ok(skipped=True, reason="gh not available")

    working_dir = _resolve_working_dir(params, context)

    pr_ref = (
        params.get("pr_number")
        or params.get("pr_url")
        or context.get("pr_number")
        or context.get("pr_url")
    )
    if not pr_ref:
        return HookResult.fail("no pr_number or pr_url in params or context")

    timeout = int(params.get("timeout", _DEFAULT_TIMEOUT))
    interval = int(params.get("interval", _DEFAULT_INTERVAL))

    elapsed = 0
    while elapsed < timeout:
        state = _poll_pr_state(str(pr_ref), working_dir)
        if state is None:
            return HookResult.fail(f"failed to query PR {pr_ref} state")
        if state == "MERGED":
            return HookResult.ok(merged=True, pr_ref=str(pr_ref))
        if state == "CLOSED":
            return HookResult.fail(f"PR {pr_ref} was closed without merging")

        time.sleep(interval)
        elapsed += interval

    return HookResult.fail(f"PR {pr_ref} not merged after {timeout}s timeout")

BranchDeleteHook dataclass

Delete a branch by name.

Branch name priority:

  1. branch param (direct override)
  2. branch context key

Skips gracefully when disabled or when the working directory is not a git repository.

Source code in packages/axm-git/src/axm_git/hooks/branch_delete.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@dataclass
class BranchDeleteHook:
    """Delete a branch by name.

    Branch name priority:

    1. ``branch`` param (direct override)
    2. ``branch`` context key

    Skips gracefully when disabled or when the working directory is not
    a git repository.
    """

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

        Args:
            context: Session context dictionary.
            **params: Optional ``branch``, ``enabled``, ``working_dir``.

        Returns:
            HookResult with ``branch`` and ``deleted`` in metadata on success.
        """
        working_dir = _resolve_working_dir(params, context)

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

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

        branch = params.get("branch") or context.get("branch")
        if not branch:
            return HookResult.fail("no branch specified in params or context")

        result = run_git(["branch", "-D", branch], git_root)
        if result.returncode != 0:
            return HookResult.fail(f"git branch -D failed: {result.stderr}")

        return HookResult.ok(branch=branch, deleted=True)
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

 required
**params Any

Optional branch, enabled, working_dir.

 {}

Returns:

Type Description
HookResult

HookResult with branch and deleted in metadata on success.
Source code in packages/axm-git/src/axm_git/hooks/branch_delete.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``branch``, ``enabled``, ``working_dir``.

    Returns:
        HookResult with ``branch`` and ``deleted`` in metadata on success.
    """
    working_dir = _resolve_working_dir(params, context)

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

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

    branch = params.get("branch") or context.get("branch")
    if not branch:
        return HookResult.fail("no branch specified in params or context")

    result = run_git(["branch", "-D", branch], git_root)
    if result.returncode != 0:
        return HookResult.fail(f"git branch -D failed: {result.stderr}")

    return HookResult.ok(branch=branch, deleted=True)

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 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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)

CreateBranchHook dataclass

Create a session branch.

Branch name priority:

  1. branch param (direct override)
  2. ticket_id + ticket_title params → branch_name_from_ticket()
  3. {prefix}/{session_id} (legacy fallback)

Skips gracefully when the working directory is not a git repository.

Source code in packages/axm-git/src/axm_git/hooks/create_branch.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@dataclass
class CreateBranchHook:
    """Create a session branch.

    Branch name priority:

    1. ``branch`` param (direct override)
    2. ``ticket_id`` + ``ticket_title`` params → ``branch_name_from_ticket()``
    3. ``{prefix}/{session_id}`` (legacy fallback)

    Skips gracefully when the working directory is not a git repository.
    """

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

        Args:
            context: Session context dictionary (must contain ``session_id``).
            **params: Optional ``branch``, ``ticket_id``, ``ticket_title``,
                ``ticket_labels``, ``prefix`` (default ``"axm"``).

        Returns:
            HookResult with ``branch`` in metadata on success.
        """
        working_dir = _resolve_working_dir(params, context)
        session_id: str = context["session_id"]

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

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

        branch = self._resolve_branch(params, session_id)

        result = run_git(["checkout", "-b", branch], git_root)
        if result.returncode != 0:
            return HookResult.fail(f"git checkout -b failed: {result.stderr}")

        return HookResult.ok(branch=branch)

    @staticmethod
    def _resolve_branch(params: dict[str, Any], session_id: str) -> str:
        """Resolve branch name from params with fallback to session_id."""
        if branch := params.get("branch"):
            return str(branch)

        ticket_id = params.get("ticket_id")
        ticket_title = params.get("ticket_title")
        if ticket_id and ticket_title:
            labels = params.get("ticket_labels", [])
            return branch_name_from_ticket(ticket_id, ticket_title, labels)

        prefix = params.get("prefix", "axm")
        return f"{prefix}/{session_id}"
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary (must contain session_id).

 required
**params Any

Optional branch, ticket_id, ticket_title, ticket_labels, prefix (default "axm").

 {}

Returns:

Type Description
HookResult

HookResult with branch in metadata on success.
Source code in packages/axm-git/src/axm_git/hooks/create_branch.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary (must contain ``session_id``).
        **params: Optional ``branch``, ``ticket_id``, ``ticket_title``,
            ``ticket_labels``, ``prefix`` (default ``"axm"``).

    Returns:
        HookResult with ``branch`` in metadata on success.
    """
    working_dir = _resolve_working_dir(params, context)
    session_id: str = context["session_id"]

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

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

    branch = self._resolve_branch(params, session_id)

    result = run_git(["checkout", "-b", branch], git_root)
    if result.returncode != 0:
        return HookResult.fail(f"git checkout -b failed: {result.stderr}")

    return HookResult.ok(branch=branch)

CreatePRHook dataclass

Create a GitHub PR with auto-merge squash.

Reads branch, commit_spec, and ticket_id from context. Runs gh pr create followed by gh pr merge --auto --squash. Skips gracefully when gh is not installed.

Source code in packages/axm-git/src/axm_git/hooks/create_pr.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@dataclass
class CreatePRHook:
    """Create a GitHub PR with auto-merge squash.

    Reads ``branch``, ``commit_spec``, and ``ticket_id`` from *context*.
    Runs ``gh pr create`` followed by ``gh pr merge --auto --squash``.
    Skips gracefully when ``gh`` is not installed.
    """

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

        Args:
            context: Session context dictionary.
            **params: Optional ``enabled`` (default ``True``),
                ``base`` (default ``"main"``), ``commit_spec``,
                ``ticket_id``.  Params take precedence over context.

        Returns:
            HookResult with ``pr_url`` and ``pr_number`` in metadata.
        """
        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        if not gh_available():
            return HookResult.ok(skipped=True, reason="gh not available")

        working_dir = _resolve_working_dir(params, context)

        commit_spec: dict[str, Any] = params.get(
            "commit_spec", context.get("commit_spec", {})
        )
        ticket_id: str = params.get("ticket_id", context.get("ticket_id", ""))
        base = params.get("base", "main")

        title = _format_pr_title(commit_spec, ticket_id)
        body = commit_spec.get("body", "")

        # Create the PR
        create_args = [
            "pr",
            "create",
            "--title",
            title,
            "--body",
            body,
            "--base",
            base,
        ]
        result = run_gh(create_args, working_dir)

        if result.returncode != 0:
            stderr = result.stderr.strip()
            if "already exists" in stderr:
                return _recover_existing_pr(working_dir)
            return HookResult.fail(f"gh pr create failed: {stderr}")

        pr_url = result.stdout.strip()

        # Extract PR number from URL
        pr_number = pr_url.rstrip("/").rsplit("/", maxsplit=1)[-1]

        # Enable auto-merge
        merge_result = run_gh(
            ["pr", "merge", pr_number, "--auto", "--squash"],
            working_dir,
        )
        if merge_result.returncode != 0:
            # Non-fatal: PR created but auto-merge not enabled
            return HookResult.ok(
                pr_url=pr_url,
                pr_number=pr_number,
                auto_merge=False,
                auto_merge_error=merge_result.stderr.strip(),
            )

        return HookResult.ok(
            pr_url=pr_url,
            pr_number=pr_number,
            auto_merge=True,
        )
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

 required
**params Any

Optional enabled (default True), base (default "main"), commit_spec, ticket_id. Params take precedence over context.

 {}

Returns:

Type Description
HookResult

HookResult with pr_url and pr_number in metadata.
Source code in packages/axm-git/src/axm_git/hooks/create_pr.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``enabled`` (default ``True``),
            ``base`` (default ``"main"``), ``commit_spec``,
            ``ticket_id``.  Params take precedence over context.

    Returns:
        HookResult with ``pr_url`` and ``pr_number`` in metadata.
    """
    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    if not gh_available():
        return HookResult.ok(skipped=True, reason="gh not available")

    working_dir = _resolve_working_dir(params, context)

    commit_spec: dict[str, Any] = params.get(
        "commit_spec", context.get("commit_spec", {})
    )
    ticket_id: str = params.get("ticket_id", context.get("ticket_id", ""))
    base = params.get("base", "main")

    title = _format_pr_title(commit_spec, ticket_id)
    body = commit_spec.get("body", "")

    # Create the PR
    create_args = [
        "pr",
        "create",
        "--title",
        title,
        "--body",
        body,
        "--base",
        base,
    ]
    result = run_gh(create_args, working_dir)

    if result.returncode != 0:
        stderr = result.stderr.strip()
        if "already exists" in stderr:
            return _recover_existing_pr(working_dir)
        return HookResult.fail(f"gh pr create failed: {stderr}")

    pr_url = result.stdout.strip()

    # Extract PR number from URL
    pr_number = pr_url.rstrip("/").rsplit("/", maxsplit=1)[-1]

    # Enable auto-merge
    merge_result = run_gh(
        ["pr", "merge", pr_number, "--auto", "--squash"],
        working_dir,
    )
    if merge_result.returncode != 0:
        # Non-fatal: PR created but auto-merge not enabled
        return HookResult.ok(
            pr_url=pr_url,
            pr_number=pr_number,
            auto_merge=False,
            auto_merge_error=merge_result.stderr.strip(),
        )

    return HookResult.ok(
        pr_url=pr_url,
        pr_number=pr_number,
        auto_merge=True,
    )

MergeSquashHook dataclass

Merge session branch back to main with squash.

Branch name priority: branch param > context["branch"]

{prefix}/{session_id} fallback.

Commit message priority: message param > [AXM] {protocol_name}: {session_id} fallback.

Source code in packages/axm-git/src/axm_git/hooks/merge_squash.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@dataclass
class MergeSquashHook:
    """Merge session branch back to main with squash.

    Branch name priority: ``branch`` param > ``context["branch"]``
    > ``{prefix}/{session_id}`` fallback.

    Commit message priority: ``message`` param >
    ``[AXM] {protocol_name}: {session_id}`` fallback.
    """

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

        Args:
            context: Session context dictionary (must contain
                ``session_id`` and ``protocol_name``).
            **params: Optional ``branch``, ``message``, ``prefix``,
                and ``target_branch``.

        Returns:
            HookResult with ``merged``, ``into``, and ``message`` in metadata.
        """
        working_dir = _resolve_working_dir(params, context)
        session_id: str = context["session_id"]
        protocol_name: str = context["protocol_name"]

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

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

        branch = self._resolve_branch(params, context, session_id)
        target = params.get("target_branch", "main")

        # Checkout target branch
        result = run_git(["checkout", target], git_root)
        if result.returncode != 0:
            return HookResult.fail(f"checkout {target} failed: {result.stderr}")

        # Merge with squash
        result = run_git(["merge", "--squash", branch], git_root)
        if result.returncode != 0:
            return HookResult.fail(f"merge --squash failed: {result.stderr}")

        # Commit
        msg = params.get("message") or f"[AXM] {protocol_name}: {session_id}"
        result = run_git(["commit", "-m", msg], git_root)
        if result.returncode != 0:
            return HookResult.fail(f"commit failed: {result.stderr}")

        return HookResult.ok(merged=branch, into=target, message=msg)

    @staticmethod
    def _resolve_branch(
        params: dict[str, Any],
        context: dict[str, Any],
        session_id: str,
    ) -> str:
        """Resolve branch name from params, context, then fallback."""
        if branch := params.get("branch"):
            return str(branch)
        if branch := context.get("branch"):
            return str(branch)
        prefix = params.get("prefix", "axm")
        return f"{prefix}/{session_id}"
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary (must contain session_id and protocol_name).

 required
**params Any

Optional branch, message, prefix, and target_branch.

 {}

Returns:

Type Description
HookResult

HookResult with merged, into, and message in metadata.
Source code in packages/axm-git/src/axm_git/hooks/merge_squash.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary (must contain
            ``session_id`` and ``protocol_name``).
        **params: Optional ``branch``, ``message``, ``prefix``,
            and ``target_branch``.

    Returns:
        HookResult with ``merged``, ``into``, and ``message`` in metadata.
    """
    working_dir = _resolve_working_dir(params, context)
    session_id: str = context["session_id"]
    protocol_name: str = context["protocol_name"]

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

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

    branch = self._resolve_branch(params, context, session_id)
    target = params.get("target_branch", "main")

    # Checkout target branch
    result = run_git(["checkout", target], git_root)
    if result.returncode != 0:
        return HookResult.fail(f"checkout {target} failed: {result.stderr}")

    # Merge with squash
    result = run_git(["merge", "--squash", branch], git_root)
    if result.returncode != 0:
        return HookResult.fail(f"merge --squash failed: {result.stderr}")

    # Commit
    msg = params.get("message") or f"[AXM] {protocol_name}: {session_id}"
    result = run_git(["commit", "-m", msg], git_root)
    if result.returncode != 0:
        return HookResult.fail(f"commit failed: {result.stderr}")

    return HookResult.ok(merged=branch, into=target, message=msg)

PreflightHook dataclass

Report working-tree status and diff as a pre-hook.

Designed for injection into protocol briefings via inject_result + inline: true.

params: path — project root (default "."). diff_lines — max diff lines (default 200, 0 to disable).

Source code in packages/axm-git/src/axm_git/hooks/preflight.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@dataclass
class PreflightHook:
    """Report working-tree status and diff as a pre-hook.

    Designed for injection into protocol briefings via
    ``inject_result`` + ``inline: true``.

    *params*:
        ``path`` — project root (default ``"."``).
        ``diff_lines`` — max diff lines (default 200, 0 to disable).
    """

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

        Args:
            context: Session context dictionary.
            **params: Optional ``path`` and ``diff_lines``.

        Returns:
            HookResult with ``files``, ``diff``, ``file_count``, and ``clean``.
        """
        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        working_dir = _resolve_working_dir(params, context, param_key="path").resolve()
        max_diff_lines: int = int(params.get("diff_lines", 200))

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

        # Scope to package when inside a workspace (git root != working dir)
        rel = working_dir.resolve().relative_to(git_root.resolve())
        pathspec = ["--", str(rel)] if str(rel) != "." else []

        # git status --porcelain [-- rel_path]
        status = run_git(["status", "--porcelain", *pathspec], git_root)
        if status.returncode != 0:
            return HookResult.fail(f"git status failed: {status.stderr}")

        files: list[dict[str, str]] = []
        for line in status.stdout.splitlines():
            if len(line) < _MIN_STATUS_LINE_LEN:
                continue
            code = line[:2].strip()
            filepath = line[3:]
            files.append({"path": filepath, "status": code})

        # git diff -U2 [-- rel_path]
        diff_content = ""
        if max_diff_lines > 0:
            diff_result = run_git(["diff", "-U2", *pathspec], git_root)
            lines = diff_result.stdout.splitlines()
            if len(lines) > max_diff_lines:
                diff_content = "\n".join(lines[:max_diff_lines])
            else:
                diff_content = diff_result.stdout.strip()

        return HookResult.ok(
            files=files,
            diff=diff_content,
            file_count=len(files),
            clean=len(files) == 0,
        )
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

 required
**params Any

Optional path and diff_lines.

 {}

Returns:

Type Description
HookResult

HookResult with files, diff, file_count, and clean.
Source code in packages/axm-git/src/axm_git/hooks/preflight.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``path`` and ``diff_lines``.

    Returns:
        HookResult with ``files``, ``diff``, ``file_count``, and ``clean``.
    """
    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    working_dir = _resolve_working_dir(params, context, param_key="path").resolve()
    max_diff_lines: int = int(params.get("diff_lines", 200))

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

    # Scope to package when inside a workspace (git root != working dir)
    rel = working_dir.resolve().relative_to(git_root.resolve())
    pathspec = ["--", str(rel)] if str(rel) != "." else []

    # git status --porcelain [-- rel_path]
    status = run_git(["status", "--porcelain", *pathspec], git_root)
    if status.returncode != 0:
        return HookResult.fail(f"git status failed: {status.stderr}")

    files: list[dict[str, str]] = []
    for line in status.stdout.splitlines():
        if len(line) < _MIN_STATUS_LINE_LEN:
            continue
        code = line[:2].strip()
        filepath = line[3:]
        files.append({"path": filepath, "status": code})

    # git diff -U2 [-- rel_path]
    diff_content = ""
    if max_diff_lines > 0:
        diff_result = run_git(["diff", "-U2", *pathspec], git_root)
        lines = diff_result.stdout.splitlines()
        if len(lines) > max_diff_lines:
            diff_content = "\n".join(lines[:max_diff_lines])
        else:
            diff_content = diff_result.stdout.strip()

    return HookResult.ok(
        files=files,
        diff=diff_content,
        file_count=len(files),
        clean=len(files) == 0,
    )

PushHook dataclass

Push the current branch to origin -u.

Reads branch from context (or detects it from HEAD). Skips gracefully when the working directory is not a git repository.

Source code in packages/axm-git/src/axm_git/hooks/push.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@dataclass
class PushHook:
    """Push the current branch to ``origin -u``.

    Reads ``branch`` from *context* (or detects it from HEAD).
    Skips gracefully when the working directory is not a git repository.
    """

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

        Args:
            context: Session context dictionary.
            **params: Optional ``enabled`` (default ``True``).

        Returns:
            HookResult with ``pushed`` and ``branch`` in metadata.
        """
        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        working_dir = _resolve_working_dir(params, context)

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

        branch = context.get("branch")
        if not branch:
            head = run_git(["rev-parse", "--abbrev-ref", "HEAD"], git_root)
            if head.returncode != 0:
                return HookResult.fail(f"failed to detect branch: {head.stderr}")
            branch = head.stdout.strip()

        result = run_git(["push", "-u", "origin", branch], git_root)
        if result.returncode != 0:
            stderr = result.stderr.strip()
            if "Everything up-to-date" in (result.stderr + result.stdout):
                return HookResult.ok(pushed=True, branch=branch)
            return HookResult.fail(f"git push failed: {stderr}")

        return HookResult.ok(pushed=True, branch=branch)
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

 required
**params Any

Optional enabled (default True).

 {}

Returns:

Type Description
HookResult

HookResult with pushed and branch in metadata.
Source code in packages/axm-git/src/axm_git/hooks/push.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``enabled`` (default ``True``).

    Returns:
        HookResult with ``pushed`` and ``branch`` in metadata.
    """
    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    working_dir = _resolve_working_dir(params, context)

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

    branch = context.get("branch")
    if not branch:
        head = run_git(["rev-parse", "--abbrev-ref", "HEAD"], git_root)
        if head.returncode != 0:
            return HookResult.fail(f"failed to detect branch: {head.stderr}")
        branch = head.stdout.strip()

    result = run_git(["push", "-u", "origin", branch], git_root)
    if result.returncode != 0:
        stderr = result.stderr.strip()
        if "Everything up-to-date" in (result.stderr + result.stdout):
            return HookResult.ok(pushed=True, branch=branch)
        return HookResult.fail(f"git push failed: {stderr}")

    return HookResult.ok(pushed=True, branch=branch)

WorktreeAddHook dataclass

Create a worktree + branch for a ticket.

Reads ticket_id, ticket_title, ticket_labels, and repo_path from context. The worktree is placed under /tmp/axm-worktrees/<ticket_id>/.

Skips gracefully when the working directory is not a git repository or the worktree already exists.

Source code in packages/axm-git/src/axm_git/hooks/worktree_add.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class WorktreeAddHook:
    """Create a worktree + branch for a ticket.

    Reads ``ticket_id``, ``ticket_title``, ``ticket_labels``, and
    ``repo_path`` from *context*.  The worktree is placed under
    ``/tmp/axm-worktrees/<ticket_id>/``.

    Skips gracefully when the working directory is not a git repository
    or the worktree already exists.
    """

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

        Args:
            context: Session context dictionary.
            **params: Optional ``repo_path``, ``ticket_id``,
                ``ticket_title``, ``ticket_labels``, ``enabled``
                (default ``True``).  Params take precedence over context.

        Returns:
            HookResult with ``worktree_path`` and ``branch`` in metadata.
        """
        if not params.get("enabled", True):
            return HookResult.ok(skipped=True, reason="git disabled")

        repo_path = Path(params.get("repo_path", context.get("repo_path", ".")))

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

        ticket_id: str = (
            params["ticket_id"] if "ticket_id" in params else context["ticket_id"]
        )
        title: str = (
            params["ticket_title"]
            if "ticket_title" in params
            else context["ticket_title"]
        )
        labels: list[str] = params.get(
            "ticket_labels", context.get("ticket_labels", [])
        )

        branch = branch_name_from_ticket(ticket_id, title, labels)
        worktree_path = Path("/tmp/axm-worktrees") / ticket_id  # noqa: S108
        worktree_path.parent.mkdir(parents=True, exist_ok=True)

        if worktree_path.exists():
            return HookResult.ok(
                skipped=True,
                reason=f"worktree already exists: {worktree_path}",
            )

        result = run_git(
            ["worktree", "add", "-b", branch, str(worktree_path), "main"],
            repo_path,
        )
        if result.returncode != 0:
            return HookResult.fail(f"git worktree add failed: {result.stderr}")

        return HookResult.ok(
            worktree_path=str(worktree_path),
            branch=branch,
        )
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context dictionary.

 required
**params Any

Optional repo_path, ticket_id, ticket_title, ticket_labels, enabled (default True). Params take precedence over context.

 {}

Returns:

Type Description
HookResult

HookResult with worktree_path and branch in metadata.
Source code in packages/axm-git/src/axm_git/hooks/worktree_add.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context dictionary.
        **params: Optional ``repo_path``, ``ticket_id``,
            ``ticket_title``, ``ticket_labels``, ``enabled``
            (default ``True``).  Params take precedence over context.

    Returns:
        HookResult with ``worktree_path`` and ``branch`` in metadata.
    """
    if not params.get("enabled", True):
        return HookResult.ok(skipped=True, reason="git disabled")

    repo_path = Path(params.get("repo_path", context.get("repo_path", ".")))

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

    ticket_id: str = (
        params["ticket_id"] if "ticket_id" in params else context["ticket_id"]
    )
    title: str = (
        params["ticket_title"]
        if "ticket_title" in params
        else context["ticket_title"]
    )
    labels: list[str] = params.get(
        "ticket_labels", context.get("ticket_labels", [])
    )

    branch = branch_name_from_ticket(ticket_id, title, labels)
    worktree_path = Path("/tmp/axm-worktrees") / ticket_id  # noqa: S108
    worktree_path.parent.mkdir(parents=True, exist_ok=True)

    if worktree_path.exists():
        return HookResult.ok(
            skipped=True,
            reason=f"worktree already exists: {worktree_path}",
        )

    result = run_git(
        ["worktree", "add", "-b", branch, str(worktree_path), "main"],
        repo_path,
    )
    if result.returncode != 0:
        return HookResult.fail(f"git worktree add failed: {result.stderr}")

    return HookResult.ok(
        worktree_path=str(worktree_path),
        branch=branch,
    )

WorktreeRemoveHook dataclass

Remove a worktree after merge.

Reads worktree_path and repo_path from context. Uses git worktree remove --force to handle dirty worktrees. Skips gracefully when the path doesn't exist or isn't a git repo.

Source code in packages/axm-git/src/axm_git/hooks/worktree_remove.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass
class WorktreeRemoveHook:
    """Remove a worktree after merge.

    Reads ``worktree_path`` and ``repo_path`` from *context*.
    Uses ``git worktree remove --force`` to handle dirty worktrees.
    Skips gracefully when the path doesn't exist or isn't a git repo.
    """

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

        Args:
            context: Session context (must contain ``repo_path``
                and ``worktree_path``).
            **params: Optional ``enabled`` (default ``True``).

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

        repo_path = Path(context.get("repo_path", "."))

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

        worktree_path = _resolve_working_dir({}, context)

        if not worktree_path.exists():
            return HookResult.ok(
                skipped=True,
                reason=f"worktree path does not exist: {worktree_path}",
            )

        result = run_git(
            ["worktree", "remove", str(worktree_path), "--force"],
            repo_path,
        )
        if result.returncode != 0:
            return HookResult.fail(
                f"git worktree remove failed: {result.stderr}",
            )

        return HookResult.ok()
execute(context, **params)

Execute the hook action.

Parameters:

Name Type Description Default
context dict[str, Any]

Session context (must contain repo_path and worktree_path).

 required
**params Any

Optional enabled (default True).

 {}

Returns:

Type Description
HookResult

HookResult on success.
Source code in packages/axm-git/src/axm_git/hooks/worktree_remove.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def execute(self, context: dict[str, Any], **params: Any) -> HookResult:
    """Execute the hook action.

    Args:
        context: Session context (must contain ``repo_path``
            and ``worktree_path``).
        **params: Optional ``enabled`` (default ``True``).

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

    repo_path = Path(context.get("repo_path", "."))

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

    worktree_path = _resolve_working_dir({}, context)

    if not worktree_path.exists():
        return HookResult.ok(
            skipped=True,
            reason=f"worktree path does not exist: {worktree_path}",
        )

    result = run_git(
        ["worktree", "remove", str(worktree_path), "--force"],
        repo_path,
    )
    if result.returncode != 0:
        return HookResult.fail(
            f"git worktree remove failed: {result.stderr}",
        )

    return HookResult.ok()