refactor(autoland): handle Ctrl+C once at the run boundary

src/stack_pr/autoland/commands.py

@@ -237,29 +237,46 @@ def _dispose_worktree(
         worktree.announce_preserved()
 
 
-def _install_signal_handler(
+def _land(
     ctx: LandingContext,
+    common: cli.CommonArgs,
+    opts: AutolandOptions,
     checkpointer: AutolandCheckpointer,
     worktree: Worktree | None,
-    opts: AutolandOptions,
 ) -> None:
-    # Installed as landing starts: from here on, the handler and _finish
-    # decide whether the worktree is kept.
+    """Execute the plan, report how the run ended, and exit if it didn't succeed.
+
+    This is the one place a run is interrupted. Ctrl+C anywhere in the plan's
+    execution (a wait's sleep, a gh call, a confirm prompt) raises
+    KeyboardInterrupt out of it, and so does a takeover by another autoland of
+    the branch, which sends this process SIGINT (``_stop_running_autoland``).
+    The run is then checkpointed at the step it was in, so ``--resume`` picks
+    that step up again, and exits 130; the callers' ``finally`` blocks release
+    the lock on the way out.
+    """
+    # From here on, the run's outcome decides whether the worktree is kept.
     if worktree is not None:
         worktree.landing = True
-
-    def handler(_sig: int, _frame: object) -> None:
-        ctx.aborted = True
+    # A process started with SIGINT ignored (a background job of a
+    # non-interactive shell) keeps ignoring it; restore Python's default so
+    # that a takeover's SIGINT still stops this run.
+    signal.signal(signal.SIGINT, signal.default_int_handler)
+    try:
+        success = engine.execute_plan(
+            ctx, common, opts, checkpointer, services=Services.production()
+        )
+    except KeyboardInterrupt:
         ctx.abort_reason = "User interrupted (Ctrl+C)"
+        # Atomic (a temporary file, then a rename): a second Ctrl+C here leaves
+        # the checkpoint taken as the step started, which resumes the same way.
         checkpointer.save(ctx)
         runtime.console.print(
             "\n[red bold]Interrupted! State saved. Resume with --resume.[/red bold]\n"
         )
         print_status(ctx)
         _dispose_worktree(worktree, opts, succeeded=False)
         sys.exit(130)
-
-    signal.signal(signal.SIGINT, handler)
+    _finish(ctx, checkpointer, worktree, opts, success=success)
 
 
 def _finish(
@@ -422,16 +439,7 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None:
             return
 
         runtime.console.print(f"[dim]State file: {checkpointer.path}[/dim]\n")
-        _install_signal_handler(ctx, checkpointer, worktree, opts)
-        _finish(
-            ctx,
-            checkpointer,
-            worktree,
-            opts,
-            success=engine.execute_plan(
-                ctx, common, opts, checkpointer, services=Services.production()
-            ),
-        )
+        _land(ctx, common, opts, checkpointer, worktree)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()
@@ -483,7 +491,6 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None:
 
         runtime.console.print("[dim]Refreshing PR state from GitHub...[/dim]")
         discovery.enrich_stack(ctx.stack)
-        ctx.aborted = False
         ctx.abort_reason = ""
 
         if ctx.current_step >= len(ctx.plan):
@@ -497,16 +504,7 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None:
         print_status(ctx)
         engine.print_execution_preview(ctx, opts)
         runtime.console.print(f"[dim]State file: {sf_path}[/dim]\n")
-        _install_signal_handler(ctx, checkpointer, worktree, opts)
-        _finish(
-            ctx,
-            checkpointer,
-            worktree,
-            opts,
-            success=engine.execute_plan(
-                ctx, common, opts, checkpointer, services=Services.production()
-            ),
-        )
+        _land(ctx, common, opts, checkpointer, worktree)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()
@@ -721,16 +719,7 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) ->
             path=state_path, branch=branch, base=common.target, plan_file=plan_file
         )
         runtime.console.print(f"[dim]State file: {checkpointer.path}[/dim]\n")
-        _install_signal_handler(ctx, checkpointer, worktree, opts)
-        _finish(
-            ctx,
-            checkpointer,
-            worktree,
-            opts,
-            success=engine.execute_plan(
-                ctx, common, opts, checkpointer, services=Services.production()
-            ),
-        )
+        _land(ctx, common, opts, checkpointer, worktree)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()

src/stack_pr/autoland/display.py

@@ -74,8 +74,8 @@ def _pick_glyphs() -> _Glyphs:
     """The richest glyph set stdout can actually encode.
 
     Writing an un-encodable character raises UnicodeEncodeError, which would
-    take down the whole landing run — print_status is called from the SIGINT
-    handler and from _finish, so even a successful autoland would die on the
+    take down the whole landing run — print_status is called when a run is
+    interrupted and from _finish, so even a successful autoland would die on the
     way out, leaving its checkpoint and worktree behind.
     """
     encoding = getattr(sys.stdout, "encoding", None) or "ascii"
@@ -265,7 +265,7 @@ def render_status_rich(ctx: LandingContext) -> Group:  # pragma: no cover
             # expand=False so the pad doesn't fill the rest of the line with
             # blanks; a wrapped detail still stays under the indent.
             parts.append(Padding(detail, (0, 0, 0, indent), expand=False))
-    if ctx.aborted:
+    if ctx.abort_reason:
         parts += [Text(), Text(f"ABORTED: {ctx.abort_reason}", style="red bold")]
     return Group(*parts)
 
@@ -281,7 +281,7 @@ def render_status_plain(ctx: LandingContext) -> str:
         lines.append("".join(text for text, _ in _header_segments(row, widths)))
         if row.detail:
             lines.append(f"{indent}{GLYPHS.detail} {row.detail}")
-    if ctx.aborted:
+    if ctx.abort_reason:
         lines += ["", f"ABORTED: {ctx.abort_reason}"]
     return "\n".join(lines)
 

src/stack_pr/autoland/engine.py

@@ -117,6 +117,5 @@ def _abort(
     ctx: LandingContext, checkpointer: AutolandCheckpointer, reason: str
 ) -> bool:
     ctx.abort_reason = reason
-    ctx.aborted = True
     checkpointer.save(ctx)
     return False

src/stack_pr/autoland/model.py

@@ -68,15 +68,10 @@ class LandingContext:
     plan: list[PlanStep] = field(default_factory=list)
     current_step: int = 0
     current_index: int = 0  # index into stack for the active land step
-    aborted: bool = False
+    # Why the run stopped, or "" while it hasn't. Saved with the checkpoint.
     abort_reason: str = ""
     last_landed_sha: str = ""  # merge commit of the last landed PR
 
-    def abort_requested(self) -> bool:
-        # A method, not a plain attribute read, so that mypy does not carry
-        # narrowing of `aborted` across calls (or the SIGINT handler) that set it.
-        return self.aborted
-
 
 def _deserialize_entry(data: dict) -> StackEntry:
     return StackEntry(

src/stack_pr/autoland/native_stack.py

@@ -83,7 +83,6 @@ def _await_native_stack_merge(
     uuid: str | None,
     *,
     opts: AutolandOptions,
-    ctx: LandingContext,
 ) -> str:
     """Wait for a requested stack merge to finish.
 
@@ -94,8 +93,6 @@ def _await_native_stack_merge(
     top = entries[-1]
     awake_elapsed = 0.0
     while True:
-        if ctx.aborted:
-            return "aborted"
         if awake_elapsed > opts.merge_timeout:
             return "timed out waiting for the stack to merge"
 
@@ -184,14 +181,16 @@ def land_as_native_stack(
     failure = ""
     abort_reason = ""
     for entry in entries:
-        if not waits.wait_for_approval(entry, opts=opts, ctx=ctx):
-            abort_reason = f"PR #{entry.pr_number} approval wait was aborted"
+        if not waits.wait_for_approval(entry, opts=opts):
+            abort_reason = (
+                f"PR #{entry.pr_number} was closed while waiting for approval"
+            )
             break
     else:
         for entry in entries:
             if entry.state == PRState.MERGED:
                 continue
-            if not waits.wait_for_checks(entry, opts=opts, ctx=ctx):
+            if not waits.wait_for_checks(entry, opts=opts):
                 abort_reason = f"PR #{entry.pr_number} checks failed after retries"
                 break
 
@@ -202,7 +201,7 @@ def land_as_native_stack(
     if (
         not abort_reason
         and bottom is not None
-        and not waits.wait_for_mergeable(bottom, opts=opts, ctx=ctx).ready
+        and not waits.wait_for_mergeable(bottom, opts=opts).ready
     ):
         abort_reason = f"PR #{bottom.pr_number} failed to merge"
 
@@ -246,9 +245,7 @@ def land_as_native_stack(
                 if "HTTP 409" not in str(e):
                     failure = f"stack merge request failed: {e}"
         if not failure:
-            failure = _await_native_stack_merge(entries, uuid, opts=opts, ctx=ctx)
-            if failure == "aborted":
-                abort_reason = "Stack merge was aborted"
+            failure = _await_native_stack_merge(entries, uuid, opts=opts)
 
     landed = all(e.state == PRState.MERGED for e in entries)
     if not landed and dissolvable:

src/stack_pr/autoland/services.py

@@ -43,21 +43,18 @@ def escape(self, text: str) -> str:
 class Waits(Protocol):
     """The long-running waits of landing a PR or checking a workflow.
 
-    Each returns whether it succeeded; False also when the run was aborted.
-    The waits record their progress on the entry or step they are given.
+    Each returns whether it succeeded, and records its progress on the entry
+    or step it is given. Ctrl+C while one is waiting raises KeyboardInterrupt
+    out of it, which the run handles at its boundary (``commands._land``).
     """
 
     def wait_for_approval(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
+        self, entry: StackEntry, *, opts: AutolandOptions
     ) -> bool: ...
 
-    def wait_for_checks(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool: ...
+    def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: ...
 
-    def enqueue_and_wait(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool: ...
+    def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool: ...
 
     def wait_for_workflow(
         self,
@@ -136,20 +133,14 @@ def escape(self, text: str) -> str:
 
 
 class _ModuleWaits:
-    def wait_for_approval(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
-        return waits.wait_for_approval(entry, opts=opts, ctx=ctx)
+    def wait_for_approval(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
+        return waits.wait_for_approval(entry, opts=opts)
 
-    def wait_for_checks(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
-        return waits.wait_for_checks(entry, opts=opts, ctx=ctx)
+    def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
+        return waits.wait_for_checks(entry, opts=opts)
 
-    def enqueue_and_wait(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
-        return waits.enqueue_and_wait(entry, opts=opts, ctx=ctx)
+    def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
+        return waits.enqueue_and_wait(entry, opts=opts)
 
     def wait_for_workflow(
         self,

src/stack_pr/autoland/state.py

@@ -72,7 +72,6 @@ def load(cls, path: Path) -> tuple[AutolandCheckpointer, LandingContext]:
             stack=[_deserialize_entry(e) for e in data["stack"]],
             plan=[_deserialize_step(s) for s in data["plan"]],
             current_step=data.get("current_step", 0),
-            aborted=bool(data.get("abort_reason")),
             abort_reason=data.get("abort_reason", ""),
             last_landed_sha=data.get("last_landed_sha", ""),
         )

src/stack_pr/autoland/steps/land.py

@@ -168,20 +168,18 @@ def execute(self, sc: StepContext) -> StepResult:
             )
 
         ui.print(sc.banner(f"Landing PR #{entry.pr_number} — {entry.title}"))
-        waits, opts, run = sc.services.waits, sc.opts, sc.run
+        waits, opts = sc.services.waits, sc.opts
         # Each wait may find the PR merged by then (e.g. by hand), which ends
         # the landing early.
-        if not waits.wait_for_approval(entry, opts=opts, ctx=run):
-            return StepResult.abort(f"PR #{entry.pr_number} approval wait was aborted")
-        if not entry.is_merged() and not waits.wait_for_checks(
-            entry, opts=opts, ctx=run
-        ):
+        if not waits.wait_for_approval(entry, opts=opts):
+            return StepResult.abort(
+                f"PR #{entry.pr_number} was closed while waiting for approval"
+            )
+        if not entry.is_merged() and not waits.wait_for_checks(entry, opts=opts):
             return StepResult.abort(
                 f"PR #{entry.pr_number} checks failed after retries"
             )
-        if not entry.is_merged() and not waits.enqueue_and_wait(
-            entry, opts=opts, ctx=run
-        ):
+        if not entry.is_merged() and not waits.enqueue_and_wait(entry, opts=opts):
             return StepResult.abort(f"PR #{entry.pr_number} failed to merge")
         landed_sha = repo.landed_sha(sc.common, entry.pr_number)