refactor(autoland): execute each step through a StepContext with injected services

src/stack_pr/autoland/__init__.py

@@ -17,7 +17,8 @@
 - ``options``: AutolandOptions, resolved from flags, config, and defaults.
 - ``runtime``: the console, sleep/wake resilience, and the retrying ``run``.
 - ``steps``: the AutolandStep base class, the registry of step types, and one
-  module per type, each owning its plan syntax, persistence, and progress.
+  module per type, each owning its plan syntax, persistence, progress, view,
+  and execution; plus StepContext/StepResult, what a step executes with.
 - ``model``: the stack, plan steps, and LandingContext, and their
   (de)serialization.
 - ``state``: the checkpoint (``--resume``) and the per-branch lock.
@@ -29,6 +30,8 @@
 - ``display``: rendering a plan's progress.
 - ``waits``: waiting for approval, checks, the merge queue, and workflows.
 - ``native_stack``: merging a run of land steps as one GitHub stack.
+- ``services``: the tools a step uses as it executes (console, GitHub, the
+  waits, the repo), and their production bindings.
 - ``engine``: executing a plan.
 - ``status``: the ``--status`` report.
 - ``commands``: the subparser, and the fresh / resume / replan flows.

src/stack_pr/autoland/commands.py

@@ -27,6 +27,7 @@
     parse_plan,
     plan_from_file,
 )
+from stack_pr.autoland.services import Services
 from stack_pr.autoland.state import AutolandCheckpointer, AutolandLock, _state_path
 from stack_pr.autoland.status import show_status
 from stack_pr.autoland.worktree import Worktree
@@ -428,7 +429,9 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None:
             checkpointer,
             worktree,
             opts,
-            success=engine.execute_plan(ctx, common, opts, checkpointer),
+            success=engine.execute_plan(
+                ctx, common, opts, checkpointer, services=Services.production()
+            ),
         )
     finally:
         if worktree is not None:
@@ -501,7 +504,9 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None:
             checkpointer,
             worktree,
             opts,
-            success=engine.execute_plan(ctx, common, opts, checkpointer),
+            success=engine.execute_plan(
+                ctx, common, opts, checkpointer, services=Services.production()
+            ),
         )
     finally:
         if worktree is not None:
@@ -723,7 +728,9 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) ->
             checkpointer,
             worktree,
             opts,
-            success=engine.execute_plan(ctx, common, opts, checkpointer),
+            success=engine.execute_plan(
+                ctx, common, opts, checkpointer, services=Services.production()
+            ),
         )
     finally:
         if worktree is not None:

src/stack_pr/autoland/display.py

