refactor(autoland): inject services everywhere and test with fakes instead of patches

src/stack_pr/autoland/__init__.py

@@ -28,24 +28,31 @@
 - ``checks``: evaluating CI checks.
 - ``plan``: generating, formatting, parsing, and replanning landing plans.
 - ``display``: rendering a plan's progress.
+- ``repo``: the local clone (fetch, ancestry, landed SHAs, rebase + resubmit).
 - ``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.
+- ``services``: the tools a plan's execution uses (console, GitHub, sleep,
+  the repo, the waits), and their production bindings.
 - ``optimize``: execution units, the plan optimization policies that group
   steps into them (e.g. ``--merge-as-stack``), and the registry of policies.
 - ``engine``: optimizing and executing a plan.
 - ``status``: the ``--status`` report.
 - ``commands``: the subparser, and the fresh / resume / replan flows.
 
-Convention for shared state and patch targets: a module-level singleton
-(``runtime.console``, the ``gh.github`` client, ``runtime.HAVE_RICH``,
-``display.GLYPHS``) lives in exactly one module, and other modules reach it
-through that module -- ``from stack_pr.autoland import runtime``, then
-``runtime.console`` -- never ``from stack_pr.autoland.runtime import console``.
-The same goes for a function that tests replace to intercept calls made from
-another module: ``runtime.run``, ``runtime.resilient_sleep``, the
-``waits.wait_for_*`` helpers, ``engine.execute_plan``, the ``discovery``
+Executing a plan (the engine, the steps, the execution units, the waits, the
+stack merge, and the repo operations) uses only the ``services.Services`` it
+is given, so tests inject fakes there instead of patching modules.
+
+Convention for the shared state that remains, and its patch targets: a
+module-level singleton (``runtime.console``, the ``gh.github`` client,
+``runtime.HAVE_RICH``, ``display.GLYPHS``) lives in exactly one module, and
+other modules reach it through that module -- ``from stack_pr.autoland import
+runtime``, then ``runtime.console`` -- never ``from stack_pr.autoland.runtime
+import console``. ``Services.production()`` binds the singletons as they are
+when it is called; the command flows' own output, ``--status``, plan parsing,
+discovery, and the shell runner use them directly. The same goes for a
+function that tests replace to intercept calls made from another module:
+``runtime.resilient_sleep``, ``engine.execute_plan``, the ``discovery``
 functions, and ``stack_pr.git.get_current_branch_name``. Each therefore has one
 canonical patch target, e.g. ``mocker.patch("stack_pr.autoland.runtime.console")``,
 and a patch can't silently miss a stale copy bound in another module. (A name

src/stack_pr/autoland/commands.py

