diff --git a/src/stack_pr/autoland/commands.py b/src/stack_pr/autoland/commands.py
index 9df5021..761d342 100644
--- a/src/stack_pr/autoland/commands.py
+++ b/src/stack_pr/autoland/commands.py
@@ -237,20 +237,38 @@ 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"
@@ -258,8 +276,7 @@ def handler(_sig: int, _frame: object) -> None:
         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()
diff --git a/src/stack_pr/autoland/display.py b/src/stack_pr/autoland/display.py
index ecff426..84a3e1d 100644
--- a/src/stack_pr/autoland/display.py
+++ b/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)
 
diff --git a/src/stack_pr/autoland/engine.py b/src/stack_pr/autoland/engine.py
index cb5f56b..071abf8 100644
--- a/src/stack_pr/autoland/engine.py
+++ b/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
diff --git a/src/stack_pr/autoland/model.py b/src/stack_pr/autoland/model.py
index 44add89..1d1ed18 100644
--- a/src/stack_pr/autoland/model.py
+++ b/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(
diff --git a/src/stack_pr/autoland/native_stack.py b/src/stack_pr/autoland/native_stack.py
index 8a02dd2..64f1ea3 100644
--- a/src/stack_pr/autoland/native_stack.py
+++ b/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:
diff --git a/src/stack_pr/autoland/services.py b/src/stack_pr/autoland/services.py
index 12d3eaa..fbd48b3 100644
--- a/src/stack_pr/autoland/services.py
+++ b/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,
diff --git a/src/stack_pr/autoland/state.py b/src/stack_pr/autoland/state.py
index f94d2af..833b321 100644
--- a/src/stack_pr/autoland/state.py
+++ b/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", ""),
         )
diff --git a/src/stack_pr/autoland/steps/land.py b/src/stack_pr/autoland/steps/land.py
index 13a3c4f..3b0738a 100644
--- a/src/stack_pr/autoland/steps/land.py
+++ b/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)
 
diff --git a/src/stack_pr/autoland/waits.py b/src/stack_pr/autoland/waits.py
index 79521a9..609fdf1 100644
--- a/src/stack_pr/autoland/waits.py
+++ b/src/stack_pr/autoland/waits.py
@@ -185,8 +185,6 @@ def wait_for_workflow(
     ancestry = _Ancestry(common)
     awake_elapsed = 0.0
     while True:
-        if ctx.aborted:
-            return False
         if awake_elapsed > opts.workflow_timeout:
             step.state = "failed"
             step.error_message = (
@@ -274,10 +272,9 @@ def _refresh_review(entry: StackEntry) -> bool:
     return entry.is_approved
 
 
-def wait_for_approval(
-    entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-) -> bool:
-    """Wait until the PR has required approvals. Returns False if aborted."""
+def wait_for_approval(entry: StackEntry, *, opts: AutolandOptions) -> bool:
+    """Wait until the PR has required approvals (or merged). Returns False if
+    the PR was closed."""
     # An unknown state (a failed read) is settled by the checks wait after this.
     pr_state = _poll_read(gh.github.pr_state, entry.pr_number)
     if pr_state == "MERGED":
@@ -300,8 +297,6 @@ def wait_for_approval(
     )
 
     while True:
-        if ctx.aborted:
-            return False
         runtime.console.print(
             f"[dim]PR #{entry.pr_number}: waiting for approval — "
             f"polling in {opts.poll_interval}s[/dim]"
@@ -333,16 +328,11 @@ def wait_for_approval(
             )
 
 
-def wait_for_checks(
-    entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-) -> bool:
+def wait_for_checks(entry: StackEntry, *, opts: AutolandOptions) -> bool:
     """Wait for required checks to pass. Returns True on success."""
     entry.state = PRState.WAITING_FOR_CHECKS
 
     while True:
-        if ctx.aborted:
-            return False
-
         pr_state = _poll_read(gh.github.pr_state, entry.pr_number)
         if pr_state == "MERGED":
             entry.state = PRState.MERGED
@@ -398,14 +388,9 @@ class MergeableResult:
     error: str = ""
 
 
-def wait_for_mergeable(
-    entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-) -> MergeableResult:
+def wait_for_mergeable(entry: StackEntry, *, opts: AutolandOptions) -> MergeableResult:
     """Wait until GitHub reports the PR as mergeable."""
     while True:
-        if ctx.aborted:
-            return MergeableResult(error="aborted")
-
         data = _poll_read(gh.github.merge_state, entry.pr_number)
         if data is None:
             runtime.resilient_sleep(opts.poll_interval)
@@ -460,25 +445,18 @@ def wait_for_mergeable(
         runtime.resilient_sleep(opts.poll_interval)
 
 
-def _rewait_after_retry(
-    entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-) -> bool:
+def _rewait_after_retry(entry: StackEntry, *, opts: AutolandOptions) -> bool:
     """Re-verify approval and checks before a queue retry."""
-    if not wait_for_approval(entry, opts=opts, ctx=ctx):
+    if not wait_for_approval(entry, opts=opts):
         return False
     entry.check_retries = 0
-    return wait_for_checks(entry, opts=opts, ctx=ctx)
+    return wait_for_checks(entry, opts=opts)
 
 
-def enqueue_and_wait(
-    entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-) -> bool:
+def enqueue_and_wait(entry: StackEntry, *, opts: AutolandOptions) -> bool:
     """Add the PR to the merge queue and wait for it to merge."""
     while True:
-        if ctx.aborted:
-            return False
-
-        mergeable_result = wait_for_mergeable(entry, opts=opts, ctx=ctx)
+        mergeable_result = wait_for_mergeable(entry, opts=opts)
         if not mergeable_result.ready:
             return False
         if mergeable_result.already_merged:
@@ -507,15 +485,13 @@ def enqueue_and_wait(
                 )
                 return False
             entry.queue_retries += 1
-            if not _rewait_after_retry(entry, opts=opts, ctx=ctx):
+            if not _rewait_after_retry(entry, opts=opts):
                 return False
             continue
 
         entry.error_message = "Waiting in merge queue..."
         awake_elapsed = 0.0
         while True:
-            if ctx.abort_requested():
-                return False
             if awake_elapsed > opts.merge_timeout:
                 entry.state = PRState.FAILED
                 entry.error_message = "Timed out waiting for merge queue"
@@ -549,7 +525,7 @@ def enqueue_and_wait(
                     )
                     return False
                 entry.queue_retries += 1
-                if not _rewait_after_retry(entry, opts=opts, ctx=ctx):
+                if not _rewait_after_retry(entry, opts=opts):
                     return False
                 break  # re-enqueue in outer loop
 
diff --git a/tests/autoland_scenarios.py b/tests/autoland_scenarios.py
index 03c2c0a..266db2b 100644
--- a/tests/autoland_scenarios.py
+++ b/tests/autoland_scenarios.py
@@ -22,7 +22,7 @@
 - stack-pr itself: ``cli.get_stack`` returns the scenario's stack,
   ``cli.command_submit`` (the resubmit after a merge) does nothing, and
   ``git.get_current_branch_name`` is pinned;
-- the SIGINT handler isn't installed, and any real subprocess fails the test.
+- any real subprocess fails the test.
 """
 
 from __future__ import annotations
@@ -435,7 +435,6 @@ def run_shell_command(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess:
     mocker.patch.object(cli, "get_stack", return_value=raw_stack)
     mocker.patch.object(cli, "command_submit")
     mocker.patch("stack_pr.git.get_current_branch_name", return_value=BRANCH)
-    mocker.patch.object(commands.signal, "signal")
 
 
 def _autoland(argv: list[str], config: configparser.ConfigParser) -> int | None:
diff --git a/tests/test_autoland.py b/tests/test_autoland.py
index 6620ece..15f3eda 100644
--- a/tests/test_autoland.py
+++ b/tests/test_autoland.py
@@ -354,11 +354,14 @@ def test_wait_for_workflow_falls_back_to_github_compare(mocker) -> None:  # noqa
     contains.assert_called_once_with("mergesha", "queuesha")
 
 
+class _StopPollingError(Exception):
+    """Raised from a wait's sleep to end the wait after one poll."""
+
+
 def test_wait_for_workflow_keeps_waiting_when_ancestry_unknown(mocker) -> None:  # noqa: ANN001
     # Neither git nor GitHub can answer: warn and keep waiting rather than
     # treating "unknown" as "the deploy included our code".
     def _runs(*_a, **_k) -> list:  # noqa: ANN002, ANN003
-        ctx.aborted = True  # stop after one poll
         return [{"headSha": "mystery", "status": "completed", "conclusion": "success"}]
 
     mocker.patch.object(gh.github, "workflow_runs", side_effect=_runs)
@@ -366,12 +369,12 @@ def _runs(*_a, **_k) -> list:  # noqa: ANN002, ANN003
     mocker.patch.object(runtime, "run", return_value=unknown)  # the fetch
     mocker.patch("stack_pr.git.run_shell_command", return_value=unknown)
     mocker.patch.object(gh.github, "contains", return_value=None)
-    mocker.patch.object(runtime, "resilient_sleep", return_value=0.0)
+    mocker.patch.object(runtime, "resilient_sleep", side_effect=_StopPollingError)
     step = WorkflowStep(workflow="deploy.yaml")
     ctx = LandingContext(last_landed_sha="mergesha")
-    assert not waits.wait_for_workflow(
-        step, opts=_opts(), common=common_args(), ctx=ctx
-    )
+    # Still waiting when the first poll's sleep comes around.
+    with pytest.raises(_StopPollingError):
+        waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx)
     assert step.state != "succeeded"
 
 
@@ -427,24 +430,21 @@ def test_ancestry_caches_verdicts(mocker) -> None:  # noqa: ANN001
 
 def test_wait_for_workflow_ignores_failed_and_incomplete(mocker) -> None:  # noqa: ANN001
     # A failed run and a still-running run on our SHA must not satisfy the
-    # checkpoint; abort so the poll loop terminates for the test.
-    calls = {"n": 0}
-
-    def _runs(*_a, **_k) -> list:  # noqa: ANN002, ANN003
-        calls["n"] += 1
-        ctx.aborted = True  # stop after one poll
-        return [
+    # checkpoint: the wait is still going when its first poll's sleep comes.
+    mocker.patch.object(
+        gh.github,
+        "workflow_runs",
+        return_value=[
             {"headSha": "mergesha", "status": "completed", "conclusion": "failure"},
             {"headSha": "mergesha", "status": "in_progress", "conclusion": None},
-        ]
-
-    mocker.patch.object(gh.github, "workflow_runs", side_effect=_runs)
-    mocker.patch.object(runtime, "resilient_sleep", return_value=0.0)
+        ],
+    )
+    mocker.patch.object(runtime, "resilient_sleep", side_effect=_StopPollingError)
     step = WorkflowStep(workflow="deploy.yaml")
     ctx = LandingContext(last_landed_sha="mergesha")
-    assert not waits.wait_for_workflow(
-        step, opts=_opts(), common=common_args(), ctx=ctx
-    )
+    with pytest.raises(_StopPollingError):
+        waits.wait_for_workflow(step, opts=_opts(), common=common_args(), ctx=ctx)
+    assert step.state != "succeeded"
 
 
 # --- plan parsing --------------------------------------------------------
@@ -940,7 +940,6 @@ def test_render_status_plain_headline_counts_completed_steps() -> None:
 
 def test_render_status_plain_reports_an_abort() -> None:
     ctx = _partly_landed_ctx()
-    ctx.aborted = True
     ctx.abort_reason = "PR #103 checks failed"
     assert "ABORTED: PR #103 checks failed" in display.render_status_plain(ctx)
 
@@ -997,7 +996,7 @@ def test_unicode_glyphs_are_used_on_a_utf8_stdout(mocker) -> None:  # noqa: ANN0
 
 def test_plain_status_is_encodable_on_a_legacy_code_page(mocker) -> None:  # noqa: ANN001
     # Printing an un-encodable glyph raises UnicodeEncodeError, and print_status
-    # runs from the SIGINT handler and from _finish — so even a successful run
+    # runs when a run is interrupted and from _finish — so even a successful run
     # would die on the way out, stranding its checkpoint and worktree.
     mocker.patch.object(display, "GLYPHS", _ASCII_GLYPHS)
     display.render_status_plain(_partly_landed_ctx()).encode("cp1252")
@@ -1055,7 +1054,6 @@ def test_rich_status_truncates_titles_but_never_the_failure_reason() -> None:
         "integration-tests, docs, and a few more besides"
     )
     ctx = LandingContext(stack=stack, plan=[LandStep(entry_index=0, pr_number=101)])
-    ctx.aborted = True
     ctx.abort_reason = (
         "Rebase failed after merging #101: Command failed: git rebase --onto "
         "origin/main abc123 def456 -- error: could not apply def456"
@@ -1189,14 +1187,12 @@ def test_state_round_trip_keeps_plan_file(tmp_path) -> None:  # noqa: ANN001
 
 def test_state_round_trip_keeps_abort_reason(tmp_path) -> None:  # noqa: ANN001
     ctx = LandingContext(stack=_stack(1), plan=generate_default_plan(_stack(1)))
-    ctx.aborted = True
     ctx.abort_reason = "CI failed on #0"
 
     sf = tmp_path / "state.json"
     AutolandCheckpointer(path=sf, branch="feat", base="main").save(ctx)
 
     _, loaded = AutolandCheckpointer.load(sf)
-    assert loaded.aborted is True
     assert loaded.abort_reason == "CI failed on #0"
 
 
@@ -1731,7 +1727,6 @@ def _patch_replan_io(mocker, answer: str):  # noqa: ANN001, ANN202
         "stack_pr.autoland.gh.github.pr_state",
         side_effect=lambda pr: "MERGED" if pr in (101, 102) else "OPEN",
     )
-    mocker.patch("stack_pr.autoland.commands.signal.signal")
     return mocker.patch("stack_pr.autoland.engine.execute_plan", return_value=True)
 
 
@@ -1797,7 +1792,6 @@ def _patch_resume_io(mocker, *, success: bool = True):  # noqa: ANN001, ANN202
     mocker.patch("stack_pr.git.get_current_branch_name", return_value="feat")
     mocker.patch("stack_pr.cli.deduce_base", side_effect=lambda c: c)
     mocker.patch("stack_pr.autoland.discovery.enrich_stack")
-    mocker.patch("stack_pr.autoland.commands.signal.signal")
     return mocker.patch("stack_pr.autoland.engine.execute_plan", return_value=success)
 
 
@@ -1816,7 +1810,6 @@ def test_resume_continues_from_the_checkpointed_step(tmp_path, mocker) -> None:
     assert ctx.current_step == 1
     assert [e.pr_number for e in ctx.stack] == [0, 1]
     # The earlier failure no longer stands: this run gets a fresh attempt.
-    assert not ctx.aborted
     assert ctx.abort_reason == ""
     # A run that finishes leaves no checkpoint behind.
     assert not sf.exists()
@@ -1962,15 +1955,17 @@ def test_wait_for_approval_by_review_decision(
 ) -> None:
     mocker.patch.object(gh.github, "pr_state", return_value="OPEN")
     mocker.patch.object(gh.github, "review_decision", return_value=decision)
-    ctx = LandingContext()
-    # Abort the wait on its first poll, so a PR that isn't approved returns.
     mocker.patch(
-        "stack_pr.autoland.runtime.resilient_sleep",
-        side_effect=lambda _s: setattr(ctx, "aborted", True),
+        "stack_pr.autoland.runtime.resilient_sleep", side_effect=_StopPollingError
     )
 
     entry = _pinned_stack([101])[0]
-    assert waits.wait_for_approval(entry, opts=_opts(), ctx=ctx) is approved
+    if approved:
+        assert waits.wait_for_approval(entry, opts=_opts())
+    else:
+        # A PR that isn't approved is still waited on at the first poll's sleep.
+        with pytest.raises(_StopPollingError):
+            waits.wait_for_approval(entry, opts=_opts())
 
 
 # --- merging a run of land steps as a GitHub stack -------------------------
@@ -2287,7 +2282,7 @@ def test_wait_for_checks_polls_again_after_a_failed_state_read(
     _gh_replies(mocker, view=[bad_reply, _OPEN], checks=[_PASSING])
     entry = _pinned_stack([101])[0]
 
-    assert waits.wait_for_checks(entry, opts=_opts(), ctx=LandingContext())
+    assert waits.wait_for_checks(entry, opts=_opts())
 
 
 @pytest.mark.usefixtures("autoland_console")
@@ -2295,7 +2290,7 @@ def test_wait_for_checks_polls_again_after_a_malformed_checks_read(mocker) -> No
     _gh_replies(mocker, view=[_OPEN], checks=['{"oops": 1}', "[1]", _PASSING])
     entry = _pinned_stack([101])[0]
 
-    assert waits.wait_for_checks(entry, opts=_opts(), ctx=LandingContext())
+    assert waits.wait_for_checks(entry, opts=_opts())
 
 
 @pytest.mark.usefixtures("autoland_console")
@@ -2312,7 +2307,7 @@ def test_wait_for_approval_polls_again_after_a_failed_state_read(mocker) -> None
     )
     entry = _pinned_stack([101])[0]
 
-    assert waits.wait_for_approval(entry, opts=_opts(), ctx=LandingContext())
+    assert waits.wait_for_approval(entry, opts=_opts())
     assert entry.state == PRState.MERGED
 
 
@@ -2322,7 +2317,7 @@ def test_wait_for_mergeable_polls_again_after_a_malformed_read(mocker) -> None:
     _gh_replies(mocker, view=["", RuntimeError("timed out"), clean])
     entry = _pinned_stack([101])[0]
 
-    result = waits.wait_for_mergeable(entry, opts=_opts(), ctx=LandingContext())
+    result = waits.wait_for_mergeable(entry, opts=_opts())
 
     assert result.ready
 
@@ -2345,7 +2340,7 @@ def test_merge_queue_wait_polls_again_after_a_failed_read(mocker) -> None:  # no
     )
     entry = _pinned_stack([101])[0]
 
-    assert waits.enqueue_and_wait(entry, opts=_opts(), ctx=LandingContext())
+    assert waits.enqueue_and_wait(entry, opts=_opts())
     assert entry.state == PRState.MERGED
 
 
@@ -2355,7 +2350,7 @@ def test_merge_queue_wait_still_times_out_while_reads_fail(mocker) -> None:  # n
     entry = _pinned_stack([101])[0]
 
     opts = _opts(poll_interval=60, merge_timeout=300)
-    assert not waits.enqueue_and_wait(entry, opts=opts, ctx=LandingContext())
+    assert not waits.enqueue_and_wait(entry, opts=opts)
     assert entry.error_message == "Timed out waiting for merge queue"
 
 
diff --git a/tests/test_autoland_golden.py b/tests/test_autoland_golden.py
index fc9e787..caff3ee 100644
--- a/tests/test_autoland_golden.py
+++ b/tests/test_autoland_golden.py
@@ -64,7 +64,6 @@ def _resume_checkpoint(
         stack=stack,
         plan=plan,
         current_step=4,
-        aborted=True,
         abort_reason="Workflow deploy.yaml failed or timed out",
         last_landed_sha=sha("merge-102"),
     )
@@ -104,7 +103,6 @@ def _replan_checkpoint(
         plan=plan,
         current_step=7,
         current_index=2,
-        aborted=True,
         abort_reason="User interrupted (Ctrl+C)",
         last_landed_sha=sha("merge-102"),
     )
diff --git a/tests/test_autoland_interrupt.py b/tests/test_autoland_interrupt.py
new file mode 100644
index 0000000..3f6d7d5
--- /dev/null
+++ b/tests/test_autoland_interrupt.py
@@ -0,0 +1,257 @@
+"""Interrupting a landing run: Ctrl+C, or a takeover's SIGINT.
+
+Either way the run sees KeyboardInterrupt wherever it happens to be (a wait,
+a confirm prompt), and the run boundary checkpoints it, exits 130, and
+releases the lock, so ``--resume`` carries on from the interrupted step.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from dataclasses import replace
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+from pytest_mock import MockerFixture
+
+from stack_pr.autoland.commands import _run_resume, _stop_running_autoland
+from stack_pr.autoland.model import (
+    ConfirmStep,
+    LandingContext,
+    LandStep,
+    PRState,
+    StackEntry,
+)
+from stack_pr.autoland.services import Services
+from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock
+from stack_pr.autoland.steps import AutolandStep
+from stack_pr.autoland.worktree import Worktree
+from tests.helpers import common_args
+from tests.test_autoland_step_execution import (
+    OPTS,
+    FakeRepo,
+    FakeUI,
+    FakeWaits,
+    services,
+    stack,
+)
+
+INTERRUPTED = "User interrupted (Ctrl+C)"
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _save(path: Path, plan: list[AutolandStep]) -> None:
+    ctx = LandingContext(stack=stack(101, 102), plan=plan)
+    AutolandCheckpointer(path=path, branch="feat", base="main").save(ctx)
+
+
+def _land_both() -> list[AutolandStep]:
+    return [
+        LandStep(entry_index=0, pr_number=101),
+        LandStep(entry_index=1, pr_number=102),
+    ]
+
+
+def _waits(approved: list[int], *, interrupt_at: int | None = None) -> FakeWaits:
+    """Waits that record each approved PR, and see Ctrl+C while waiting for
+    *interrupt_at*'s approval."""
+
+    def approval(entry: StackEntry) -> bool:
+        if entry.pr_number == interrupt_at:
+            raise KeyboardInterrupt
+        approved.append(entry.pr_number)
+        return True
+
+    return FakeWaits(approval=approval)
+
+
+class InterruptedUI(FakeUI):
+    """Ctrl+C at the first prompt."""
+
+    def input(self, prompt: str) -> str:
+        self.lines.append(prompt)
+        raise KeyboardInterrupt
+
+
+@pytest.fixture
+def resume_io(mocker: MockerFixture) -> Mock:
+    """Stub the git side of a resume; return the Services.production mock."""
+    mocker.patch("stack_pr.git.get_current_branch_name", return_value="feat")
+    mocker.patch("stack_pr.cli.deduce_base", side_effect=lambda c: c)
+    mocker.patch("stack_pr.autoland.discovery.enrich_stack")
+    return mocker.patch.object(Services, "production")
+
+
+@pytest.fixture
+def worktree_events(mocker: MockerFixture) -> list[str]:
+    """Replace the --branch worktree with one that records what happens to it."""
+    events: list[str] = []
+
+    class FakeWorktree(Worktree):
+        def create(self) -> None:
+            self.path = Path("worktree")
+
+        def remove(self) -> None:
+            events.append("removed")
+
+        def announce_preserved(self) -> None:
+            events.append("preserved")
+
+    mocker.patch("stack_pr.autoland.commands.Worktree", FakeWorktree)
+    return events
+
+
+def _resume(sf: Path, **overrides: object) -> None:
+    _run_resume(common_args(), replace(OPTS, resume=True, state_file=sf, **overrides))
+
+
+def _printed(console: Mock) -> str:
+    return "\n".join(str(a) for c in console.print.call_args_list for a in c.args)
+
+
+def _assert_interrupted(sf: Path, console: Mock, *, at_step: int) -> LandingContext:
+    _, saved = AutolandCheckpointer.load(sf)
+    assert saved.abort_reason == INTERRUPTED
+    assert saved.current_step == at_step
+    assert "Interrupted! State saved. Resume with --resume." in _printed(console)
+    # The lock is free for the --resume.
+    lock = AutolandLock.for_state(sf)
+    assert lock.acquire()
+    lock.release()
+    return saved
+
+
+@pytest.mark.parametrize(
+    ("always_cleanup", "worktree"),
+    [(False, ["preserved"]), (True, ["removed"])],
+    ids=["keeps-worktree", "always-cleanup"],
+)
+def test_ctrl_c_during_a_wait_checkpoints_and_exits_130(
+    tmp_path: Path,
+    autoland_console: Mock,
+    resume_io: Mock,
+    *,
+    worktree_events: list[str],
+    always_cleanup: bool,
+    worktree: list[str],
+) -> None:
+    sf = tmp_path / "state.json"
+    _save(sf, _land_both())
+    resume_io.return_value = services(
+        waits=_waits([], interrupt_at=102), repo=FakeRepo()
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        _resume(sf, branch="feat", always_cleanup=always_cleanup)
+
+    assert exc.value.code == 130
+    saved = _assert_interrupted(sf, autoland_console, at_step=1)
+    assert saved.stack[0].state == PRState.MERGED
+    # Kept after an interruption, like after any failure, unless asked not to.
+    assert worktree_events == worktree
+
+
+def test_ctrl_c_at_a_confirm_prompt_checkpoints_and_exits_130(
+    tmp_path: Path, autoland_console: Mock, resume_io: Mock
+) -> None:
+    sf = tmp_path / "state.json"
+    _save(sf, [LandStep(entry_index=0, pr_number=101), ConfirmStep()])
+    resume_io.return_value = services(InterruptedUI(), repo=FakeRepo())
+
+    with pytest.raises(SystemExit) as exc:
+        _resume(sf)
+
+    assert exc.value.code == 130
+    saved = _assert_interrupted(sf, autoland_console, at_step=1)
+    confirm = saved.plan[1]
+    assert isinstance(confirm, ConfirmStep)
+    assert not confirm.confirmed
+
+
+@pytest.mark.usefixtures("autoland_console")
+def test_resume_after_ctrl_c_continues_from_the_interrupted_step(
+    tmp_path: Path, resume_io: Mock
+) -> None:
+    sf = tmp_path / "state.json"
+    _save(sf, _land_both())
+    approved: list[int] = []
+    resume_io.return_value = services(
+        waits=_waits(approved, interrupt_at=102), repo=FakeRepo()
+    )
+    with pytest.raises(SystemExit):
+        _resume(sf)
+    assert approved == [101]
+
+    resume_io.return_value = services(waits=_waits(approved), repo=FakeRepo())
+    _resume(sf)
+
+    # #101 isn't landed again; #102 is landed from the start of its step.
+    assert approved == [101, 102]
+    assert not sf.exists()
+
+
+# A run blocked in a wait in its own process, started with SIGINT ignored the
+# way a non-interactive shell starts a background job.
+_BLOCKED_RUN = """
+import signal, sys, time
+from dataclasses import replace
+from pathlib import Path
+from unittest import mock
+
+signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+from stack_pr.autoland import commands
+from stack_pr.autoland.services import Services
+from tests.helpers import common_args
+from tests.test_autoland_step_execution import OPTS, FakeRepo, FakeWaits, services
+
+def block(entry):
+    print("waiting", flush=True)
+    time.sleep(60)
+    return True
+
+sf = Path({path!r})
+with mock.patch("stack_pr.git.get_current_branch_name", return_value="feat"), \\
+        mock.patch("stack_pr.cli.deduce_base", side_effect=lambda c: c), \\
+        mock.patch("stack_pr.autoland.discovery.enrich_stack"), \\
+        mock.patch.object(Services, "production",
+                          return_value=services(waits=FakeWaits(approval=block),
+                                                repo=FakeRepo())):
+    commands._run_resume(common_args(), replace(OPTS, resume=True, state_file=sf))
+"""
+
+
+def test_takeover_stops_a_run_blocked_in_a_wait(
+    tmp_path: Path, autoland_console: Mock
+) -> None:
+    sf = tmp_path / "state.json"
+    _save(sf, _land_both())
+    run = subprocess.Popen(
+        [sys.executable, "-c", _BLOCKED_RUN.format(path=str(sf))],
+        cwd=REPO_ROOT,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        assert run.stdout is not None
+        for line in run.stdout:
+            if line.strip() == "waiting":
+                break
+        autoland_console.input.return_value = "y"
+        lock = AutolandLock.for_state(sf)
+
+        assert _stop_running_autoland(lock)
+        lock.release()
+
+        output, _ = run.communicate(timeout=10)
+        assert run.returncode == 130, output
+        assert "Interrupted! State saved. Resume with --resume." in output
+        _, saved = AutolandCheckpointer.load(sf)
+        assert saved.abort_reason == INTERRUPTED
+        assert saved.current_step == 0
+    finally:
+        run.kill()
+        run.wait()
diff --git a/tests/test_autoland_optimize.py b/tests/test_autoland_optimize.py
index f4f9e91..47bca31 100644
--- a/tests/test_autoland_optimize.py
+++ b/tests/test_autoland_optimize.py
@@ -282,11 +282,15 @@ def test_a_stack_merge_that_falls_through_lands_the_rest_one_at_a_time(
 def test_a_stack_merge_abort_stops_before_the_step(mocker) -> None:  # noqa: ANN001
     run = LandingContext(stack=stack(101, 102), plan=land(0, 1))
     _stack_merge(
-        mocker, [NativeStackMergeResult(abort_reason="Stack merge was aborted")]
+        mocker,
+        [NativeStackMergeResult(abort_reason="Rebase failed after merging #102: boom")],
     )
     recorder = _Recorder(run)
 
-    assert _unit(run).execute(recorder.context()) == "Stack merge was aborted"
+    assert (
+        _unit(run).execute(recorder.context())
+        == "Rebase failed after merging #102: boom"
+    )
     assert recorder.events == [("enter", 0)]
 
 
diff --git a/tests/test_autoland_step_execution.py b/tests/test_autoland_step_execution.py
index 405e18c..b8073bb 100644
--- a/tests/test_autoland_step_execution.py
+++ b/tests/test_autoland_step_execution.py
@@ -105,21 +105,15 @@ class FakeWaits:
     calls: list[str] = field(default_factory=list)
     workflow_targets: list[str] = field(default_factory=list)
 
-    def wait_for_approval(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
+    def wait_for_approval(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
         self.calls.append("approval")
         return self.approval(entry)
 
-    def wait_for_checks(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
+    def wait_for_checks(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
         self.calls.append("checks")
         return self.checks(entry)
 
-    def enqueue_and_wait(
-        self, entry: StackEntry, *, opts: AutolandOptions, ctx: LandingContext
-    ) -> bool:
+    def enqueue_and_wait(self, entry: StackEntry, *, opts: AutolandOptions) -> bool:
         self.calls.append("queue")
         return self.queue(entry)
 
@@ -294,7 +288,7 @@ def test_already_landed_step_reports_its_sha_only_at_the_end_of_the_prefix(
 @pytest.mark.parametrize(
     ("waits", "reason"),
     [
-        (FakeWaits(approval=fails), "PR #101 approval wait was aborted"),
+        (FakeWaits(approval=fails), "PR #101 was closed while waiting for approval"),
         (FakeWaits(checks=fails), "PR #101 checks failed after retries"),
         (FakeWaits(queue=fails), "PR #101 failed to merge"),
     ],
@@ -454,7 +448,6 @@ def test_engine_records_the_reason_a_step_aborted(tmp_path: Path) -> None:
         ctx, common_args(), OPTS, checkpointer, services=services(FakeUI())
     )
 
-    assert ctx.aborted
     assert ctx.abort_reason.startswith("Confirm step received EOF")
     _, saved = AutolandCheckpointer.load(checkpointer.path)
     assert saved.abort_reason == ctx.abort_reason