@@ -311,13 +311,3 @@ def _describe_step(step: PlanStep, ctx: LandingContext) -> str:
     differently.
     """
     return step.view(ctx.stack, is_next=False, assumed_completed=False).description
-
-
-def _next_steps_lines(
-    plan: list[PlanStep], from_index: int, ctx: LandingContext
-) -> list[str]:
-    """Numbered, markup-safe descriptions of the plan steps after `from_index`."""
-    return [
-        f"  {n}. {_escape_markup(_describe_step(s, ctx))}"
-        for n, s in enumerate(plan[from_index + 1 :], 1)
-    ]

src/stack_pr/autoland/engine.py

@@ -1,189 +1,137 @@
-"""Executing a landing plan step by step."""
+"""Executing a landing plan step by step.
+
+The engine owns the run: it walks the plan from ``ctx.current_step``,
+checkpoints before each step, hands each step a ``StepContext``, records the
+run-level facts a step reports in its ``StepResult``, and stops on an abort.
+What a step does is up to the step (``AutolandStep.execute``).
+"""
 
 from __future__ import annotations
 
+import functools
+
 from stack_pr import cli
-from stack_pr.autoland import runtime, waits
-from stack_pr.autoland.display import _escape_markup, _next_steps_lines
-from stack_pr.autoland.model import ConfirmStep, LandingContext, LandStep, WorkflowStep
+from stack_pr.autoland.display import _describe_step
+from stack_pr.autoland.model import LandingContext
 from stack_pr.autoland.native_stack import land_as_native_stack, native_stack_run
 from stack_pr.autoland.options import AutolandOptions
 from stack_pr.autoland.plan import landed_prefix_end
+from stack_pr.autoland.services import Services
 from stack_pr.autoland.state import AutolandCheckpointer
+from stack_pr.autoland.steps import StepContext
 
 
 def execute_plan(
     ctx: LandingContext,
     common: cli.CommonArgs,
     opts: AutolandOptions,
     checkpointer: AutolandCheckpointer,
+    *,
+    services: Services | None = None,
 ) -> bool:
     """Execute the landing plan from ctx.current_step. Returns True on success."""
-    # A workflow checkpoint after the already-landed prefix of a plan should
-    # wait for the code of the last PR that prefix landed. Only that last one
-    # matters, so don't pay for a fetch + lookup on each of the others.
-    last_prelanded = landed_prefix_end(ctx.plan)
-    # The last step of a run whose stack merge fell through; the rest of that
-    # run lands one PR at a time rather than retrying the stack merge.
-    unstacked_through = -1
+    if services is None:
+        services = Services.production()
+    prefix_end = landed_prefix_end(ctx.plan)
+    native_stack = _NativeStackMerge(ctx, common, opts, checkpointer)
+
+    def upcoming(index: int) -> list[str]:
+        return [_describe_step(s, ctx) for s in ctx.plan[index + 1 :]]
+
+    def landed_sha() -> str:
+        if not ctx.last_landed_sha:
+            _record_landed_sha(ctx, services.repo.landed_sha(common))
+        return ctx.last_landed_sha
 
     for step_idx in range(ctx.current_step, len(ctx.plan)):
         step = ctx.plan[step_idx]
         ctx.current_step = step_idx
         checkpointer.save(ctx)
 
-        if isinstance(step, LandStep) and step.already_landed:
-            runtime.console.print(
-                f"\n[green]PR #{step.pr_number} already landed, skipping[/green]"
-            )
-            if step_idx == last_prelanded:
-                waits._refresh_last_landed_sha(ctx, common, step.pr_number)
-
-        elif isinstance(step, LandStep):
-            entry = ctx.stack[step.entry_index]
-            ctx.current_index = step.entry_index
-
-            run_entries = (
-                native_stack_run(ctx, step_idx)
-                if opts.merge_as_stack and step_idx > unstacked_through
-                else []
-            )
-            if run_entries:
-                result = land_as_native_stack(
-                    run_entries, ctx=ctx, common=common, opts=opts
-                )
-                if result.abort_reason:
-                    return _abort(ctx, checkpointer, result.abort_reason)
-                if not result.landed:
-                    unstacked_through = step_idx + len(run_entries) - 1
-                checkpointer.save(ctx)
-
-            if entry.is_merged():
-                runtime.console.print(
-                    f"\n[green]PR #{entry.pr_number} already merged, skipping[/green]"
-                )
-                waits._refresh_last_landed_sha(ctx, common, entry.pr_number)
-                continue
-
-            runtime.console.print(
-                f"\n{'=' * 60}\n[bold]Step {step_idx + 1}/{len(ctx.plan)}: "
-                f"Landing PR #{entry.pr_number} — {entry.title}[/bold]\n{'=' * 60}"
+        abort_reason = native_stack.before_step(step_idx)
+        if abort_reason:
+            return _abort(ctx, checkpointer, abort_reason)
+
+        result = step.execute(
+            StepContext(
+                index=step_idx,
+                total=len(ctx.plan),
+                run=ctx,
+                opts=opts,
+                common=common,
+                services=services,
+                checkpoint=lambda: checkpointer.save(ctx),
+                upcoming=functools.partial(upcoming, step_idx),
+                landed_sha=landed_sha,
+                ends_landed_prefix=step_idx == prefix_end,
             )
-
-            if not waits.wait_for_approval(entry, opts=opts, ctx=ctx):
-                return _abort(
-                    ctx,
-                    checkpointer,
-                    f"PR #{entry.pr_number} approval wait was aborted",
-                )
-
-            if entry.is_merged():
-                waits._refresh_last_landed_sha(ctx, common, entry.pr_number)
-            else:
-                if not waits.wait_for_checks(entry, opts=opts, ctx=ctx):
-                    return _abort(
-                        ctx,
-                        checkpointer,
-                        f"PR #{entry.pr_number} checks failed after retries",
-                    )
-
-                if entry.is_merged():
-                    waits._refresh_last_landed_sha(ctx, common, entry.pr_number)
-                else:
-                    if not waits.enqueue_and_wait(entry, opts=opts, ctx=ctx):
-                        return _abort(
-                            ctx,
-                            checkpointer,
-                            f"PR #{entry.pr_number} failed to merge",
-                        )
-                    waits._refresh_last_landed_sha(ctx, common, entry.pr_number)
-
-            # Rebase + resubmit whenever commits remain above the one we just
-            # landed — not only when more *land steps* follow. On a partial
-            # land, the PRs we're leaving open still need their bases rebased
-            # onto the newly-landed commit.
-            has_commits_above = step.entry_index < len(ctx.stack) - 1
-            if has_commits_above:
-                try:
-                    waits.rebase_and_resubmit(common)
-                except Exception as e:  # noqa: BLE001 - report any resubmit failure
-                    return _abort(
-                        ctx,
-                        checkpointer,
-                        f"Rebase failed after merging #{entry.pr_number}: {e}",
-                    )
-
-        elif isinstance(step, WorkflowStep):
-            # "succeeded" before the loop reaches it means --replan carried the
-            # result over from the run this one replaced.
-            if step.state in ("skipped", "succeeded"):
-                continue
-            runtime.console.print(
-                f"\n{'=' * 60}\n[bold]Step {step_idx + 1}/{len(ctx.plan)}: "
-                f"Workflow checkpoint — {step.workflow}[/bold]\n{'=' * 60}"
-            )
-            if not ctx.last_landed_sha:
-                waits._refresh_last_landed_sha(ctx, common)
-            if not waits.wait_for_workflow(step, opts=opts, common=common, ctx=ctx):
-                return _abort(
-                    ctx,
-                    checkpointer,
-                    f"Workflow {step.workflow} failed or timed out",
-                )
-
-        elif isinstance(step, ConfirmStep):
-            if step.confirmed:
-                continue
-            question = (
-                f'Confirm "{_escape_markup(step.condition)}" is complete — '
-                "ready to proceed?"
-                if step.condition
-                else "Ready to proceed?"
-            )
-            lines = [
-                f"\n{'=' * 60}",
-                (
-                    f"[bold yellow]Step {step_idx + 1}/{len(ctx.plan)}: "
-                    "Manual confirmation required[/bold yellow]"
-                ),
-                f"{'=' * 60}\n",
-                f"[bold]{question}[/bold]\n",
-            ]
-            next_lines = _next_steps_lines(ctx.plan, step_idx, ctx)
-            if next_lines:
-                lines.append("[bold]Next steps:[/bold]")
-                lines.extend(next_lines)
-            else:
-                lines.append("[dim]This is the final step in the plan.[/dim]")
-            runtime.console.print("\n".join(lines))
-            while True:
-                try:
-                    answer = runtime.console.input(
-                        "[yellow]Type y/Y then Enter to continue "
-                        "(Ctrl+C to abort): [/yellow]"
-                    ).strip()
-                except EOFError:
-                    return _abort(
-                        ctx,
-                        checkpointer,
-                        "Confirm step received EOF — cannot confirm in "
-                        "non-interactive mode",
-                    )
-                if answer in ("y", "Y"):
-                    break
-                runtime.console.print("[dim]Type 'y' or 'Y' to confirm.[/dim]")
-            step.confirmed = True
-            runtime.console.print("[green]Confirmed[/green]")
+        )
+        _record_landed_sha(ctx, result.landed_sha)
+        if result.aborted:
+            return _abort(ctx, checkpointer, result.abort_reason)
 
     ctx.current_step = len(ctx.plan)
     checkpointer.save(ctx)
     return True
 
 
+def _record_landed_sha(ctx: LandingContext, sha: str) -> None:
+    # An unknown SHA ("") keeps the last one known.
+    if sha:
+        ctx.last_landed_sha = sha
+
+
 def _abort(
     ctx: LandingContext, checkpointer: AutolandCheckpointer, reason: str
 ) -> bool:
     ctx.abort_reason = reason
     ctx.aborted = True
     checkpointer.save(ctx)
     return False
+
+
+class _NativeStackMerge:
+    """TEMPORARY: merge a run of consecutive land steps as one GitHub stack.
+
+    With ``--merge-as-stack``, before each step the engine asks whether a run
+    of two or more open land steps starts there, and if so lands the run with
+    one stack merge (``native_stack.land_as_native_stack``). The land steps
+    then find their PRs merged and only skip. A run whose stack merge fell
+    through is landed one step at a time instead, without retrying the merge.
+
+    This lives in the engine, not in ``LandStep``, because steps are
+    independent of each other. It is a stopgap: a plan-optimization pass that
+    groups land steps into one execution unit replaces it.
+    """
+
+    def __init__(
+        self,
+        ctx: LandingContext,
+        common: cli.CommonArgs,
+        opts: AutolandOptions,
+        checkpointer: AutolandCheckpointer,
+    ) -> None:
+        self._ctx = ctx
+        self._common = common
+        self._opts = opts
+        self._checkpointer = checkpointer
+        # The last step of a run whose stack merge fell through.
+        self._unstacked_through = -1
+
+    def before_step(self, step_idx: int) -> str:
+        """Stack-merge the run starting at *step_idx*, if any. Returns the
+        reason the plan must stop, or ""."""
+        if not self._opts.merge_as_stack or step_idx <= self._unstacked_through:
+            return ""
+        run_entries = native_stack_run(self._ctx, step_idx)
+        if not run_entries:
+            return ""
+        result = land_as_native_stack(
+            run_entries, ctx=self._ctx, common=self._common, opts=self._opts
+        )
+        if result.abort_reason:
+            return result.abort_reason
+        if not result.landed:
+            self._unstacked_through = step_idx + len(run_entries) - 1
+        self._checkpointer.save(self._ctx)
+        return ""