fix(recovery): probe for a free attempt-preserve-dirty ref name

src/bmad_loop/recovery_flow.py

@@ -425,10 +425,26 @@ def preserve_attempt_worktree(self, task: StoryTask, *, allow_pause: bool) -> No
         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}"
+        # orphan the 1st attempt's snapshot. ``task.attempt`` discriminates the
+        # retries of one arming but is NOT monotonic across the story's life:
+        # runs.rearm_escalation resets it to 0, and a resolve session that commits
+        # nothing leaves HEAD == baseline, so the post-resolve re-drive's rollback
+        # recomputes the exact {slug}-{baseline}-{attempt} name of the pre-resolve
+        # rollback and would overwrite that snapshot, destroying the only copy of
+        # the first attempt's work. Probe for a free name instead of trusting the
+        # counter: uniqueness is enforced against the refs that actually exist.
+        # The probe runs INSIDE the try: `ref_exists` spawns git, and a timeout or
+        # spawn fault arrives as GitError/GitSpawnError rather than a return code.
+        # Uncaught it would crash the rollback here — the one thing this handler
+        # exists to prevent — so a probe that cannot run degrades into the same
+        # "preservation is observation" path as a snapshot that cannot be written.
+        base_ref = f"refs/attempt-preserve-dirty/{slug}-{baseline[:8]}-{task.attempt}"
+        ref = base_ref
+        serial = 2
         try:
+            while verify.ref_exists(workspace.root, ref):
+                ref = f"{base_ref}-r{serial}"
+                serial += 1
             parked = verify.snapshot_worktree(
                 workspace.root, ref, baseline_untracked=task.baseline_untracked
             )

src/bmad_loop/verify.py

