refactor(autoland): give each plan step type its own class

pyproject.toml

@@ -170,6 +170,13 @@ multiline-quotes = "double"
     "PLR0915",
     "TRY300",
 ]
+# Step types implement one interface: an override may ignore arguments that
+# others need, and the base class's optional hooks default to doing nothing.
+"src/stack_pr/autoland/steps/*.py" = [
+    "ARG002",
+    "ARG003",
+    "B027",
+]
 # Allow certain useful patterns in tests
 "tests/**/*.py" = [
     "S101", # allow assert statements within if statements

src/stack_pr/autoland/__init__.py

@@ -16,6 +16,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.
 - ``model``: the stack, plan steps, and LandingContext, and their
   (de)serialization.
 - ``state``: the checkpoint (``--resume``) and the per-branch lock.

src/stack_pr/autoland/display.py

@@ -145,9 +145,10 @@ def _icon(outcome: _Outcome) -> str:
     PRState.FAILED: _Outcome.FAILED,
 }
 
-# "skipped" is set only by _mark_skipped, i.e. for a workflow that sits inside a
-# plan's already-landed prefix — it must have run before those PRs could merge,
-# so the plan assumes it did rather than claiming it was deliberately skipped.
+# "skipped" is set only by WorkflowStep.assume_completed, i.e. for a workflow
+# that sits inside a plan's already-landed prefix — it must have run before
+# those PRs could merge, so the plan assumes it did rather than claiming it was
+# deliberately skipped.
 _WORKFLOW_STEP_STATUS = {
     "pending": (_Outcome.PENDING, "Pending", "dim"),
     "waiting": (_Outcome.ACTIVE, "Waiting for workflow", "blue"),
@@ -222,6 +223,7 @@ def _step_status(
     # ConfirmStep. A confirmation inside the landed prefix was never actually
     # given in this run — it is inferred from the fact that later PRs merged —
     # so say so rather than claiming the user confirmed it.
+    assert isinstance(step, ConfirmStep)  # the only step type left
     if step.confirmed:
         if is_assumed_completed(plan, index):
             return _Outcome.DONE, "Assumed completed", "green"
@@ -240,6 +242,7 @@ def _step_target(step: PlanStep, ctx: LandingContext) -> tuple[str, str]:
         return f"#{entry.pr_number}", entry.title or "(untitled)"
     if isinstance(step, WorkflowStep):
         return step.workflow, ""
+    assert isinstance(step, ConfirmStep)  # the only step type left
     return step.condition or "(no condition)", ""
 
 
@@ -422,6 +425,7 @@ def _describe_step(step: PlanStep, ctx: LandingContext) -> str:
     if isinstance(step, WorkflowStep):
         return f"Wait for workflow {key}"
     # ConfirmStep
+    assert isinstance(step, ConfirmStep)  # the only step type left
     if step.condition:
         return f"Manual confirmation: {key}"
     return "Manual confirmation"

src/stack_pr/autoland/model.py

@@ -2,9 +2,27 @@
 
 from __future__ import annotations
 
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field
 from enum import Enum
-from typing import Union
+
+from stack_pr.autoland.steps import (
+    AutolandStep,
+    ConfirmStep,
+    LandStep,
+    WorkflowStep,
+    step_from_dict,
+)
+
+__all__ = [
+    "AutolandStep",
+    "ConfirmStep",
+    "LandStep",
+    "LandingContext",
+    "PRState",
+    "PlanStep",
+    "StackEntry",
+    "WorkflowStep",
+]
 
 # ---------------------------------------------------------------------------
 # Data types
@@ -47,51 +65,9 @@ def is_merged(self) -> bool:
         return self.state == PRState.MERGED
 
 
-@dataclass
-class LandStep:
-    """Land a PR in the stack through the merge queue.
-
-    ``entry_index`` indexes into ``LandingContext.stack``, or is ``-1`` for a
-    PR that has already landed: such a PR is no longer part of the stack, so
-    the step is skipped at execution time.
-
-    ``pr_number`` is set when the plan pinned a specific PR (``l 123``) and is
-    ``None`` for a bare, positional ``l``. It is the only thing identifying an
-    already-landed step, so it is always set when ``entry_index`` is ``-1``.
-    """
-
-    entry_index: int
-    pr_number: int | None = None
-
-    @property
-    def already_landed(self) -> bool:
-        return self.entry_index < 0
-
-
-@dataclass
-class WorkflowStep:
-    """Wait for a GitHub Actions workflow to succeed with the landed code."""
-
-    workflow: str
-    state: str = "pending"  # pending, waiting, succeeded, failed, skipped
-    error_message: str = ""
-
-
-@dataclass
-class ConfirmStep:
-    """Pause for manual confirmation before continuing.
-
-    ``condition`` is an optional human-readable thing to verify before
-    proceeding (e.g. ``"QA sign-off complete"``). When set, it is shown in the
-    confirmation prompt; when empty, a generic prompt is shown. Either way the
-    step waits until the user types ``y``/``Y`` and presses Enter.
-    """
-
-    condition: str = ""
-    confirmed: bool = False
-
-
-PlanStep = Union[LandStep, WorkflowStep, ConfirmStep]
+# The step types live in the steps package; re-exported here, where the rest of
+# autoland has always imported them from. PlanStep predates the base class.
+PlanStep = AutolandStep
 
 
 @dataclass
@@ -112,9 +88,6 @@ def abort_requested(self) -> bool:
         return self.aborted
 
 
-_STEP_TYPES = {LandStep: "land", WorkflowStep: "workflow", ConfirmStep: "confirm"}
-
-
 def _deserialize_entry(data: dict) -> StackEntry:
     return StackEntry(
         pr_url=data["pr_url"],
@@ -130,14 +103,9 @@ def _deserialize_entry(data: dict) -> StackEntry:
 
 
 def _serialize_step(step: PlanStep) -> dict:
-    # dataclasses.asdict gives the fields; tag the type for deserialization.
-    return {"type": _STEP_TYPES[type(step)], **asdict(step)}
+    return step.to_dict()
 
 
 def _deserialize_step(data: dict) -> PlanStep:
-    fields = {k: v for k, v in data.items() if k != "type"}
-    if data["type"] == "land":
-        return LandStep(**fields)
-    if data["type"] == "workflow":
-        return WorkflowStep(**fields)
-    return ConfirmStep(**fields)
+    """Raises ValueError for a step type no registered step class handles."""
+    return step_from_dict(data)