@@ -243,6 +243,8 @@ def _land(
     opts: AutolandOptions,
     checkpointer: AutolandCheckpointer,
     worktree: Worktree | None,
+    *,
+    services: Services,
 ) -> None:
     """Execute the plan, report how the run ended, and exit if it didn't succeed.
 
@@ -263,7 +265,7 @@ def _land(
     signal.signal(signal.SIGINT, signal.default_int_handler)
     try:
         success = engine.execute_plan(
-            ctx, common, opts, checkpointer, services=Services.production()
+            ctx, common, opts, checkpointer, services=services
         )
     except KeyboardInterrupt:
         ctx.abort_reason = "User interrupted (Ctrl+C)"
@@ -349,7 +351,16 @@ def _ask_replan_or_overwrite(state_path: Path) -> str | None:
     return None
 
 
-def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None:
+def _run_fresh(
+    common: cli.CommonArgs,
+    opts: AutolandOptions,
+    *,
+    services: Services | None = None,
+) -> None:
+    """Land a new plan. *services* (default: production) is what it is
+    executed with."""
+    if services is None:
+        services = Services.production()
     branch = opts.branch or git.get_current_branch_name()
     state_path = opts.state_file or AutolandCheckpointer.default_path(branch)
 
@@ -386,7 +397,7 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None:
                 return
             replan = choice == "replan"
         if replan:
-            _replan(common, opts, state_path)
+            _replan(common, opts, state_path, services=services)
             return
 
         if opts.branch:
@@ -433,21 +444,30 @@ def _run_fresh(common: cli.CommonArgs, opts: AutolandOptions) -> None:
         )
 
         print_status(ctx)
-        engine.print_execution_preview(ctx, opts)
+        engine.print_execution_preview(ctx, opts, ui=services.ui)
         if opts.dry_run:
             runtime.console.print("\n[yellow]Dry run — exiting.[/yellow]")
             return
 
         runtime.console.print(f"[dim]State file: {checkpointer.path}[/dim]\n")
-        _land(ctx, common, opts, checkpointer, worktree)
+        _land(ctx, common, opts, checkpointer, worktree, services=services)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()
         if lock is not None:
             lock.release()
 
 
-def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None:
+def _run_resume(
+    common: cli.CommonArgs,
+    opts: AutolandOptions,
+    *,
+    services: Services | None = None,
+) -> None:
+    """Resume the run checkpointed in the state file. *services* (default:
+    production) is what it is executed with."""
+    if services is None:
+        services = Services.production()
     sf_path = _state_path(opts)
 
     if not sf_path.exists():
@@ -502,9 +522,9 @@ def _run_resume(common: cli.CommonArgs, opts: AutolandOptions) -> None:
             return
 
         print_status(ctx)
-        engine.print_execution_preview(ctx, opts)
+        engine.print_execution_preview(ctx, opts, ui=services.ui)
         runtime.console.print(f"[dim]State file: {sf_path}[/dim]\n")
-        _land(ctx, common, opts, checkpointer, worktree)
+        _land(ctx, common, opts, checkpointer, worktree, services=services)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()
@@ -634,12 +654,20 @@ def _replacement_plan(
         sys.exit(1)
 
 
-def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) -> None:
+def _replan(
+    common: cli.CommonArgs,
+    opts: AutolandOptions,
+    state_path: Path,
+    *,
+    services: Services | None = None,
+) -> None:
     """Replace the plan of the run checkpointed at *state_path*, keep its
-    progress, and continue.
+    progress, and continue, executing with *services* (default: production).
 
     The caller holds the lock, except on a dry run, which only previews.
     """
+    if services is None:
+        services = Services.production()
     try:
         old_checkpointer, old = AutolandCheckpointer.load(state_path)
     except (OSError, ValueError, KeyError) as e:
@@ -678,7 +706,7 @@ def _replan(common: cli.CommonArgs, opts: AutolandOptions, state_path: Path) ->
         ctx = LandingContext(stack=stack, plan=plan)
 
         print_status(ctx)
-        engine.print_execution_preview(ctx, opts)
+        engine.print_execution_preview(ctx, opts, ui=services.ui)
         if lost:
             runtime.console.print(
                 "\n[yellow]Done in the previous run, but not carried over (changed, "
@@ -719,7 +747,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")
-        _land(ctx, common, opts, checkpointer, worktree)
+        _land(ctx, common, opts, checkpointer, worktree, services=services)
     finally:
         if worktree is not None:
             worktree.remove_unless_landing()

src/stack_pr/autoland/engine.py

@@ -14,13 +14,12 @@
 import functools
 
 from stack_pr import cli
-from stack_pr.autoland import runtime
 from stack_pr.autoland.display import _describe_step
 from stack_pr.autoland.model import LandingContext
 from stack_pr.autoland.optimize import PolicyRegistry, UnitContext, optimize_plan
 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.services import UI, Services
 from stack_pr.autoland.state import AutolandCheckpointer
 from stack_pr.autoland.steps import StepContext, StepResult
 
@@ -98,13 +97,14 @@ def print_execution_preview(
     ctx: LandingContext,
     opts: AutolandOptions,
     *,
+    ui: UI,
     policies: PolicyRegistry | None = None,
 ) -> None:
     """Announce what the optimized plan does beyond executing its steps one by
     one. Derived from the same units ``execute_plan`` executes."""
     for unit in optimize_plan(ctx, opts, policies):
         for line in unit.preview():
-            runtime.console.print(line)
+            ui.print(line)
 
 
 def _record_landed_sha(ctx: LandingContext, sha: str) -> None: