From fd9475168409286c7c0909bfe1c7f1f659858bcd Mon Sep 17 00:00:00 2001 From: Micah Smith Date: Mon, 12 Oct 2026 17:27:34 -0400 Subject: [PATCH] refactor(autoland): give each plan step type its own class MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Phase 1 of the autoland step-abstraction stack. Plan steps were plain dataclasses joined by `PlanStep = Union[LandStep, WorkflowStep, ConfirmStep]`, with `isinstance` chains in every module that handled a plan. This adds an `AutolandStep` base class and moves the per-type behavior for plan syntax, persistence, and replanning progress onto the step classes. - New `stack_pr.autoland.steps` package: `base.py` (the `AutolandStep` ABC, a `PlanParser` context holding the stack, `pr_is_merged`, the next positional index, and the current line number, and the step registry), plus `land.py`, `workflow.py`, and `confirm.py`. The step classes keep their names and fields and are re-exported from `model.py`. `PlanStep` is now an alias of `AutolandStep`. - Each step type declares `kind` (state-file tag), `keyword` (plan syntax), and `label`. `@register_step` adds it to a registry that is looked up by kind or keyword. Registering the same kind or keyword twice raises an error. - Persistence: `to_dict()` / `from_dict()` replace `model._STEP_TYPES` and the type switch in `_deserialize_step`. The state file is byte-for-byte unchanged (`STATE_VERSION` stays 1). - Plan syntax: `parse(args, parser)` and `plan_line(stack)` per step. `parse_plan` and `format_plan_for_editor` now dispatch through the registry. The comment-column rule is a per-step `aligns_comments` flag (false for confirm). A bare `w` still counts as an unrecognized step, via `bare_keyword_ok`. PR-reference/URL resolution moves into `steps/land.py` unchanged. - Progress: `assume_completed()`, `is_checkpoint_done()`, `replan_key(landed_prs)`, `credit_from(old)`, `landed_pr(stack)`, and `marks_landed_prefix` replace `plan._mark_skipped`, `_checkpoint_done`, `_credit`, and the type switch in `_checkpoint_keys`. `landed_prefix_end`, `is_assumed_completed`, and `carry_over_progress` stay in `plan.py` and now call these methods. - `engine.py`, `display.py`, `status.py`, and `native_stack.py` keep their `isinstance` dispatch for now; later phases handle them. `display.py` and `status.py` only get `assert isinstance(step, ConfirmStep)` at their "the remaining type" fallthroughs, because mypy can no longer narrow a base class to the last member of the old union. **Behavior change:** a state file step with an unknown `type` now fails to load with `Unknown plan step type: ''`. Before, it was silently loaded as a `ConfirmStep`, which was a latent bug. `--resume`/`--replan` already report `ValueError`s from loading as "Failed to load state file". ## Test plan - [x] Golden autoland scenarios (`tests/test_autoland_golden.py`) pass with no golden files regenerated, including the checkpoint JSON - [x] New `tests/test_autoland_steps.py`: registry lookup by kind and keyword, duplicate kind/keyword registration error (with no partial registration), unknown state-file step type error, and a round trip of every registered step type through the state file and through plan text (`format_plan_for_editor` -> `parse_plan`) - [x] Full suite passes with `GH_CONFIG_DIR` empty and `GH_TOKEN` unset 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 5.5 (1M context) stack-info: PR: https://github.com/micahjsmith/stack-pr/pull/69, branch: micahjsmith/stack/3 --- pyproject.toml | 7 + src/stack_pr/autoland/__init__.py | 2 + src/stack_pr/autoland/display.py | 10 +- src/stack_pr/autoland/model.py | 84 +++----- src/stack_pr/autoland/plan.py | 248 +++--------------------- src/stack_pr/autoland/status.py | 5 +- src/stack_pr/autoland/steps/__init__.py | 33 ++++ src/stack_pr/autoland/steps/base.py | 171 ++++++++++++++++ src/stack_pr/autoland/steps/confirm.py | 58 ++++++ src/stack_pr/autoland/steps/land.py | 172 ++++++++++++++++ src/stack_pr/autoland/steps/workflow.py | 55 ++++++ tests/test_autoland_steps.py | 130 +++++++++++++ 12 files changed, 696 insertions(+), 279 deletions(-) create mode 100644 src/stack_pr/autoland/steps/__init__.py create mode 100644 src/stack_pr/autoland/steps/base.py create mode 100644 src/stack_pr/autoland/steps/confirm.py create mode 100644 src/stack_pr/autoland/steps/land.py create mode 100644 src/stack_pr/autoland/steps/workflow.py create mode 100644 tests/test_autoland_steps.py diff --git a/pyproject.toml b/pyproject.toml index 62c69b5..d34aad8 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/stack_pr/autoland/__init__.py b/src/stack_pr/autoland/__init__.py index e3ef3cf..002d164 100644 --- a/src/stack_pr/autoland/__init__.py +++ b/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. diff --git a/src/stack_pr/autoland/display.py b/src/stack_pr/autoland/display.py index 3cdbd0e..a56dac9 100644 --- a/src/stack_pr/autoland/display.py +++ b/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" diff --git a/src/stack_pr/autoland/model.py b/src/stack_pr/autoland/model.py index caf86ea..8175cd5 100644 --- a/src/stack_pr/autoland/model.py +++ b/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) diff --git a/src/stack_pr/autoland/plan.py b/src/stack_pr/autoland/plan.py index d23f47f..b55a2aa 100644 --- a/src/stack_pr/autoland/plan.py +++ b/src/stack_pr/autoland/plan.py @@ -3,7 +3,6 @@ from __future__ import annotations import os -import re import shlex import subprocess import sys @@ -13,13 +12,13 @@ from stack_pr.autoland import gh, runtime from stack_pr.autoland.model import ( - ConfirmStep, LandingContext, LandStep, PlanStep, StackEntry, WorkflowStep, ) +from stack_pr.autoland.steps import REGISTRY, PlanParser, ReplanKey from stack_pr.shell_commands import run_shell_command # --------------------------------------------------------------------------- @@ -68,81 +67,25 @@ def format_plan_for_editor(stack: list[StackEntry], plan: list[PlanStep]) -> str "#", ] - # Each step as (keyword, line, trailing comment), so the comments can be - # aligned once the longest step is known. The steps are emitted as one - # contiguous block, which is the unit comments are aligned within. - steps: list[tuple[str, str, str]] = [] - for step in plan: - if isinstance(step, LandStep): - if step.already_landed: - steps.append(("l", f"l {step.pr_number}", "already landed")) - continue - entry = stack[step.entry_index] - steps.append(("l", f"l {entry.pr_number}", entry.title or "")) - elif isinstance(step, WorkflowStep): - steps.append(("w", f"w {step.workflow}", "")) - elif isinstance(step, ConfirmStep): - steps.append(("c", f"c {step.condition}".rstrip(), "")) - - # Only 'l' and 'w' steps decide the column: a confirm condition is free - # text that routinely runs long, and letting it decide would drag every - # comment off to the right. - longest = max((len(line) for kw, line, _ in steps if kw in ("l", "w")), default=0) + # Each step as (aligns comments?, line, trailing comment), so the comments + # can be aligned once the longest step is known. The steps are emitted as + # one contiguous block, which is the unit comments are aligned within. + steps = [(step.aligns_comments, *step.plan_line(stack)) for step in plan] + + # Only some steps decide the column (see AutolandStep.aligns_comments): a + # confirm condition is free text that routinely runs long, and letting it + # decide would drag every comment off to the right. + longest = max((len(line) for aligns, line, _ in steps if aligns), default=0) column = max(PLAN_COMMENT_COLUMN, longest + PLAN_COMMENT_GAP) lines.extend( f"{line}{' ' * max(column - len(line), 1)}# {comment}" if comment else line - for _kw, line, comment in steps + for _aligns, line, comment in steps ) lines.append("") return "\n".join(lines) -# A pinned PR reference: a bare number, or the PR's URL. The owner and repo are -# captured so a URL can be checked against the repo being landed into. Note that -# the '#123' spelling is deliberately unsupported: '#' starts a comment, so -# 'l #123' is indistinguishable from a bare 'l' with a comment after it. -RE_PR_URL = re.compile(r"^https?://[^/]+/([^/]+)/([^/]+)/pull/(\d+)/?$") - - -def _current_owner_repo() -> tuple[str, str]: - """The repository autoland is landing into, as ``(owner, name)``.""" - try: - return gh.github.owner_repo() - except (RuntimeError, KeyError) as e: - raise ValueError(f"Could not determine the current repository: {e}") from e - - -def _parse_pr_ref(ref: str, line_num: int) -> int: - """Parse the argument of an ``l`` step into a PR number. - - A URL must point at the repository being landed into. Everything downstream - — the merge check, the merge queue, the stack itself — is resolved against - that single repo, so a PR elsewhere could not be landed even if its number - were understood; saying so beats silently landing whatever PR happens to - carry the same number here. - """ - m = RE_PR_URL.match(ref) - if m: - owner, repo, number = m.groups() - this_owner, this_repo = _current_owner_repo() - # GitHub treats owner/repo case-insensitively, so a URL that differs - # only in case still points at this repo. - if (owner.lower(), repo.lower()) != (this_owner.lower(), this_repo.lower()): - raise ValueError( - f"Line {line_num}: {ref} points at {owner}/{repo}, but this " - f"stack lands into {this_owner}/{this_repo} — landing PRs " - "across repositories is not currently supported" - ) - return int(number) - if ref.isdigit(): - return int(ref) - raise ValueError( - f"Line {line_num}: 'l' takes a PR number or URL, not {ref!r} " - "(use a bare 'l' to land the next PR in the stack)" - ) - - def _pr_is_merged(pr_number: int) -> bool: """Whether GitHub reports *pr_number* as merged.""" try: @@ -151,14 +94,6 @@ def _pr_is_merged(pr_number: int) -> bool: raise ValueError(f"Could not look up PR #{pr_number} on GitHub: {e}") from e -def _mark_skipped(step: PlanStep) -> None: - """Mark a step as already satisfied by an earlier, partial run.""" - if isinstance(step, WorkflowStep): - step.state = "skipped" - elif isinstance(step, ConfirmStep): - step.confirmed = True - - def parse_plan( text: str, stack: list[StackEntry], @@ -175,10 +110,7 @@ def parse_plan( against GitHub and is injectable for testing. """ steps: list[PlanStep] = [] - # The next stack entry an 'l' step may claim. Landing goes bottom-to-top, - # so live land steps must take stack entries in order, starting at 0. - next_index = 0 - land_steps = 0 + parser = PlanParser(stack=stack, pr_is_merged=pr_is_merged) for line_num, raw_line in enumerate(text.splitlines(), 1): line = raw_line.strip() @@ -187,37 +119,18 @@ def parse_plan( if " #" in line: line = line[: line.index(" #")].strip() - if line == "l" or line.startswith("l "): - ref = line[2:].strip() if line.startswith("l ") else "" - land = _resolve_land_step( - ref, - line_num, - stack=stack, - next_index=next_index, - pr_is_merged=pr_is_merged, - ) - steps.append(land) - land_steps += 1 - # An already-landed step claims no stack entry, so the next live - # 'l' still expects the PR currently at the bottom of the stack. - if not land.already_landed: - next_index += 1 - elif line.startswith("w "): - workflow = line[2:].strip() - if not workflow: - raise ValueError(f"Line {line_num}: 'w' requires a workflow name") - steps.append(WorkflowStep(workflow=workflow)) - elif line == "c" or line.startswith("c "): - # The condition is optional: a bare 'c' just pauses to confirm. - condition = line[2:].strip() if line.startswith("c ") else "" - steps.append(ConfirmStep(condition=condition)) - else: + # A step is its keyword, then (after a space) its argument, if any. + keyword, sep, args = line.partition(" ") + step_type = REGISTRY.for_keyword(keyword) + if step_type is None or (not sep and not step_type.bare_keyword_ok): raise ValueError(f"Line {line_num}: unrecognized step: {raw_line!r}") + parser.line_num = line_num + steps.append(step_type.parse(args.strip(), parser)) # A partial land is allowed (land only the bottom N PRs), but the plan must # land at least one PR. Because 'l' steps take stack entries bottom-to-top, # the landed PRs are always a prefix of the stack; the rest stay open. - if land_steps == 0: + if not any(isinstance(step, LandStep) for step in steps): raise ValueError("Plan has no 'l' steps — nothing to land") # Everything before the last already-landed 'l' step happened in an earlier @@ -226,7 +139,7 @@ def parse_plan( # rather than re-prompting for sign-off on work that already shipped. for index, step in enumerate(steps): if is_assumed_completed(steps, index): - _mark_skipped(step) + step.assume_completed() return steps @@ -242,10 +155,7 @@ def landed_prefix_end(plan: list[PlanStep]) -> int: -1 makes that comparison work out on its own, but it slices from the *end* of a list if it is ever used as a bound directly. """ - return max( - (i for i, s in enumerate(plan) if isinstance(s, LandStep) and s.already_landed), - default=-1, - ) + return max((i for i, s in enumerate(plan) if s.marks_landed_prefix), default=-1) def is_assumed_completed(plan: list[PlanStep], index: int) -> bool: @@ -261,52 +171,21 @@ def is_assumed_completed(plan: list[PlanStep], index: int) -> bool: # Replanning: carry a previous run's progress over to a new plan # --------------------------------------------------------------------------- -# A workflow or confirm step's identity for replanning: its kind, its text, and -# the PRs landed before it. A step means "once these PRs have landed, this -# holds", so a result recorded under the same PRs is still true in a new plan. -_CheckpointKey = tuple[str, str, frozenset[int]] - - -def _land_pr(step: LandStep, stack: list[StackEntry]) -> int | None: - if step.pr_number is not None: - return step.pr_number - return None if step.already_landed else stack[step.entry_index].pr_number - def _checkpoint_keys( plan: list[PlanStep], stack: list[StackEntry] -) -> list[_CheckpointKey | None]: - """Each step's ``_CheckpointKey``, or None for a land step.""" +) -> list[ReplanKey | None]: + """Each step's ``ReplanKey`` (see ``AutolandStep.replan_key``), or None.""" landed: set[int] = set() - keys: list[_CheckpointKey | None] = [] + keys: list[ReplanKey | None] = [] for step in plan: - if isinstance(step, LandStep): - pr = _land_pr(step, stack) - if pr is not None: - landed.add(pr) - keys.append(None) - elif isinstance(step, WorkflowStep): - keys.append(("w", step.workflow, frozenset(landed))) - else: - keys.append(("c", step.condition.strip(), frozenset(landed))) + keys.append(step.replan_key(frozenset(landed))) + pr = step.landed_pr(stack) + if pr is not None: + landed.add(pr) return keys -def _checkpoint_done(step: PlanStep) -> bool: - if isinstance(step, WorkflowStep): - return step.state in ("succeeded", "skipped") - return isinstance(step, ConfirmStep) and step.confirmed - - -def _credit(new: PlanStep, old: PlanStep) -> None: - """Record on *new* the result *old* reached in the previous run.""" - if isinstance(new, WorkflowStep) and isinstance(old, WorkflowStep): - new.state = old.state - new.error_message = "" - elif isinstance(new, ConfirmStep): - new.confirmed = True - - def carry_over_progress( old: LandingContext, plan: list[PlanStep], stack: list[StackEntry] ) -> list[int]: @@ -322,9 +201,9 @@ def carry_over_progress( Returns the indices (into ``old.plan``) of completed checkpoints that found no match in *plan*, so the caller can show what does not carry over. """ - unmatched: dict[_CheckpointKey, list[int]] = {} + unmatched: dict[ReplanKey, list[int]] = {} for index, key in enumerate(_checkpoint_keys(old.plan, old.stack)): - if key is not None and _checkpoint_done(old.plan[index]): + if key is not None and old.plan[index].is_checkpoint_done(): unmatched.setdefault(key, []).append(index) for step, key in zip(plan, _checkpoint_keys(plan, stack)): @@ -333,75 +212,12 @@ def carry_over_progress( previous = old.plan[unmatched[key].pop(0)] # A step parsing already credited (inside the landed prefix) keeps that # credit, but still consumes its match so it isn't reported as lost. - if not _checkpoint_done(step): - _credit(step, previous) + if not step.is_checkpoint_done(): + step.credit_from(previous) return sorted(i for indices in unmatched.values() for i in indices) -def _resolve_land_step( - ref: str, - line_num: int, - *, - stack: list[StackEntry], - next_index: int, - pr_is_merged: Callable[[int], bool], -) -> LandStep: - """Resolve one ``l`` line against the stack. Raises ValueError if it can't.""" - if not ref: - # Bare 'l': take the next PR in the stack, as plans always have. - if next_index >= len(stack): - raise ValueError( - f"Line {line_num}: too many 'l' steps — only {len(stack)} PRs in stack" - ) - return LandStep(entry_index=next_index, pr_number=stack[next_index].pr_number) - - pr_number = _parse_pr_ref(ref, line_num) - index = next( - (i for i, e in enumerate(stack) if e.pr_number == pr_number), - None, - ) - - if index is None: - # Not in the stack: the only benign explanation is that it already - # landed and was rebased away. Confirm that with GitHub rather than - # silently skipping a typo'd or unrelated PR number. - if not pr_is_merged(pr_number): - raise ValueError( - f"Line {line_num}: PR #{pr_number} is not in the stack and has " - "not been merged — check the PR number, or that you are landing " - "the stack this plan was written for" - ) - if next_index: - # Already-landed steps are the completed prefix of a plan, so one - # appearing after a step we still have to land means the stack - # merged out of the order the plan describes. - raise ValueError( - f"Line {line_num}: PR #{pr_number} has already merged, but the " - f"plan lands it after PR #{stack[next_index - 1].pr_number}, " - "which is still open — the stack no longer matches this plan" - ) - return LandStep(entry_index=-1, pr_number=pr_number) - - if index < next_index: - # An earlier 'l' step already claimed this stack entry. Catching this - # here also keeps `stack[next_index]` below in range: every remaining - # case has next_index < index < len(stack). - raise ValueError( - f"Line {line_num}: PR #{pr_number} is already landed by an earlier " - "'l' step — each PR can be landed only once" - ) - - if index != next_index: - expected = stack[next_index].pr_number - raise ValueError( - f"Line {line_num}: plan lands PR #{pr_number} next, but the next PR " - f"in the stack is #{expected} — the stack no longer matches this plan" - ) - - return LandStep(entry_index=index, pr_number=pr_number) - - # The conventional suffix for a landing plan. Plans autoland writes carry it, # and editors key off it (see editors/vscode) to highlight the plan syntax. PLAN_SUFFIX = ".autoland-plan" diff --git a/src/stack_pr/autoland/status.py b/src/stack_pr/autoland/status.py index a573db0..b2f3625 100644 --- a/src/stack_pr/autoland/status.py +++ b/src/stack_pr/autoland/status.py @@ -20,7 +20,7 @@ print_status, ) from stack_pr.autoland.model import ( - _STEP_TYPES, + ConfirmStep, LandingContext, LandStep, PlanStep, @@ -210,7 +210,7 @@ def _print_status_text(report: _StatusReport) -> None: def _step_json(step: PlanStep, row: _StepRow, ctx: LandingContext) -> dict: data: dict[str, Any] = { "number": int(row.number.rstrip(".")), - "type": _STEP_TYPES[type(step)], + "type": step.kind, "outcome": row.outcome.value, "status": row.status, "is_next": row.is_next, @@ -223,6 +223,7 @@ def _step_json(step: PlanStep, row: _StepRow, ctx: LandingContext) -> dict: elif isinstance(step, WorkflowStep): data["workflow"] = step.workflow else: + assert isinstance(step, ConfirmStep) # the only step type left data["condition"] = step.condition data["detail"] = row.detail or None return data diff --git a/src/stack_pr/autoland/steps/__init__.py b/src/stack_pr/autoland/steps/__init__.py new file mode 100644 index 0000000..f738530 --- /dev/null +++ b/src/stack_pr/autoland/steps/__init__.py @@ -0,0 +1,33 @@ +"""Plan step types: the ``AutolandStep`` base class and one module per type. + +Importing this package imports every step module, so each step type is in the +registry before a plan is parsed or a state file is loaded. +""" + +from __future__ import annotations + +from stack_pr.autoland.steps.base import ( + REGISTRY, + AutolandStep, + PlanParser, + ReplanKey, + StepRegistry, + register_step, + step_from_dict, +) +from stack_pr.autoland.steps.confirm import ConfirmStep +from stack_pr.autoland.steps.land import LandStep +from stack_pr.autoland.steps.workflow import WorkflowStep + +__all__ = [ + "REGISTRY", + "AutolandStep", + "ConfirmStep", + "LandStep", + "PlanParser", + "ReplanKey", + "StepRegistry", + "WorkflowStep", + "register_step", + "step_from_dict", +] diff --git a/src/stack_pr/autoland/steps/base.py b/src/stack_pr/autoland/steps/base.py new file mode 100644 index 0000000..f5500a6 --- /dev/null +++ b/src/stack_pr/autoland/steps/base.py @@ -0,0 +1,171 @@ +"""The base class every plan step derives from, and the registry of step types. + +A step type owns everything that differs by kind of step: how it is spelled in +a plan file, how it is stored in the state file, and how it counts toward a +run's progress. Code that walks a plan calls these methods instead of +switching on the step's type, so adding a step type means adding a class here +rather than a branch in every module that handles plans. +""" + +from __future__ import annotations + +import dataclasses +from abc import ABC, abstractmethod +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, ClassVar, TypeVar + +if TYPE_CHECKING: + from stack_pr.autoland.model import StackEntry + +# A workflow or confirm step's identity for replanning: its keyword, its text, +# and the PRs landed before it. A step means "once these PRs have landed, this +# holds", so a result recorded under the same PRs is still true in a new plan. +ReplanKey = tuple[str, str, frozenset[int]] + +_S = TypeVar("_S", bound="AutolandStep") +_T = TypeVar("_T", bound="type[AutolandStep]") + + +@dataclass +class PlanParser: + """What parsing a plan line needs beyond the line itself. + + One parser is shared by every line of a plan, so a step type can keep + state across lines in it (``next_index``, advanced by each live ``l``). + """ + + stack: list[StackEntry] + # Whether GitHub reports a PR as merged; injectable for testing. + pr_is_merged: Callable[[int], bool] + # The next stack entry a positional 'l' step may claim. Landing goes + # bottom-to-top, so live land steps take stack entries in order from 0. + next_index: int = 0 + # The 1-based line being parsed, for error messages. + line_num: int = 0 + + +@dataclass +class AutolandStep(ABC): + """One step of a landing plan. + + Subclasses are dataclasses whose fields are the step's persisted state; + the class attributes below identify the type. + """ + + # The type tag in the state file. + kind: ClassVar[str] + # The keyword that starts the step's line in a plan. + keyword: ClassVar[str] + # The step type's human-readable name. + label: ClassVar[str] + # Whether the step's plan line decides where plan comments are aligned. + aligns_comments: ClassVar[bool] = True + # Whether the keyword alone, with no argument, is a valid plan line. + bare_keyword_ok: ClassVar[bool] = True + + # --- plan syntax --------------------------------------------------------- + + @classmethod + @abstractmethod + def parse(cls: type[_S], args: str, parser: PlanParser) -> _S: + """Build the step from the text after its keyword (stripped; may be + empty). Raises ValueError, naming ``parser.line_num``, if malformed.""" + + @abstractmethod + def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: + """The step as a plan line, and a trailing comment ("" for none).""" + + # --- persistence --------------------------------------------------------- + + def to_dict(self) -> dict[str, Any]: + """The step as stored in the state file: its type tag and its fields.""" + return {"type": self.kind, **dataclasses.asdict(self)} + + @classmethod + def from_dict(cls: type[_S], data: dict[str, Any]) -> _S: + """Restore a step written by ``to_dict``.""" + fields = {k: v for k, v in data.items() if k != "type"} + step: _S = cls(**fields) + return step + + # --- progress ------------------------------------------------------------ + + @property + def marks_landed_prefix(self) -> bool: + """Whether the plan up to and including this step already happened.""" + return False + + def landed_pr(self, stack: list[StackEntry]) -> int | None: + """The PR that has landed once this step is done, if any.""" + return None + + def assume_completed(self) -> None: + """Mark the step as satisfied by an earlier, partial run.""" + + def is_checkpoint_done(self) -> bool: + """Whether the step reached a result a replanned run may reuse.""" + return False + + def replan_key(self, landed_prs: frozenset[int]) -> ReplanKey | None: + """The step's identity for replanning, given the PRs landed before it. + + None for a step whose progress never carries over to a new plan. + """ + return None + + def credit_from(self, old: AutolandStep) -> None: + """Record on this step the result *old*, its match in the previous + run's plan, reached.""" + + +class StepRegistry: + """The step types, looked up by state-file kind or by plan keyword.""" + + def __init__(self) -> None: + self._by_kind: dict[str, type[AutolandStep]] = {} + self._by_keyword: dict[str, type[AutolandStep]] = {} + + def register(self, cls: _T) -> _T: + """Add *cls*. Raises ValueError if its kind or keyword is taken.""" + if cls.kind in self._by_kind: + raise ValueError( + f"Step kind {cls.kind!r} is already registered by " + f"{self._by_kind[cls.kind].__name__}" + ) + if cls.keyword in self._by_keyword: + raise ValueError( + f"Step keyword {cls.keyword!r} is already registered by " + f"{self._by_keyword[cls.keyword].__name__}" + ) + self._by_kind[cls.kind] = cls + self._by_keyword[cls.keyword] = cls + return cls + + def for_kind(self, kind: str) -> type[AutolandStep]: + """The step type stored under *kind*. Raises ValueError if unknown.""" + try: + return self._by_kind[kind] + except KeyError: + raise ValueError(f"Unknown plan step type: {kind!r}") from None + + def for_keyword(self, keyword: str) -> type[AutolandStep] | None: + """The step type spelled *keyword* in a plan, or None if there is none.""" + return self._by_keyword.get(keyword) + + def types(self) -> list[type[AutolandStep]]: + """Every registered step type, in registration order.""" + return list(self._by_kind.values()) + + +REGISTRY = StepRegistry() + + +def register_step(cls: _T) -> _T: + """Class decorator adding a step type to the registry.""" + return REGISTRY.register(cls) + + +def step_from_dict(data: dict[str, Any]) -> AutolandStep: + """Restore a step from the state file. Raises ValueError for an unknown type.""" + return REGISTRY.for_kind(data["type"]).from_dict(data) diff --git a/src/stack_pr/autoland/steps/confirm.py b/src/stack_pr/autoland/steps/confirm.py new file mode 100644 index 0000000..d758218 --- /dev/null +++ b/src/stack_pr/autoland/steps/confirm.py @@ -0,0 +1,58 @@ +"""``c []``: pause for manual confirmation.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from stack_pr.autoland.steps.base import ( + AutolandStep, + PlanParser, + ReplanKey, + register_step, +) + +if TYPE_CHECKING: + from stack_pr.autoland.model import StackEntry + + +@register_step +@dataclass +class ConfirmStep(AutolandStep): + """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. + """ + + kind = "confirm" + keyword = "c" + label = "Confirm" + # A condition is free text that routinely runs long; letting it decide the + # comment column would drag every comment off to the right. + aligns_comments = False + + condition: str = "" + confirmed: bool = False + + @classmethod + def parse(cls, args: str, parser: PlanParser) -> ConfirmStep: + # The condition is optional: a bare 'c' just pauses to confirm. + return cls(condition=args) + + def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: + return f"c {self.condition}".rstrip(), "" + + def assume_completed(self) -> None: + self.confirmed = True + + def is_checkpoint_done(self) -> bool: + return self.confirmed + + def replan_key(self, landed_prs: frozenset[int]) -> ReplanKey | None: + return (self.keyword, self.condition.strip(), landed_prs) + + def credit_from(self, old: AutolandStep) -> None: + self.confirmed = True diff --git a/src/stack_pr/autoland/steps/land.py b/src/stack_pr/autoland/steps/land.py new file mode 100644 index 0000000..dbedbbc --- /dev/null +++ b/src/stack_pr/autoland/steps/land.py @@ -0,0 +1,172 @@ +"""``l []``: land a PR in the stack.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from stack_pr.autoland import gh +from stack_pr.autoland.steps.base import AutolandStep, PlanParser, register_step + +if TYPE_CHECKING: + from stack_pr.autoland.model import StackEntry + +# A pinned PR reference: a bare number, or the PR's URL. The owner and repo are +# captured so a URL can be checked against the repo being landed into. Note that +# the '#123' spelling is deliberately unsupported: '#' starts a comment, so +# 'l #123' is indistinguishable from a bare 'l' with a comment after it. +RE_PR_URL = re.compile(r"^https?://[^/]+/([^/]+)/([^/]+)/pull/(\d+)/?$") + + +@register_step +@dataclass +class LandStep(AutolandStep): + """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``. + """ + + kind = "land" + keyword = "l" + label = "Land" + + entry_index: int + pr_number: int | None = None + + @property + def already_landed(self) -> bool: + return self.entry_index < 0 + + @classmethod + def parse(cls, args: str, parser: PlanParser) -> LandStep: + """Resolve one ``l`` line against the stack. + + An ``l`` step may pin the PR it lands (``l 123`` or ``l ``) + instead of taking the next PR in the stack positionally. A pinned PR + that has already landed is no longer in the stack; such a step + resolves to ``entry_index=-1`` and is skipped at execution time, so one + plan file stays valid as the stack lands piece by piece. + """ + step = _resolve_land_step(args, parser) + # An already-landed step claims no stack entry, so the next live 'l' + # still expects the PR currently at the bottom of the stack. + if not step.already_landed: + parser.next_index += 1 + return step + + def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: + if self.already_landed: + return f"l {self.pr_number}", "already landed" + entry = stack[self.entry_index] + return f"l {entry.pr_number}", entry.title or "" + + @property + def marks_landed_prefix(self) -> bool: + return self.already_landed + + def landed_pr(self, stack: list[StackEntry]) -> int | None: + if self.pr_number is not None: + return self.pr_number + return None if self.already_landed else stack[self.entry_index].pr_number + + +def _current_owner_repo() -> tuple[str, str]: + """The repository autoland is landing into, as ``(owner, name)``.""" + try: + return gh.github.owner_repo() + except (RuntimeError, KeyError) as e: + raise ValueError(f"Could not determine the current repository: {e}") from e + + +def _parse_pr_ref(ref: str, line_num: int) -> int: + """Parse the argument of an ``l`` step into a PR number. + + A URL must point at the repository being landed into. Everything downstream + — the merge check, the merge queue, the stack itself — is resolved against + that single repo, so a PR elsewhere could not be landed even if its number + were understood; saying so beats silently landing whatever PR happens to + carry the same number here. + """ + m = RE_PR_URL.match(ref) + if m: + owner, repo, number = m.groups() + this_owner, this_repo = _current_owner_repo() + # GitHub treats owner/repo case-insensitively, so a URL that differs + # only in case still points at this repo. + if (owner.lower(), repo.lower()) != (this_owner.lower(), this_repo.lower()): + raise ValueError( + f"Line {line_num}: {ref} points at {owner}/{repo}, but this " + f"stack lands into {this_owner}/{this_repo} — landing PRs " + "across repositories is not currently supported" + ) + return int(number) + if ref.isdigit(): + return int(ref) + raise ValueError( + f"Line {line_num}: 'l' takes a PR number or URL, not {ref!r} " + "(use a bare 'l' to land the next PR in the stack)" + ) + + +def _resolve_land_step(ref: str, parser: PlanParser) -> LandStep: + """Resolve one ``l`` line against the stack. Raises ValueError if it can't.""" + stack, next_index, line_num = parser.stack, parser.next_index, parser.line_num + if not ref: + # Bare 'l': take the next PR in the stack, as plans always have. + if next_index >= len(stack): + raise ValueError( + f"Line {line_num}: too many 'l' steps — only {len(stack)} PRs in stack" + ) + return LandStep(entry_index=next_index, pr_number=stack[next_index].pr_number) + + pr_number = _parse_pr_ref(ref, line_num) + index = next( + (i for i, e in enumerate(stack) if e.pr_number == pr_number), + None, + ) + + if index is None: + # Not in the stack: the only benign explanation is that it already + # landed and was rebased away. Confirm that with GitHub rather than + # silently skipping a typo'd or unrelated PR number. + if not parser.pr_is_merged(pr_number): + raise ValueError( + f"Line {line_num}: PR #{pr_number} is not in the stack and has " + "not been merged — check the PR number, or that you are landing " + "the stack this plan was written for" + ) + if next_index: + # Already-landed steps are the completed prefix of a plan, so one + # appearing after a step we still have to land means the stack + # merged out of the order the plan describes. + raise ValueError( + f"Line {line_num}: PR #{pr_number} has already merged, but the " + f"plan lands it after PR #{stack[next_index - 1].pr_number}, " + "which is still open — the stack no longer matches this plan" + ) + return LandStep(entry_index=-1, pr_number=pr_number) + + if index < next_index: + # An earlier 'l' step already claimed this stack entry. Catching this + # here also keeps `stack[next_index]` below in range: every remaining + # case has next_index < index < len(stack). + raise ValueError( + f"Line {line_num}: PR #{pr_number} is already landed by an earlier " + "'l' step — each PR can be landed only once" + ) + + if index != next_index: + expected = stack[next_index].pr_number + raise ValueError( + f"Line {line_num}: plan lands PR #{pr_number} next, but the next PR " + f"in the stack is #{expected} — the stack no longer matches this plan" + ) + + return LandStep(entry_index=index, pr_number=pr_number) diff --git a/src/stack_pr/autoland/steps/workflow.py b/src/stack_pr/autoland/steps/workflow.py new file mode 100644 index 0000000..3b14664 --- /dev/null +++ b/src/stack_pr/autoland/steps/workflow.py @@ -0,0 +1,55 @@ +"""``w ``: wait for a GitHub Actions workflow.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from stack_pr.autoland.steps.base import ( + AutolandStep, + PlanParser, + ReplanKey, + register_step, +) + +if TYPE_CHECKING: + from stack_pr.autoland.model import StackEntry + + +@register_step +@dataclass +class WorkflowStep(AutolandStep): + """Wait for a GitHub Actions workflow to succeed with the landed code.""" + + kind = "workflow" + keyword = "w" + label = "Workflow" + # A bare 'w' names no workflow; it is not a 'w' step at all. + bare_keyword_ok = False + + workflow: str + state: str = "pending" # pending, waiting, succeeded, failed, skipped + error_message: str = "" + + @classmethod + def parse(cls, args: str, parser: PlanParser) -> WorkflowStep: + if not args: + raise ValueError(f"Line {parser.line_num}: 'w' requires a workflow name") + return cls(workflow=args) + + def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: + return f"w {self.workflow}", "" + + def assume_completed(self) -> None: + self.state = "skipped" + + def is_checkpoint_done(self) -> bool: + return self.state in ("succeeded", "skipped") + + def replan_key(self, landed_prs: frozenset[int]) -> ReplanKey | None: + return (self.keyword, self.workflow, landed_prs) + + def credit_from(self, old: AutolandStep) -> None: + if isinstance(old, WorkflowStep): + self.state = old.state + self.error_message = "" diff --git a/tests/test_autoland_steps.py b/tests/test_autoland_steps.py new file mode 100644 index 0000000..7d74359 --- /dev/null +++ b/tests/test_autoland_steps.py @@ -0,0 +1,130 @@ +"""The plan step registry and the per-step-type plan syntax and persistence.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from stack_pr.autoland.model import LandingContext, StackEntry +from stack_pr.autoland.plan import format_plan_for_editor, parse_plan +from stack_pr.autoland.state import AutolandCheckpointer +from stack_pr.autoland.steps import ( + REGISTRY, + AutolandStep, + ConfirmStep, + LandStep, + PlanParser, + StepRegistry, + WorkflowStep, +) + +STACK = [ + StackEntry(pr_url=f"https://github.com/o/r/pull/{n}", pr_number=n, branch=f"b{n}") + for n in (101, 102) +] + +# One freshly parsed instance of every step type, as a plan line would produce. +# Adding a step type without adding it here fails the coverage test below. +SAMPLES: dict[type[AutolandStep], AutolandStep] = { + LandStep: LandStep(entry_index=0, pr_number=101), + WorkflowStep: WorkflowStep(workflow="deploy.yaml"), + ConfirmStep: ConfirmStep(condition="QA sign-off complete"), +} + + +def test_every_registered_step_type_has_a_sample() -> None: + assert set(REGISTRY.types()) == set(SAMPLES) + + +@pytest.mark.parametrize( + ("kind", "keyword", "step_type"), + [ + ("land", "l", LandStep), + ("workflow", "w", WorkflowStep), + ("confirm", "c", ConfirmStep), + ], +) +def test_registry_looks_up_step_types( + kind: str, keyword: str, step_type: type[AutolandStep] +) -> None: + assert REGISTRY.for_kind(kind) is step_type + assert REGISTRY.for_keyword(keyword) is step_type + + +def test_registry_unknown_keyword_is_none() -> None: + assert REGISTRY.for_keyword("z") is None + + +def test_registry_unknown_kind_is_an_error() -> None: + with pytest.raises(ValueError, match="Unknown plan step type: 'bogus'"): + REGISTRY.for_kind("bogus") + + +@dataclass +class _Dummy(AutolandStep): + kind = "dummy" + keyword = "d" + label = "Dummy" + + @classmethod + def parse(cls, args: str, parser: PlanParser) -> _Dummy: + return cls() + + def plan_line(self, stack: list[StackEntry]) -> tuple[str, str]: + return "d", "" + + +@pytest.mark.parametrize( + ("kind", "keyword", "message"), + [ + ("land", "d", "Step kind 'land' is already registered by LandStep"), + ("dummy", "l", "Step keyword 'l' is already registered by LandStep"), + ], +) +def test_registry_rejects_duplicate_registration( + kind: str, keyword: str, message: str +) -> None: + registry = StepRegistry() + registry.register(LandStep) + clash = type("Clash", (_Dummy,), {"kind": kind, "keyword": keyword}) + with pytest.raises(ValueError, match=message): + registry.register(clash) + # A rejected type is not half-registered under its other name. + assert registry.types() == [LandStep] + assert registry.for_keyword("d") is None + + +@pytest.mark.parametrize("step", SAMPLES.values(), ids=lambda s: type(s).__name__) +def test_step_survives_state_file_round_trip( + step: AutolandStep, tmp_path: Path +) -> None: + sf = tmp_path / "state.json" + ctx = LandingContext(stack=STACK, plan=[step]) + AutolandCheckpointer(path=sf, branch="feat", base="main").save(ctx) + + assert json.loads(sf.read_text())["plan"][0]["type"] == step.kind + _, loaded = AutolandCheckpointer.load(sf) + assert loaded.plan == [step] + + +@pytest.mark.parametrize("step", SAMPLES.values(), ids=lambda s: type(s).__name__) +def test_step_survives_plan_text_round_trip(step: AutolandStep) -> None: + # Every plan must land something, so pair a non-land step with a land. + plan = [step] if isinstance(step, LandStep) else [SAMPLES[LandStep], step] + text = format_plan_for_editor(STACK, plan) + assert parse_plan(text, STACK) == plan + + +def test_unknown_step_type_in_state_file_is_an_error(tmp_path: Path) -> None: + sf = tmp_path / "state.json" + ctx = LandingContext(stack=STACK, plan=[SAMPLES[LandStep]]) + AutolandCheckpointer(path=sf, branch="feat", base="main").save(ctx) + data = json.loads(sf.read_text()) + data["plan"].append({"type": "bogus", "condition": ""}) + sf.write_text(json.dumps(data)) + + with pytest.raises(ValueError, match="Unknown plan step type: 'bogus'"): + AutolandCheckpointer.load(sf)