From 5124a512599d597fbff9d093b0f77f7efbb2e290 Mon Sep 17 00:00:00 2001 From: pbean Date: Wed, 22 Jul 2026 15:02:07 -0700 Subject: [PATCH] refactor(engine): extract rollback/preserve cluster into RecoveryFlow (#244) Findings F-3 of the brownfield refactor assessment, PR 2/2 (PR 1 was the worktree cluster, #265). Engine's rollback/preserve cluster is an independent recovery state machine; carve it into a `RecoveryFlow` collaborator built from narrow deps (repo paths, policy, run state, journal, run dir) plus a getter for the engine's swappable active workspace and a few engine callbacks (emit a plugin hook, save state, escalate a task, escalation-pause). The collaborator never receives the whole Engine. Same pattern as WorktreeFlow: Engine keeps same-name delegating private methods (`_rollback_or_pause`, `_safe_reset`, `_restore_patch`, `_prune_preserve_refs`, `_preserve_attempt_*`, `_pause_for_manual_recovery`, `_protected_relpaths`) so test_engine*.py stays byte-untouched and the SweepEngine/StoriesEngine subclasses (which override nothing in this cluster, and call `_rollback_or_pause` /`_safe_reset` through the inherited delegators) are unchanged. `emit` is late-bound so a monkeypatched `Engine._emit` still wins; `RunPaused` is raised via the injected `escalation_pause`/`escalate` callbacks so recovery_flow need not import engine (avoids a runtime<->engine cycle). Behavior-preserving only. New unit tests exercise the collaborator in isolation in tests/test_recovery_flow.py. engine.py shrinks by ~333 lines. The hook-bus cluster is left in place (explicitly out of scope). Closes #244. --- src/bmad_loop/engine.py | 409 +++----------------------- src/bmad_loop/recovery_flow.py | 473 ++++++++++++++++++++++++++++++ tests/test_recovery_flow.py | 510 +++++++++++++++++++++++++++++++++ 3 files changed, 1021 insertions(+), 371 deletions(-) create mode 100644 src/bmad_loop/recovery_flow.py create mode 100644 tests/test_recovery_flow.py diff --git a/src/bmad_loop/engine.py b/src/bmad_loop/engine.py index e927306..5f7b2f7 100644 --- a/src/bmad_loop/engine.py +++ b/src/bmad_loop/engine.py @@ -40,9 +40,10 @@ SessionRecord, StoryTask, ) -from .platform_util import atomic_replace, retrying_unlink, safe_ref_segment, safe_segment +from .platform_util import atomic_replace, retrying_unlink, safe_segment from .plugins import HookBus, HookContext, PluginRegistry from .policy import Policy +from .recovery_flow import RecoveryFlow from .runs import clear_graceful_stop, graceful_stop_requested, kill_session from .sprintstatus import ACTIONABLE_STATUSES from .sprintstatus import advance as sprint_advance @@ -270,6 +271,26 @@ def __init__( workspace_get=lambda: self.workspace, workspace_set=lambda ws: setattr(self, "workspace", ws), ) + # Attempt rollback + recovery-ref preservation flow (issue #244 PR 2/2). + # Same narrow-deps + engine-callbacks pattern as _worktree_flow: `emit` is + # late-bound (a lambda, not the bound method) so a test's monkeypatched + # `_emit` still wins; `workspace_get` reads the engine's worktree-swappable + # active workspace; `escalate` routes an intent-gap restore failure through + # the engine's escalation; `escalation_pause` raises RunPaused for it + # (injected so recovery_flow need not import engine — that would reintroduce + # a runtime<->engine cycle). + self._recovery_flow = RecoveryFlow( + paths=self.paths, + policy=self.policy, + state=self.state, + journal=self.journal, + run_dir=self.run_dir, + workspace_get=lambda: self.workspace, + emit=lambda *a, **k: self._emit(*a, **k), + save=self._save, + escalate=self._escalate, + escalation_pause=self._escalation_pause, + ) def _escalation_pause( self, reason: str, story_key: str = "", *, cause: BaseException | None = None @@ -666,394 +687,40 @@ def _first(epic: int | None): return story return _first(None) + # --------------------------------------- attempt rollback / recovery refs + + # Same-name delegators onto the RecoveryFlow collaborator (issue #244 PR 2/2): + # the rollback/preserve cluster moved to recovery_flow.py; these keep the + # Engine surface (its tests + the SweepEngine/StoriesEngine subclasses) + # byte-compatible, so `self._` calls still resolve here. + def _protected_relpaths(self) -> tuple[str, ...]: - """Repo-relative posix paths of the BMAD artifact folders. These are - orchestrator-owned: never counted as a dev attempt's dirtiness (the - resolve workflow corrects the frozen spec here) and preserved through - rollback. Folders configured outside the repo are skipped — nothing to - protect there.""" - out: list[str] = [] - for protected in ( - self.workspace.paths.output_folder, - self.workspace.paths.implementation_artifacts, - self.workspace.paths.planning_artifacts, - ): - try: - rel = protected.relative_to(self.workspace.root).as_posix() - except ValueError: - continue # configured outside the repo; nothing to protect here - # "." (folder == repo root) as an exclude/keep prefix would cover the - # whole tree — drop it so a misconfig can't disable the dirty check. - if rel and rel != ".": - out.append(rel) - return tuple(out) + return self._recovery_flow.protected_relpaths() def _rollback_or_pause(self, task: StoryTask, *, cause: str = "stopped") -> None: - """Recover from an attempt that won't proceed. - - No-op when the tree is already at the attempt's baseline (nothing this - attempt touched, ignoring orchestrator-owned artifact folders): neither a - reset nor a pause is needed. This is also what lets the manual-recovery - instructions terminate — after the operator resets and resumes, the - now-clean tree skips straight through instead of re-pausing on the - still-set ``baseline_commit``. - - A ``cause="resolved"`` re-drive is human-initiated (the operator ran the - resolve workflow and re-armed the story), so it always auto-recovers and - never pauses, regardless of ``scm.rollback_on_failure``. For the entire - re-drive (``task.resolved_redrive``, latched at resume and cleared once the - correction is committed) the BMAD artifact folders are treated as - orchestrator-owned: excluded from the dirty check (the corrected spec must - not read as a failed attempt) and preserved through every reset — so a - later mid-re-drive retry/defer reset can't silently revert the correction. - - Otherwise (a stopped/abandoned attempt) recovery depends on where the - attempt ran. Inside a mounted unit worktree it always auto-recovers: the - worktree is disposable, the attempt's work is parked on preserve refs - before the reset, and ``scm.rollback_on_failure`` gates *in-place* - (isolation="none") recovery only (#161). In the main checkout the flag - governs: OFF (default) leaves the working tree untouched and emits a bold - manual-recovery notice that pauses the run (stop-and-wait); ON does a - clean reset to baseline. Either way pre-existing untracked files are - preserved; there is no blanket ``git clean``.""" - resolved = cause == "resolved" - # preserve the corrected spec for the whole re-drive, not just the first - # reset; the auto-recover (pause-vs-reset) decision below is unaffected. - redrive = resolved or task.resolved_redrive - protected = self._protected_relpaths() if redrive else () - # Un-determinable dirty check (git timeout/failure, #156) ⇒ assume dirty: - # never skip recovery on an unproven "clean", never crash the run. The - # normal branching below then decides — OFF pauses (worktree kept), ON / - # resolved auto-recovers behind its preserve steps, keeping the re-drive - # pause-free contract intact. - dirty = True - if task.baseline_commit: - try: - dirty = verify.attempt_dirty( - self.workspace.root, - task.baseline_commit, - task.baseline_untracked, - exclude=protected, - ) - except verify.GitError as exc: - self.journal.append( - "rollback-dirty-check-failed", story_key=task.story_key, error=str(exc) - ) - if task.baseline_commit and not dirty: - self.journal.append("rollback-skipped-clean", story_key=task.story_key) - return - # A mounted unit worktree is disposable by design: its branch never - # touches the operator's checkout and the attempt's work is parked on - # recovery refs before any reset. `scm.rollback_on_failure` gates - # *in-place* (isolation="none") recovery only — pausing here would emit - # main-checkout reset instructions for a tree the operator never works - # in (#161). Compared by path, not by policy: a worktree-mode call that - # reaches this method *outside* a mounted unit (e.g. resume with no - # worktree recorded) still targets the main checkout and must pause. - in_unit_worktree = self.workspace.root != self.paths.repo_root - if resolved or in_unit_worktree or self.policy.scm.rollback_on_failure: - self.journal.append( - "rollback-auto", - story_key=task.story_key, - baseline=task.baseline_commit or "", - note="reverting tracked changes + run-created untracked files", - ) - # Give a plugin (the Unity engine) a chance to quiesce before the reset - # rewrites tracked files under it — e.g. save + close open scenes so a - # git reset --hard can't leave a shared Editor showing a run-freezing - # "scene changed on disk" modal. Observe-only, like pre_worktree_teardown: - # the returned ctx is ignored and never routed through _vetoed — a failed - # quiesce must never block a rollback. - self._emit("pre_rollback", task) - # A re-drive (resolved / mid-re-drive) is contractually pause-free, so it - # preserves best-effort but never blocks; a plain rollback pauses rather - # than reset past work it could not park. - self._preserve_attempt_commits(task, allow_pause=not redrive) - # Park the attempt's uncommitted diff too, so the reset below (and its - # untracked cleanup) can't silently destroy in-progress work. Runs only - # if _preserve_attempt_commits did not pause (plain-rollback preserve - # failure); best-effort, never blocks. - self._preserve_attempt_worktree(task) - self._safe_reset(task, preserve=protected) - # Refresh the plugin's view of the now-reset tree (the Unity engine - # re-imports assets). Observe-only for the same reason as pre_rollback. - self._emit("post_rollback", task) - return - self._pause_for_manual_recovery(task, task.baseline_commit or "") - return # unreachable: _pause_for_manual_recovery always raises + self._recovery_flow.rollback_or_pause(task, cause=cause) def _safe_reset(self, task: StoryTask, *, preserve: tuple[str, ...] = ()) -> None: - """Revert tracked changes to the task baseline and remove only the - untracked files this run created — never a blanket `git clean`. Used by - the gated/resolved rollback and by internal ledger recovery (sweep - migration), which restores the orchestrator's own state and must not - pause. The BMAD artifact folders are always kept from untracked deletion; - ``preserve`` (set only on a resolved re-drive) additionally keeps their - *tracked* content alive through the reset, so a just-corrected spec is not - reverted. Sweep passes no ``preserve`` — it wants the broken ledger gone.""" - verify.safe_rollback( - self.workspace.root, - task.baseline_commit or "", - baseline_untracked=task.baseline_untracked, - keep=(".bmad-loop", *self._protected_relpaths()), - preserve=preserve, - ) + self._recovery_flow.safe_reset(task, preserve=preserve) def _restore_patch(self, task: StoryTask) -> None: - """Re-apply the latched intent-gap patch (BMAD-METHOD #2564) onto the - baseline tree so the re-driven session resumes review (step-04) on the - restored diff instead of re-implementing. No-op unless a restore is latched. - - Applied from inside `_dev_phase`'s loop, right before each dispatch that - runs against a fresh baseline (the first attempt and every non-fixable - rollback retry — the loop gates this on ``feedback is None``, so a - fixable-feedback retry that KEEPS the attempt's tree is never double-applied, - and the patch file's own untracked/tracked content is excluded from - `baseline_untracked` because that snapshot is taken before the first apply). - This is the plan's "apply after every baseline reset" seam, placed here - rather than in `_rollback_or_pause` so the patch always lands after the - clean baseline_untracked snapshot — avoiding a mid-re-drive reset preserving - the patch's own new files and then colliding with the re-apply. - - On apply failure we escalate rather than dispatch a session onto a - half-restored tree; the task is mid-dispatch (DEV_RUNNING), so step it to - the escalatable DEV_VERIFY phase first (`_escalate` raises RunPaused).""" - if not task.restore_patch: - return - patch = verify.resolve_restore_path(task.restore_patch, self.workspace.root) - try: - verify.apply_patch(self.workspace.root, patch) - except verify.GitError as e: - self.journal.append( - "attempt-restore-failed", - story_key=task.story_key, - patch=task.restore_patch, - error=str(e), - ) - # Call-site invariant: `_dev_phase` advances the task to DEV_RUNNING - # immediately before dispatch, and this runs on that path only — so the - # step to DEV_VERIFY is unconditional. It is required because `_escalate` - # cannot transition out of DEV_RUNNING directly. - advance(task, Phase.DEV_VERIFY) - self._escalate(task, f"intent-gap restore patch failed to apply: {e}") - self.journal.append("attempt-restored", story_key=task.story_key, patch=task.restore_patch) + self._recovery_flow.restore_patch(task) def _prune_preserve_refs(self) -> None: - """Bounded retention for both recovery-ref families at run start — the - attempt-preserve/* branches and the refs/attempt-preserve-dirty/* - worktree snapshots: keep the newest scm.preserve_keep of each by - committer date, delete the tail (mirrors the runs/cleanup retention - knobs — without it the refs grow unbounded on a long-lived project). - Best-effort: a git failure is journalled per family and never blocks or - pauses the run — the refs are a safety net, not run state — and a - failure in one family never skips the other. preserve_keep = 0 disables - pruning entirely.""" - keep = self.policy.scm.preserve_keep - if keep <= 0: - return - for family, prune in ( - ("attempt-preserve", verify.prune_preserve_refs), - ("attempt-preserve-dirty", verify.prune_preserve_dirty_refs), - ): - try: - deleted = prune(self.workspace.root, keep) - except Exception as exc: # noqa: BLE001 - housekeeping must never crash the - # run: a git timeout/OSError here would otherwise escape to the crash - # handler, so anything beyond the expected GitError is journalled too - # A partial prune (PrunePreserveError) already deleted refs before one - # stuck — that destructive half must stay structurally auditable, not - # buried in the error string. - partial = getattr(exc, "deleted", []) - if partial: - self.journal.append(f"{family}-pruned", count=len(partial), refs=partial) - failed = getattr(exc, "failed", []) - if failed: - self.journal.append(f"{family}-prune-failed", error=str(exc), failed=failed) - else: - self.journal.append(f"{family}-prune-failed", error=str(exc)) - continue - if deleted: - self.journal.append(f"{family}-pruned", count=len(deleted), refs=deleted) + self._recovery_flow.prune_preserve_refs() def _preserve_attempt_commits(self, task: StoryTask, *, allow_pause: bool) -> None: - """Before an auto-rollback's hard reset, park any commits the attempt made - above its baseline under a named recovery ref, so `reset --hard baseline` - can't silently orphan committed work (it survives `git gc` and is - recoverable by name, not just the reflog). No-op when the attempt added no - commits — an uncommitted-only revert is the intended, non-destructive case. - - If commits exist but the ref cannot be created: with ``allow_pause`` (a - plain rollback) refuse to reset — pause for manual recovery rather than - destroy the work. On a re-drive (``allow_pause=False``) the caller's - contract forbids pausing, so journal the failure and let the reset proceed - (the re-drive is a human-directed discard of the failed attempt).""" - baseline = task.baseline_commit - if not baseline: - return - commits = verify.commits_above(self.workspace.root, baseline) - if not commits: - return - head = verify.rev_parse_head(self.workspace.root) # the tip the recovery ref parks at - # run_id can be an arbitrary user `--run-id`; ref-sanitize it (same - # identity-for-clean-ids / digest-for-dirty contract as the unit branches) so - # an exotic/overlong id can't blow the ref-name limit, fail `git branch`, and - # drop the recovery ref (which on a re-drive would then reset past the work - # anyway). - slug = safe_ref_segment(self.state.run_id) - try: - ref = verify.preserve_commits( - self.workspace.root, - baseline, - f"attempt-preserve/{slug}-{head[:8]}", - commits=commits, - ) - except verify.GitError: - ref = None # branch creation failed — treat as a preservation failure - if ref is None: - # commits exist (just enumerated) but the ref did not take. - self.journal.append("attempt-preserve-failed", story_key=task.story_key, head=head) - if allow_pause: - # the commits at HEAD could not be parked — the notice must NOT tell - # the operator to blindly `reset --hard` (that would discard them). - self._pause_for_manual_recovery(task, baseline, preserve_failed=True) - return # re-drive: never pause — proceed to the (human-directed) reset - self.journal.append( - "attempt-commits-preserved", story_key=task.story_key, ref=ref, count=len(commits) - ) + self._recovery_flow.preserve_attempt_commits(task, allow_pause=allow_pause) def _preserve_attempt_worktree(self, task: StoryTask) -> None: - """Before an auto-rollback's hard reset, park the attempt's *uncommitted* - working-tree changes (tracked edits + run-created untracked files) under a - named recovery ref, so `reset --hard baseline` and its untracked cleanup - can't silently destroy in-progress work. Complements - `_preserve_attempt_commits` (which parks *committed* work above baseline); - together they cover the whole attempt. No-op when the tree is clean vs HEAD - — the intended non-destructive uncommitted-revert case. Best-effort: a - capture failure is journaled but never blocks the (human-directed re-drive - or policy-gated) reset — the recovery ref is a safety net, not a gate.""" - baseline = task.baseline_commit - if not baseline: - return - # Same ref-sanitized slug as _preserve_attempt_commits so an exotic/overlong - # --run-id can't blow the ref-name limit and drop the ref. - slug = safe_ref_segment(self.state.run_id) - # ``baseline_commit`` is fixed across the whole dev retry loop, so keying the - # ref on the baseline alone would make a 2nd dirty rollback reuse the name and - # orphan the 1st attempt's snapshot. ``task.attempt`` only ever increments - # (never resets), so it uniquely discriminates each retry's recovery ref. - ref = f"refs/attempt-preserve-dirty/{slug}-{baseline[:8]}-{task.attempt}" - try: - parked = verify.snapshot_worktree( - self.workspace.root, ref, baseline_untracked=task.baseline_untracked - ) - except verify.GitError as exc: - # Keep the git failure detail (commit-tree/update-ref stderr): if the - # following reset destroys work, this is the only breadcrumb explaining - # why the safety-net snapshot couldn't be captured. - self.journal.append( - "attempt-worktree-preserve-failed", story_key=task.story_key, error=str(exc) - ) - return - if parked: - self.journal.append("attempt-worktree-preserved", story_key=task.story_key, ref=parked) + self._recovery_flow.preserve_attempt_worktree(task) def _pause_for_manual_recovery( self, task: StoryTask, baseline: str, *, preserve_failed: bool = False ) -> None: - """Leave the tree untouched, surface bold manual-recovery instructions, and - pause the run. Always raises RunPaused. Three notice shapes: (a, default) - the OFF path for a stopped/abandoned in-place attempt with no commits of - its own — plain manual-rollback steps; (b, ``preserve_failed``) rollback is - ON/resolved but the attempt's commits above baseline could not be parked on - a recovery ref, so an automatic ``reset --hard`` would silently discard - them — a distinct notice that names the at-risk commits and never tells the - operator to blindly reset; (c) the OFF path but the attempt COMMITTED work - above its baseline (#100: a completed session whose run died before the - orchestrator folded the result) — instructing a bare ``reset --hard`` there - would discard finished, possibly already-pushed commits, so this notice - tells the operator to save and check integration state first. A *resolved* - escalation never reaches here — `_rollback_or_pause` auto-recovers that - human-initiated re-drive regardless of `scm.rollback_on_failure`.""" - short = baseline[:12] or "" - # Name the tree every instruction targets. Usually the main checkout, - # but a preserve-failure pause can fire while a unit worktree is - # mounted — a bare "current HEAD" / `git reset --hard` there reads as - # the operator's own checkout (whose HEAD is typically *at* the - # baseline, making the quoted commit range empty) and invites a - # destructive reset of a tree the attempt never touched (#161). - root = self.workspace.root - commits: list[str] = [] - if baseline: - # advisory probe: a git fault here must not block the pause itself - try: - commits = verify.commits_above(root, baseline) - except verify.GitError: - commits = [] - if preserve_failed: - notice = ( - "**ACTION REQUIRED — commits could not be auto-preserved**\n" - f"Story **{task.story_key}**'s attempt committed work above its " - "baseline, but a recovery ref for those commits could not be created, " - "so the automatic rollback was refused rather than `reset --hard` " - "past (and discard) them. **Your commits are intact at the current " - f"HEAD of `{root}`.**\n" - f' 1. **Save them first** — e.g. `git -C "{root}" branch my-rescue ' - f"HEAD` (the commits are `{short}..HEAD` there).\n" - " 2. Only once they are safe, discard the attempt if you want to: " - f'`git -C "{root}" reset --hard {short}`, then review/remove leftover ' - "untracked files.\n" - f"Then run `bmad-loop resume {self.state.run_id}`." - ) - elif commits: - notice = ( - "**ACTION REQUIRED — manual recovery needed (committed work present)**\n" - f"Story **{task.story_key}**'s attempt was stopped with auto-rollback " - "OFF, and it **committed work above its baseline**. **Your commits " - f"are intact at the current HEAD of `{root}`.** They may already be " - "integrated or pushed to a remote — do NOT reset before checking.\n" - f' 1. **Save them first** — e.g. `git -C "{root}" branch my-rescue ' - f"HEAD` (the commits are `{short}..HEAD` there).\n" - " 2. Check whether they are already integrated (merged, pushed to " - "a remote, referenced by open PRs) before discarding anything.\n" - " 3. Only if you decide to discard the attempt: " - f'`git -C "{root}" reset --hard {short}`, then review/remove leftover ' - "untracked files.\n" - f"Then run `bmad-loop resume {self.state.run_id}`." - ) - else: - why = ( - f"Story **{task.story_key}**'s attempt was stopped and auto-rollback " - f"is OFF, so the working tree at `{root}` was left exactly as-is " - "for you to inspect.\n" - ) - notice = ( - "**ACTION REQUIRED — manual rollback needed**\n" - f"{why}" - "To discard this attempt yourself:\n" - " 1. **BACK UP any untracked files you want to keep** — the reset " - "below deletes uncommitted work.\n" - f' 2. `git -C "{root}" reset --hard {short}` then review/remove ' - "leftover untracked files.\n" - " 3. **Restore the files you backed up in step 1.**\n" - f"Then run `bmad-loop resume {self.state.run_id}`. To let the " - "orchestrator do a safe automatic rollback next time, enable " - "`[scm] rollback_on_failure` (it discards the attempt's uncommitted " - "work but never deletes pre-existing untracked files)." - ) - self.journal.append( - "rollback-manual-required", - story_key=task.story_key, - baseline=baseline, - commits=len(commits), + self._recovery_flow.pause_for_manual_recovery( + task, baseline, preserve_failed=preserve_failed ) - gates.notify( - self.policy, - self.run_dir, - f"ACTION REQUIRED: manual rollback for {task.story_key}", - notice, - ) - self._save() - raise RunPaused(notice, PAUSE_ESCALATION, task.story_key) def _finish_inflight(self) -> None: """Complete or roll back tasks interrupted by a pause or crash.""" diff --git a/src/bmad_loop/recovery_flow.py b/src/bmad_loop/recovery_flow.py new file mode 100644 index 0000000..53e5ee5 --- /dev/null +++ b/src/bmad_loop/recovery_flow.py @@ -0,0 +1,473 @@ +"""Attempt rollback + recovery-ref preservation flow. + +Extracted from :class:`bmad_loop.engine.Engine` (issue #244, PR 2/2): the +rollback/preserve cluster is an independent recovery state machine. It lives +here as a collaborator built from narrow dependencies (repo paths, the policy, +run state, journal, run dir) plus a getter for the engine's swappable active +workspace and a handful of engine callbacks (emit a plugin hook, save state, +escalate a task, and escalation-pause). The collaborator never receives the +whole ``Engine`` — it cannot reach engine internals beyond those callables. + +``Engine`` keeps same-name private methods that delegate here, so its tests and +the ``SweepEngine``/``StoriesEngine`` subclasses (which override nothing in this +cluster) see an unchanged surface. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, Callable, NoReturn + +from . import gates, verify +from .model import Phase +from .platform_util import safe_ref_segment +from .statemachine import advance + +if TYPE_CHECKING: + from .bmadconfig import ProjectPaths + from .journal import Journal + from .model import RunState, StoryTask + from .policy import Policy + from .workspace import Workspace + + +class RecoveryFlow: + """Roll back or pause a stopped/abandoned attempt, parking any work it did on + named recovery refs before the reset. + + Built once per engine from narrow deps + engine callbacks (see module + docstring). Behavior is identical to the cluster it was carved out of; the + only structural changes are that engine-owned effects go through injected + callables: ``emit`` fires a plugin hook (late-bound so a monkeypatched + ``Engine._emit`` still wins), ``save`` persists run state, ``escalate`` + routes an intent-gap restore failure through the engine's escalation, and + ``escalation_pause`` raises the engine's ``RunPaused`` (injected so this + module need not import ``engine`` — that would reintroduce a runtime<->engine + import cycle). ``workspace_get`` reads the engine's live (worktree-swappable) + active workspace.""" + + def __init__( + self, + *, + paths: ProjectPaths, + policy: Policy, + state: RunState, + journal: Journal, + run_dir: Path, + workspace_get: Callable[[], Workspace], + emit: Callable[..., object], + save: Callable[[], None], + escalate: Callable[[StoryTask, str], None], + escalation_pause: Callable[..., NoReturn], + ) -> None: + self.paths = paths + self.policy = policy + self.state = state + self.journal = journal + self.run_dir = run_dir + # Read live (a getter, not a captured ref) so a run that swaps the + # engine's `self.workspace` to a mounted unit worktree is seen here. + self._workspace_get = workspace_get + # Injected late-bound so a test patching `engine._emit` still wins + # (recovery_flow's own binding wouldn't). + self._emit = emit + self._save = save + self._escalate = escalate + self._pause = escalation_pause + + def protected_relpaths(self) -> tuple[str, ...]: + """Repo-relative posix paths of the BMAD artifact folders. These are + orchestrator-owned: never counted as a dev attempt's dirtiness (the + resolve workflow corrects the frozen spec here) and preserved through + rollback. Folders configured outside the repo are skipped — nothing to + protect there.""" + workspace = self._workspace_get() + out: list[str] = [] + for protected in ( + workspace.paths.output_folder, + workspace.paths.implementation_artifacts, + workspace.paths.planning_artifacts, + ): + try: + rel = protected.relative_to(workspace.root).as_posix() + except ValueError: + continue # configured outside the repo; nothing to protect here + # "." (folder == repo root) as an exclude/keep prefix would cover the + # whole tree — drop it so a misconfig can't disable the dirty check. + if rel and rel != ".": + out.append(rel) + return tuple(out) + + def rollback_or_pause(self, task: StoryTask, *, cause: str = "stopped") -> None: + """Recover from an attempt that won't proceed. + + No-op when the tree is already at the attempt's baseline (nothing this + attempt touched, ignoring orchestrator-owned artifact folders): neither a + reset nor a pause is needed. This is also what lets the manual-recovery + instructions terminate — after the operator resets and resumes, the + now-clean tree skips straight through instead of re-pausing on the + still-set ``baseline_commit``. + + A ``cause="resolved"`` re-drive is human-initiated (the operator ran the + resolve workflow and re-armed the story), so it always auto-recovers and + never pauses, regardless of ``scm.rollback_on_failure``. For the entire + re-drive (``task.resolved_redrive``, latched at resume and cleared once the + correction is committed) the BMAD artifact folders are treated as + orchestrator-owned: excluded from the dirty check (the corrected spec must + not read as a failed attempt) and preserved through every reset — so a + later mid-re-drive retry/defer reset can't silently revert the correction. + + Otherwise (a stopped/abandoned attempt) recovery depends on where the + attempt ran. Inside a mounted unit worktree it always auto-recovers: the + worktree is disposable, the attempt's work is parked on preserve refs + before the reset, and ``scm.rollback_on_failure`` gates *in-place* + (isolation="none") recovery only (#161). In the main checkout the flag + governs: OFF (default) leaves the working tree untouched and emits a bold + manual-recovery notice that pauses the run (stop-and-wait); ON does a + clean reset to baseline. Either way pre-existing untracked files are + preserved; there is no blanket ``git clean``.""" + workspace = self._workspace_get() + resolved = cause == "resolved" + # preserve the corrected spec for the whole re-drive, not just the first + # reset; the auto-recover (pause-vs-reset) decision below is unaffected. + redrive = resolved or task.resolved_redrive + protected = self.protected_relpaths() if redrive else () + # Un-determinable dirty check (git timeout/failure, #156) ⇒ assume dirty: + # never skip recovery on an unproven "clean", never crash the run. The + # normal branching below then decides — OFF pauses (worktree kept), ON / + # resolved auto-recovers behind its preserve steps, keeping the re-drive + # pause-free contract intact. + dirty = True + if task.baseline_commit: + try: + dirty = verify.attempt_dirty( + workspace.root, + task.baseline_commit, + task.baseline_untracked, + exclude=protected, + ) + except verify.GitError as exc: + self.journal.append( + "rollback-dirty-check-failed", story_key=task.story_key, error=str(exc) + ) + if task.baseline_commit and not dirty: + self.journal.append("rollback-skipped-clean", story_key=task.story_key) + return + # A mounted unit worktree is disposable by design: its branch never + # touches the operator's checkout and the attempt's work is parked on + # recovery refs before any reset. `scm.rollback_on_failure` gates + # *in-place* (isolation="none") recovery only — pausing here would emit + # main-checkout reset instructions for a tree the operator never works + # in (#161). Compared by path, not by policy: a worktree-mode call that + # reaches this method *outside* a mounted unit (e.g. resume with no + # worktree recorded) still targets the main checkout and must pause. + in_unit_worktree = workspace.root != self.paths.repo_root + if resolved or in_unit_worktree or self.policy.scm.rollback_on_failure: + self.journal.append( + "rollback-auto", + story_key=task.story_key, + baseline=task.baseline_commit or "", + note="reverting tracked changes + run-created untracked files", + ) + # Give a plugin (the Unity engine) a chance to quiesce before the reset + # rewrites tracked files under it — e.g. save + close open scenes so a + # git reset --hard can't leave a shared Editor showing a run-freezing + # "scene changed on disk" modal. Observe-only, like pre_worktree_teardown: + # the returned ctx is ignored and never routed through _vetoed — a failed + # quiesce must never block a rollback. + self._emit("pre_rollback", task) + # A re-drive (resolved / mid-re-drive) is contractually pause-free, so it + # preserves best-effort but never blocks; a plain rollback pauses rather + # than reset past work it could not park. + self.preserve_attempt_commits(task, allow_pause=not redrive) + # Park the attempt's uncommitted diff too, so the reset below (and its + # untracked cleanup) can't silently destroy in-progress work. Runs only + # if preserve_attempt_commits did not pause (plain-rollback preserve + # failure); best-effort, never blocks. + self.preserve_attempt_worktree(task) + self.safe_reset(task, preserve=protected) + # Refresh the plugin's view of the now-reset tree (the Unity engine + # re-imports assets). Observe-only for the same reason as pre_rollback. + self._emit("post_rollback", task) + return + self.pause_for_manual_recovery(task, task.baseline_commit or "") + return # unreachable: pause_for_manual_recovery always raises + + def safe_reset(self, task: StoryTask, *, preserve: tuple[str, ...] = ()) -> None: + """Revert tracked changes to the task baseline and remove only the + untracked files this run created — never a blanket `git clean`. Used by + the gated/resolved rollback and by internal ledger recovery (sweep + migration), which restores the orchestrator's own state and must not + pause. The BMAD artifact folders are always kept from untracked deletion; + ``preserve`` (set only on a resolved re-drive) additionally keeps their + *tracked* content alive through the reset, so a just-corrected spec is not + reverted. Sweep passes no ``preserve`` — it wants the broken ledger gone.""" + workspace = self._workspace_get() + verify.safe_rollback( + workspace.root, + task.baseline_commit or "", + baseline_untracked=task.baseline_untracked, + keep=(".bmad-loop", *self.protected_relpaths()), + preserve=preserve, + ) + + def restore_patch(self, task: StoryTask) -> None: + """Re-apply the latched intent-gap patch (BMAD-METHOD #2564) onto the + baseline tree so the re-driven session resumes review (step-04) on the + restored diff instead of re-implementing. No-op unless a restore is latched. + + Applied from inside `_dev_phase`'s loop, right before each dispatch that + runs against a fresh baseline (the first attempt and every non-fixable + rollback retry — the loop gates this on ``feedback is None``, so a + fixable-feedback retry that KEEPS the attempt's tree is never double-applied, + and the patch file's own untracked/tracked content is excluded from + `baseline_untracked` because that snapshot is taken before the first apply). + This is the plan's "apply after every baseline reset" seam, placed here + rather than in `_rollback_or_pause` so the patch always lands after the + clean baseline_untracked snapshot — avoiding a mid-re-drive reset preserving + the patch's own new files and then colliding with the re-apply. + + On apply failure we escalate rather than dispatch a session onto a + half-restored tree; the task is mid-dispatch (DEV_RUNNING), so step it to + the escalatable DEV_VERIFY phase first (`_escalate` raises RunPaused).""" + if not task.restore_patch: + return + workspace = self._workspace_get() + patch = verify.resolve_restore_path(task.restore_patch, workspace.root) + try: + verify.apply_patch(workspace.root, patch) + except verify.GitError as e: + self.journal.append( + "attempt-restore-failed", + story_key=task.story_key, + patch=task.restore_patch, + error=str(e), + ) + # Call-site invariant: `_dev_phase` advances the task to DEV_RUNNING + # immediately before dispatch, and this runs on that path only — so the + # step to DEV_VERIFY is unconditional. It is required because `_escalate` + # cannot transition out of DEV_RUNNING directly. + advance(task, Phase.DEV_VERIFY) + self._escalate(task, f"intent-gap restore patch failed to apply: {e}") + self.journal.append("attempt-restored", story_key=task.story_key, patch=task.restore_patch) + + def prune_preserve_refs(self) -> None: + """Bounded retention for both recovery-ref families at run start — the + attempt-preserve/* branches and the refs/attempt-preserve-dirty/* + worktree snapshots: keep the newest scm.preserve_keep of each by + committer date, delete the tail (mirrors the runs/cleanup retention + knobs — without it the refs grow unbounded on a long-lived project). + Best-effort: a git failure is journalled per family and never blocks or + pauses the run — the refs are a safety net, not run state — and a + failure in one family never skips the other. preserve_keep = 0 disables + pruning entirely.""" + keep = self.policy.scm.preserve_keep + if keep <= 0: + return + workspace = self._workspace_get() + for family, prune in ( + ("attempt-preserve", verify.prune_preserve_refs), + ("attempt-preserve-dirty", verify.prune_preserve_dirty_refs), + ): + try: + deleted = prune(workspace.root, keep) + except Exception as exc: # noqa: BLE001 - housekeeping must never crash the + # run: a git timeout/OSError here would otherwise escape to the crash + # handler, so anything beyond the expected GitError is journalled too + # A partial prune (PrunePreserveError) already deleted refs before one + # stuck — that destructive half must stay structurally auditable, not + # buried in the error string. + partial = getattr(exc, "deleted", []) + if partial: + self.journal.append(f"{family}-pruned", count=len(partial), refs=partial) + failed = getattr(exc, "failed", []) + if failed: + self.journal.append(f"{family}-prune-failed", error=str(exc), failed=failed) + else: + self.journal.append(f"{family}-prune-failed", error=str(exc)) + continue + if deleted: + self.journal.append(f"{family}-pruned", count=len(deleted), refs=deleted) + + def preserve_attempt_commits(self, task: StoryTask, *, allow_pause: bool) -> None: + """Before an auto-rollback's hard reset, park any commits the attempt made + above its baseline under a named recovery ref, so `reset --hard baseline` + can't silently orphan committed work (it survives `git gc` and is + recoverable by name, not just the reflog). No-op when the attempt added no + commits — an uncommitted-only revert is the intended, non-destructive case. + + If commits exist but the ref cannot be created: with ``allow_pause`` (a + plain rollback) refuse to reset — pause for manual recovery rather than + destroy the work. On a re-drive (``allow_pause=False``) the caller's + contract forbids pausing, so journal the failure and let the reset proceed + (the re-drive is a human-directed discard of the failed attempt).""" + baseline = task.baseline_commit + if not baseline: + return + workspace = self._workspace_get() + commits = verify.commits_above(workspace.root, baseline) + if not commits: + return + head = verify.rev_parse_head(workspace.root) # the tip the recovery ref parks at + # run_id can be an arbitrary user `--run-id`; ref-sanitize it (same + # identity-for-clean-ids / digest-for-dirty contract as the unit branches) so + # an exotic/overlong id can't blow the ref-name limit, fail `git branch`, and + # drop the recovery ref (which on a re-drive would then reset past the work + # anyway). + slug = safe_ref_segment(self.state.run_id) + try: + ref = verify.preserve_commits( + workspace.root, + baseline, + f"attempt-preserve/{slug}-{head[:8]}", + commits=commits, + ) + except verify.GitError: + ref = None # branch creation failed — treat as a preservation failure + if ref is None: + # commits exist (just enumerated) but the ref did not take. + self.journal.append("attempt-preserve-failed", story_key=task.story_key, head=head) + if allow_pause: + # the commits at HEAD could not be parked — the notice must NOT tell + # the operator to blindly `reset --hard` (that would discard them). + self.pause_for_manual_recovery(task, baseline, preserve_failed=True) + return # re-drive: never pause — proceed to the (human-directed) reset + self.journal.append( + "attempt-commits-preserved", story_key=task.story_key, ref=ref, count=len(commits) + ) + + def preserve_attempt_worktree(self, task: StoryTask) -> None: + """Before an auto-rollback's hard reset, park the attempt's *uncommitted* + working-tree changes (tracked edits + run-created untracked files) under a + named recovery ref, so `reset --hard baseline` and its untracked cleanup + can't silently destroy in-progress work. Complements + `_preserve_attempt_commits` (which parks *committed* work above baseline); + together they cover the whole attempt. No-op when the tree is clean vs HEAD + — the intended non-destructive uncommitted-revert case. Best-effort: a + capture failure is journaled but never blocks the (human-directed re-drive + or policy-gated) reset — the recovery ref is a safety net, not a gate.""" + baseline = task.baseline_commit + if not baseline: + return + workspace = self._workspace_get() + # Same ref-sanitized slug as preserve_attempt_commits so an exotic/overlong + # --run-id can't blow the ref-name limit and drop the ref. + slug = safe_ref_segment(self.state.run_id) + # ``baseline_commit`` is fixed across the whole dev retry loop, so keying the + # ref on the baseline alone would make a 2nd dirty rollback reuse the name and + # orphan the 1st attempt's snapshot. ``task.attempt`` only ever increments + # (never resets), so it uniquely discriminates each retry's recovery ref. + ref = f"refs/attempt-preserve-dirty/{slug}-{baseline[:8]}-{task.attempt}" + try: + parked = verify.snapshot_worktree( + workspace.root, ref, baseline_untracked=task.baseline_untracked + ) + except verify.GitError as exc: + # Keep the git failure detail (commit-tree/update-ref stderr): if the + # following reset destroys work, this is the only breadcrumb explaining + # why the safety-net snapshot couldn't be captured. + self.journal.append( + "attempt-worktree-preserve-failed", story_key=task.story_key, error=str(exc) + ) + return + if parked: + self.journal.append("attempt-worktree-preserved", story_key=task.story_key, ref=parked) + + def pause_for_manual_recovery( + self, task: StoryTask, baseline: str, *, preserve_failed: bool = False + ) -> None: + """Leave the tree untouched, surface bold manual-recovery instructions, and + pause the run. Always raises RunPaused. Three notice shapes: (a, default) + the OFF path for a stopped/abandoned in-place attempt with no commits of + its own — plain manual-rollback steps; (b, ``preserve_failed``) rollback is + ON/resolved but the attempt's commits above baseline could not be parked on + a recovery ref, so an automatic ``reset --hard`` would silently discard + them — a distinct notice that names the at-risk commits and never tells the + operator to blindly reset; (c) the OFF path but the attempt COMMITTED work + above its baseline (#100: a completed session whose run died before the + orchestrator folded the result) — instructing a bare ``reset --hard`` there + would discard finished, possibly already-pushed commits, so this notice + tells the operator to save and check integration state first. A *resolved* + escalation never reaches here — `_rollback_or_pause` auto-recovers that + human-initiated re-drive regardless of `scm.rollback_on_failure`.""" + workspace = self._workspace_get() + short = baseline[:12] or "" + # Name the tree every instruction targets. Usually the main checkout, + # but a preserve-failure pause can fire while a unit worktree is + # mounted — a bare "current HEAD" / `git reset --hard` there reads as + # the operator's own checkout (whose HEAD is typically *at* the + # baseline, making the quoted commit range empty) and invites a + # destructive reset of a tree the attempt never touched (#161). + root = workspace.root + commits: list[str] = [] + if baseline: + # advisory probe: a git fault here must not block the pause itself + try: + commits = verify.commits_above(root, baseline) + except verify.GitError: + commits = [] + if preserve_failed: + notice = ( + "**ACTION REQUIRED — commits could not be auto-preserved**\n" + f"Story **{task.story_key}**'s attempt committed work above its " + "baseline, but a recovery ref for those commits could not be created, " + "so the automatic rollback was refused rather than `reset --hard` " + "past (and discard) them. **Your commits are intact at the current " + f"HEAD of `{root}`.**\n" + f' 1. **Save them first** — e.g. `git -C "{root}" branch my-rescue ' + f"HEAD` (the commits are `{short}..HEAD` there).\n" + " 2. Only once they are safe, discard the attempt if you want to: " + f'`git -C "{root}" reset --hard {short}`, then review/remove leftover ' + "untracked files.\n" + f"Then run `bmad-loop resume {self.state.run_id}`." + ) + elif commits: + notice = ( + "**ACTION REQUIRED — manual recovery needed (committed work present)**\n" + f"Story **{task.story_key}**'s attempt was stopped with auto-rollback " + "OFF, and it **committed work above its baseline**. **Your commits " + f"are intact at the current HEAD of `{root}`.** They may already be " + "integrated or pushed to a remote — do NOT reset before checking.\n" + f' 1. **Save them first** — e.g. `git -C "{root}" branch my-rescue ' + f"HEAD` (the commits are `{short}..HEAD` there).\n" + " 2. Check whether they are already integrated (merged, pushed to " + "a remote, referenced by open PRs) before discarding anything.\n" + " 3. Only if you decide to discard the attempt: " + f'`git -C "{root}" reset --hard {short}`, then review/remove leftover ' + "untracked files.\n" + f"Then run `bmad-loop resume {self.state.run_id}`." + ) + else: + why = ( + f"Story **{task.story_key}**'s attempt was stopped and auto-rollback " + f"is OFF, so the working tree at `{root}` was left exactly as-is " + "for you to inspect.\n" + ) + notice = ( + "**ACTION REQUIRED — manual rollback needed**\n" + f"{why}" + "To discard this attempt yourself:\n" + " 1. **BACK UP any untracked files you want to keep** — the reset " + "below deletes uncommitted work.\n" + f' 2. `git -C "{root}" reset --hard {short}` then review/remove ' + "leftover untracked files.\n" + " 3. **Restore the files you backed up in step 1.**\n" + f"Then run `bmad-loop resume {self.state.run_id}`. To let the " + "orchestrator do a safe automatic rollback next time, enable " + "`[scm] rollback_on_failure` (it discards the attempt's uncommitted " + "work but never deletes pre-existing untracked files)." + ) + self.journal.append( + "rollback-manual-required", + story_key=task.story_key, + baseline=baseline, + commits=len(commits), + ) + gates.notify( + self.policy, + self.run_dir, + f"ACTION REQUIRED: manual rollback for {task.story_key}", + notice, + ) + self._save() + self._pause(notice, task.story_key) diff --git a/tests/test_recovery_flow.py b/tests/test_recovery_flow.py new file mode 100644 index 0000000..7eb1a19 --- /dev/null +++ b/tests/test_recovery_flow.py @@ -0,0 +1,510 @@ +"""Unit tests for the RecoveryFlow collaborator (issue #244 PR 2/2). + +RecoveryFlow was carved out of Engine's rollback/preserve cluster. These +exercise it in isolation — built from narrow deps + stub engine callbacks, no +Engine instance — which is the point of the extraction. End-to-end behavior +under a real Engine stays covered by test_engine.py. +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest +from conftest import git + +from bmad_loop import verify +from bmad_loop.gates import ATTENTION_FILE +from bmad_loop.model import Phase, StoryTask +from bmad_loop.policy import GatesPolicy, LimitsPolicy, NotifyPolicy, Policy, ScmPolicy +from bmad_loop.recovery_flow import RecoveryFlow +from bmad_loop.verify import GitError, rev_parse_head +from bmad_loop.workspace import Workspace + +QUIET = NotifyPolicy(desktop=False, file=True) + + +def _policy(**scm) -> Policy: + return Policy( + gates=GatesPolicy(mode="none"), + notify=QUIET, + scm=ScmPolicy(**scm), + limits=LimitsPolicy(), + ) + + +class _RecordingJournal: + def __init__(self) -> None: + self.entries: list[tuple[str, dict]] = [] + + def append(self, event: str, **fields) -> None: + self.entries.append((event, fields)) + + def events(self) -> list[str]: + return [e for e, _ in self.entries] + + def fields(self, event: str) -> dict: + for e, f in self.entries: + if e == event: + return f + raise KeyError(event) + + +class _Pause(Exception): + """Stand-in for the engine's RunPaused, raised by the injected escalation_pause + (and the escalate callback) so these tests need not import the engine.""" + + def __init__(self, reason: str, story_key: str) -> None: + super().__init__(reason) + self.reason = reason + self.story_key = story_key + + +def _fake_workspace(root: Path, *, output=None, impl=None, plan=None): + """A duck-typed Workspace for the pure `protected_relpaths` predicate — root + + the three artifact folders, no git required.""" + root = Path(root) + paths = SimpleNamespace( + output_folder=output if output is not None else root / "_bmad-output", + implementation_artifacts=( + impl if impl is not None else root / "_bmad-output" / "implementation-artifacts" + ), + planning_artifacts=( + plan if plan is not None else root / "_bmad-output" / "planning-artifacts" + ), + ) + return SimpleNamespace(root=root, paths=paths) + + +def _make_flow( + *, + workspace, + paths=None, + policy: Policy | None = None, + state=None, + journal: _RecordingJournal | None = None, + run_dir: Path | None = None, +): + """Build a RecoveryFlow wired to recording stubs. The returned flow carries a + ``.calls`` namespace tallying the injected callbacks for assertions. ``paths`` + defaults to the workspace root (so the flow reads as "main checkout"); pass a + different ``repo_root`` to simulate a mounted unit worktree.""" + calls = SimpleNamespace(saves=0, emits=[], pauses=[], escalates=[]) + + def _save() -> None: + calls.saves += 1 + + def _emit(stage, task=None, **fields): + calls.emits.append(stage) + return None + + def _escalate(task, reason) -> None: + calls.escalates.append((task.story_key, reason)) + raise _Pause(reason, task.story_key) + + def _pause(reason, story_key="", *, cause=None): + calls.pauses.append((reason, story_key)) + raise _Pause(reason, story_key) + + flow = RecoveryFlow( + paths=paths if paths is not None else SimpleNamespace(repo_root=workspace.root), + policy=policy if policy is not None else _policy(), + state=state if state is not None else SimpleNamespace(run_id="run-1"), + journal=journal if journal is not None else _RecordingJournal(), + run_dir=run_dir if run_dir is not None else workspace.root, + workspace_get=lambda: workspace, + emit=_emit, + save=_save, + escalate=_escalate, + escalation_pause=_pause, + ) + flow.calls = calls + return flow + + +def _task(repo: Path, story_key: str = "1-1-a") -> StoryTask: + task = StoryTask(story_key=story_key, epic=1) + task.baseline_commit = rev_parse_head(repo) + task.baseline_untracked = [] + return task + + +# --------------------------------------------------------------- protected paths + + +def test_protected_relpaths_lists_bmad_folders(tmp_path): + flow = _make_flow(workspace=_fake_workspace(tmp_path)) + assert flow.protected_relpaths() == ( + "_bmad-output", + "_bmad-output/implementation-artifacts", + "_bmad-output/planning-artifacts", + ) + + +def test_protected_relpaths_skips_folders_outside_repo(tmp_path): + # a planning folder configured outside the repo raises ValueError on + # relative_to and is dropped — nothing to protect there. + outside = tmp_path.parent / "elsewhere" / "planning" + ws = _fake_workspace(tmp_path, plan=outside) + flow = _make_flow(workspace=ws) + assert flow.protected_relpaths() == ( + "_bmad-output", + "_bmad-output/implementation-artifacts", + ) + + +def test_protected_relpaths_drops_repo_root_prefix(tmp_path): + # a folder == repo root would relativize to "." and, used as a keep/exclude + # prefix, cover the whole tree — it must be dropped. + ws = _fake_workspace(tmp_path, output=tmp_path) + flow = _make_flow(workspace=ws) + assert "." not in flow.protected_relpaths() + assert "_bmad-output/implementation-artifacts" in flow.protected_relpaths() + + +# --------------------------------------------------------------- rollback_or_pause + + +def test_rollback_skips_clean_tree(project): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(rollback_on_failure=False)) + task = _task(project.project) # HEAD == baseline, no untracked + + flow.rollback_or_pause(task) # must NOT raise + + assert "rollback-skipped-clean" in flow.journal.events() + assert flow.calls.pauses == [] + assert flow.calls.emits == [] # no pre/post_rollback on the clean short-circuit + + +def test_rollback_auto_resets_when_flag_on(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(rollback_on_failure=True)) + task = _task(repo) + (repo / "src.txt").write_text("committed attempt\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "attempt work") + + flow.rollback_or_pause(task) # must NOT raise + + assert rev_parse_head(repo) == task.baseline_commit # reset to baseline + assert flow.calls.emits == ["pre_rollback", "post_rollback"] + assert flow.calls.pauses == [] + assert "rollback-auto" in flow.journal.events() + + +def test_rollback_off_pauses_and_leaves_tree(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(rollback_on_failure=False)) + task = _task(repo) + (repo / "dirty.txt").write_text("uncommitted work\n") # untracked → dirty + + with pytest.raises(_Pause): + flow.rollback_or_pause(task) + + assert (repo / "dirty.txt").exists() # tree untouched + assert flow.calls.pauses # escalation_pause fired + assert "rollback-manual-required" in flow.journal.events() + + +def test_rollback_in_unit_worktree_auto_recovers_even_when_off(project, tmp_path): + # workspace.root != paths.repo_root ⇒ a mounted unit worktree: rollback OFF + # still auto-recovers (the flag gates in-place recovery only, #161). + repo = project.project + ws = Workspace.default(project) + flow = _make_flow( + workspace=ws, + paths=SimpleNamespace(repo_root=tmp_path / "some-other-main-checkout"), + policy=_policy(rollback_on_failure=False), + ) + task = _task(repo) + (repo / "src.txt").write_text("worktree attempt\n") + + flow.rollback_or_pause(task) # must NOT pause despite rollback OFF + + assert flow.calls.pauses == [] + assert flow.calls.emits == ["pre_rollback", "post_rollback"] + assert "rollback-auto" in flow.journal.events() + + +def test_rollback_resolved_cause_auto_recovers_when_off(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(rollback_on_failure=False)) + task = _task(repo) + (repo / "src.txt").write_text("re-drive edit\n") + + flow.rollback_or_pause(task, cause="resolved") # human-initiated → never pauses + + assert flow.calls.pauses == [] + assert "rollback-auto" in flow.journal.events() + + +def test_rollback_dirty_check_git_fault_degrades_to_dirty(project, monkeypatch): + # #156: an un-determinable dirty check assumes dirty — OFF then pauses rather + # than skip-clean, and never crashes. + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(rollback_on_failure=False)) + task = _task(repo) + + def boom(*a, **k): + raise GitError("git diff timed out") + + monkeypatch.setattr(verify, "attempt_dirty", boom) + + with pytest.raises(_Pause): + flow.rollback_or_pause(task) + + assert "rollback-dirty-check-failed" in flow.journal.events() + assert "rollback-skipped-clean" not in flow.journal.events() + + +# --------------------------------------------------------------- preserve refs + + +def test_preserve_attempt_commits_parks_committed_work(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(repo) + (repo / "src.txt").write_text("committed\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "attempt commit") + + flow.preserve_attempt_commits(task, allow_pause=True) + + assert "attempt-commits-preserved" in flow.journal.events() + ref = flow.journal.fields("attempt-commits-preserved")["ref"] + assert ref.startswith("attempt-preserve/") + git(repo, "rev-parse", "--verify", ref) # the recovery branch exists + + +def test_preserve_attempt_commits_noop_without_commits(project): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(project.project) # HEAD == baseline, nothing committed above it + + flow.preserve_attempt_commits(task, allow_pause=True) + + assert "attempt-commits-preserved" not in flow.journal.events() + assert flow.calls.pauses == [] + + +def test_preserve_attempt_commits_pauses_when_ref_fails_and_allowed(project, monkeypatch): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(repo) + (repo / "src.txt").write_text("committed\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "attempt commit") + + def no_ref(*a, **k): + raise GitError("branch creation failed") + + monkeypatch.setattr(verify, "preserve_commits", no_ref) + + with pytest.raises(_Pause): + flow.preserve_attempt_commits(task, allow_pause=True) + + assert "attempt-preserve-failed" in flow.journal.events() + # preserve_failed=True routes through the committed-work notice + assert "could not be auto-preserved" in flow.calls.pauses[-1][0] + + +def test_preserve_attempt_commits_no_pause_on_redrive(project, monkeypatch): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(repo) + (repo / "src.txt").write_text("committed\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "attempt commit") + + monkeypatch.setattr(verify, "preserve_commits", lambda *a, **k: None) # ref did not take + + flow.preserve_attempt_commits(task, allow_pause=False) # re-drive: never pauses + + assert "attempt-preserve-failed" in flow.journal.events() + assert flow.calls.pauses == [] + + +def test_preserve_attempt_worktree_snapshots_dirty_tree(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(repo) + task.attempt = 2 + (repo / "src.txt").write_text("uncommitted edit\n") # tracked file dirtied + + flow.preserve_attempt_worktree(task) + + assert "attempt-worktree-preserved" in flow.journal.events() + ref = flow.journal.fields("attempt-worktree-preserved")["ref"] + assert ref.startswith("refs/attempt-preserve-dirty/") + git(repo, "rev-parse", "--verify", ref) # the snapshot ref exists + + +# --------------------------------------------------------------- safe_reset + + +def test_safe_reset_reverts_tracked_and_keeps_baseline(project): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(repo) + (repo / "src.txt").write_text("committed then to be reset\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "attempt") + + flow.safe_reset(task) + + assert rev_parse_head(repo) == task.baseline_commit + + +# --------------------------------------------------------------- prune + + +def test_prune_preserve_refs_disabled_when_keep_zero(project): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(preserve_keep=0)) + flow.prune_preserve_refs() + assert flow.journal.events() == [] + + +def test_prune_preserve_refs_journals_deleted_per_family(project, monkeypatch): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(preserve_keep=3)) + monkeypatch.setattr(verify, "prune_preserve_refs", lambda repo, keep: ["a", "b"]) + monkeypatch.setattr(verify, "prune_preserve_dirty_refs", lambda repo, keep: []) + + flow.prune_preserve_refs() + + assert flow.journal.fields("attempt-preserve-pruned")["count"] == 2 + # empty deletion for the other family journals nothing + assert "attempt-preserve-dirty-pruned" not in flow.journal.events() + + +def test_prune_preserve_refs_error_journaled_and_other_family_still_runs(project, monkeypatch): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, policy=_policy(preserve_keep=3)) + + def stuck(repo, keep): + exc = RuntimeError("update-ref stuck") + exc.deleted = ["r1"] # a partial prune already deleted this before stalling + exc.failed = ["r2"] + raise exc + + monkeypatch.setattr(verify, "prune_preserve_refs", stuck) + monkeypatch.setattr(verify, "prune_preserve_dirty_refs", lambda repo, keep: ["d1"]) + + flow.prune_preserve_refs() # a failure in one family must never crash or skip the other + + events = flow.journal.events() + assert "attempt-preserve-pruned" in events # partial deletions stay auditable + assert flow.journal.fields("attempt-preserve-prune-failed")["failed"] == ["r2"] + assert "attempt-preserve-dirty-pruned" in events # the second family still ran + + +# --------------------------------------------------------------- manual recovery + + +def test_pause_stopped_wording_no_commits(project, tmp_path): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, run_dir=tmp_path) + task = _task(project.project) + + with pytest.raises(_Pause) as excinfo: + flow.pause_for_manual_recovery(task, task.baseline_commit) + + reason = excinfo.value.reason + assert "failed" not in reason + assert "manual rollback needed" in reason + assert flow.calls.saves == 1 + assert "rollback-manual-required" in flow.journal.events() + # notify wrote a line to the run dir's attention file (QUIET file=True) + assert "manual rollback for 1-1-a" in (tmp_path / ATTENTION_FILE).read_text() + + +def test_pause_committed_wording_names_at_risk_commits(project, tmp_path): + repo = project.project + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, run_dir=tmp_path) + task = _task(repo) + (repo / "src.txt").write_text("committed work above baseline\n") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "finished but unfolded") + + with pytest.raises(_Pause) as excinfo: + flow.pause_for_manual_recovery(task, task.baseline_commit) + + reason = excinfo.value.reason + assert "committed work above its baseline" in reason + assert "do NOT reset before checking" in reason + + +def test_pause_preserve_failed_wording(project, tmp_path): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws, run_dir=tmp_path) + task = _task(project.project) + + with pytest.raises(_Pause) as excinfo: + flow.pause_for_manual_recovery(task, task.baseline_commit, preserve_failed=True) + + assert "could not be auto-preserved" in excinfo.value.reason + + +# --------------------------------------------------------------- restore_patch + + +def test_restore_patch_noop_without_latch(project): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(project.project) # restore_patch is None by default + + flow.restore_patch(task) + + assert flow.journal.events() == [] + assert flow.calls.escalates == [] + + +def test_restore_patch_applies_and_journals(project, monkeypatch): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(project.project) + task.restore_patch = "patch.diff" + applied = [] + monkeypatch.setattr(verify, "resolve_restore_path", lambda raw, root: Path(root) / raw) + monkeypatch.setattr(verify, "apply_patch", lambda repo, patch: applied.append(patch)) + + flow.restore_patch(task) + + assert applied # the patch was applied + assert "attempt-restored" in flow.journal.events() + assert flow.calls.escalates == [] + + +def test_restore_patch_escalates_on_apply_failure(project, monkeypatch): + ws = Workspace.default(project) + flow = _make_flow(workspace=ws) + task = _task(project.project) + task.restore_patch = "patch.diff" + task.phase = Phase.DEV_RUNNING # the call-site invariant restore_patch relies on + monkeypatch.setattr(verify, "resolve_restore_path", lambda raw, root: Path(root) / raw) + + def boom(repo, patch): + raise GitError("does not apply") + + monkeypatch.setattr(verify, "apply_patch", boom) + + with pytest.raises(_Pause): + flow.restore_patch(task) + + assert task.phase == Phase.DEV_VERIFY # stepped to the escalatable phase first + assert flow.calls.escalates # routed through the engine's escalation + assert "attempt-restore-failed" in flow.journal.events() + assert "attempt-restored" not in flow.journal.events() # never reached on failure