Skip to content

Await merge

await_merge

Await-merge hook action.

Polls a GitHub PR until it reaches the MERGED state.

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
Python
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
@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, object], **params: object) -> 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 = resolve_pr_ref(params, context)
        if not pr_ref:
            return HookResult.fail("no pr_number or pr_url in params or context")

        timeout = int(cast("int", params.get("timeout", _DEFAULT_TIMEOUT)))
        interval = int(cast("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, object]

Session context dictionary (must contain pr_number or pr_url).

 required
**params object

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
Python
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
def execute(self, context: dict[str, object], **params: object) -> 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 = resolve_pr_ref(params, context)
    if not pr_ref:
        return HookResult.fail("no pr_number or pr_url in params or context")

    timeout = int(cast("int", params.get("timeout", _DEFAULT_TIMEOUT)))
    interval = int(cast("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")

resolve_pr_ref(params, context)

Resolve a PR reference from params or context.

Checks pr_number then pr_url in params first, falling back to context. Returns None when no reference is found.

Source code in packages/axm-git/src/axm_git/hooks/await_merge.py
Python
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def resolve_pr_ref(
    params: dict[str, object],
    context: dict[str, object],
) -> object | None:
    """Resolve a PR reference from params or context.

    Checks ``pr_number`` then ``pr_url`` in *params* first,
    falling back to *context*.  Returns ``None`` when no
    reference is found.
    """
    ref = (
        params.get("pr_number")
        or params.get("pr_url")
        or context.get("pr_number")
        or context.get("pr_url")
    )
    return ref if ref else None