@@ -524,6 +524,21 @@ def snapshot_worktree(
     return ref_name
 
 
+def ref_exists(repo: Path, refname: str) -> bool:
+    """Whether ``refname`` — a FULL refname, e.g. ``refs/attempt-preserve-dirty/…``
+    — currently exists. Sibling of :func:`branch_exists`, which prepends
+    ``refs/heads/`` and so cannot see the snapshot refs that live outside it.
+
+    A non-zero exit reads as "absent" — `show-ref --verify` returns 1 for both a
+    missing ref and a malformed name, and the caller's subsequent ref write
+    surfaces any real error. Spawn and timeout faults are NOT swallowed: they
+    arrive typed from `_run_git` as GitError/GitSpawnError and propagate, so a
+    caller that must not overwrite an existing ref cannot mistake "git could not
+    run" for "the name is free"."""
+    rc, _ = _git(repo, "show-ref", "--verify", "--quiet", refname)
+    return rc == 0
+
+
 def safe_rollback(
     repo: Path,
     baseline: str,

tests/test_engine.py

@@ -4653,6 +4653,52 @@ def test_rollback_preserves_distinct_refs_across_repeated_dirty_rollbacks(projec
     assert git(repo, "show", f"{refs[1]}:src.txt") == "attempt 1 edit"
 
 
+def test_rollback_preserve_ref_unique_when_attempt_counter_repeats(project):
+    """runs.rearm_escalation resets task.attempt to 0, and a resolve session that
+    commits nothing leaves HEAD at the same baseline — so the post-resolve re-drive's
+    rollback recomputes the exact {slug}-{baseline}-{attempt} ref name of the
+    pre-resolve rollback. The engine must probe for a free name instead of trusting
+    the counter: the 2nd rollback may never overwrite the 1st attempt's snapshot."""
+    policy = Policy(
+        gates=GatesPolicy(mode="none"),
+        notify=QUIET,
+        scm=ScmPolicy(rollback_on_failure=True),
+    )
+    engine, _ = make_engine(project, [], policy=policy)
+    repo = project.project
+    task = StoryTask(story_key="1-1-a", epic=1)
+    task.baseline_commit = rev_parse_head(repo)
+    task.baseline_untracked = []
+
+    # pre-resolve rollback: attempt 1 escalates dirty
+    task.attempt = 1
+    (repo / "src.txt").write_text("first arming edit\n")
+    engine._rollback_or_pause(task)
+    assert rev_parse_head(repo) == task.baseline_commit
+
+    # rearm_escalation resets the counter; the re-drive's first retry lands on the
+    # SAME attempt number against the SAME baseline (resolve committed nothing)
+    task.attempt = 1
+    (repo / "src.txt").write_text("re-drive edit\n")
+    engine._rollback_or_pause(task)
+    assert rev_parse_head(repo) == task.baseline_commit
+
+    refs = [e["ref"] for e in engine.journal.entries() if e["kind"] == "attempt-worktree-preserved"]
+    assert len(refs) == 2
+    assert len(set(refs)) == 2  # the colliding name was suffixed, not overwritten
+    assert git(repo, "show", f"{refs[0]}:src.txt") == "first arming edit"
+    assert git(repo, "show", f"{refs[1]}:src.txt") == "re-drive edit"
+    assert refs[1] == f"{refs[0]}-r2"  # deterministic probe-and-suffix shape
+
+    # a third collision keeps escalating the serial instead of clobbering -r2
+    task.attempt = 1
+    (repo / "src.txt").write_text("third edit\n")
+    engine._rollback_or_pause(task)
+    refs = [e["ref"] for e in engine.journal.entries() if e["kind"] == "attempt-worktree-preserved"]
+    assert refs[2] == f"{refs[0]}-r3"
+    assert git(repo, "show", f"{refs[2]}:src.txt") == "third edit"
+
+
 def test_rollback_preserve_ref_slug_survives_a_ref_illegal_run_id(project):
     """A `--run-id` carrying ref-illegal sequences must not drop the recovery ref.
     Characterization for the safe_ref_segment swap — the old inline alnum/`_-` slug

tests/test_recovery_flow.py

@@ -710,6 +710,34 @@ def test_snapshot_oserror_degrades_into_the_typed_path(project, monkeypatch):
     assert (repo / "src.txt").read_text() == "uncommitted work\n"
 
 
+def test_ref_probe_git_fault_degrades_like_a_failed_snapshot(project, monkeypatch):
+    """The free-refname probe spawns git before the snapshot does, so it is the
+    first place a spawn/timeout fault can surface. `ref_exists` deliberately does
+    not swallow those (mistaking "git could not run" for "the name is free" would
+    overwrite the very snapshot the probe exists to protect), so the probe has to
+    sit inside the handler that turns a preservation fault into a pause.
+
+    Ablation target: move the probe loop back above the `try` and this fails with
+    the raw GitSpawnError instead of the typed pause."""
+    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("uncommitted work\n")
+
+    def _fail(*_a, **_k):
+        raise verify.GitSpawnError("git: command not found")
+
+    monkeypatch.setattr(verify, "ref_exists", _fail)
+
+    with pytest.raises(_Pause):  # the typed pause, not the GitSpawnError
+        flow.rollback_or_pause(task)
+
+    entry = flow.journal.fields("attempt-worktree-preserve-failed")
+    assert "command not found" in entry["error"]
+    assert (repo / "src.txt").read_text() == "uncommitted work\n"  # work not reset away
+
+
 def test_notice_probe_oserror_does_not_swallow_the_pause(project, monkeypatch):
     """`pause_for_manual_recovery`'s advisory `commits_above` probe runs while the
     fault that broke the snapshot is still in force, so it is the likeliest place

tests/test_verify.py

@@ -2525,6 +2525,21 @@ def test_snapshot_worktree_survives_reset_and_gc(project):
     assert git(repo, "show", f"{ref}:new_test.txt") == "untracked new file"
 
 
+def test_ref_exists_sees_full_refnames_outside_heads(project):
+    """`ref_exists` must resolve FULL refnames in both families the engine probes:
+    the refs/attempt-preserve-dirty/* snapshots (outside refs/heads/, invisible to
+    `branch_exists`) and ordinary branches given as refs/heads/<name>. Absent refs
+    — and git failures, per the best-effort contract — read as False."""
+    repo = project.project
+    ref = "refs/attempt-preserve-dirty/run-abc12345-1"
+    assert verify.ref_exists(repo, ref) is False
+    git(repo, "update-ref", ref, "HEAD")
+    assert verify.ref_exists(repo, ref) is True
+    assert verify.ref_exists(repo, "refs/heads/main") is True
+    assert verify.ref_exists(repo, "refs/heads/no-such-branch") is False
+    assert verify.ref_exists(repo / "no-such-repo", ref) is False  # git failure -> absent
+
+
 def test_snapshot_worktree_noop_clean_tree(project):
     """A clean tree (identical to HEAD) has nothing uncommitted to park: returns
     None and creates no ref, so a plain reset proceeds unchanged."""