Pin the last two guards and keep the refusal stash contract explicit

The focused Fable delta review of cd04248b0 found the head red on one
pre-existing expectation and two guards that no test would miss:

- tests/test_configured_session_prestart.py pins the startup-refusal stash
  exactly; it now expects the `detail` key a refused provision fills (empty
  for a blocked route), the contract fix batch 2 introduced.
- `_deletable` is strict: the snapshot root itself (which holds every live
  snapshot) is never a deletable path; `_remove_paths` and the payload stale
  delete reuse the one guard; a registry row naming the root deletes nothing
  (test).
- The busy-first-section guard is two-sided: the typed refusal must arrive
  after the one lock wait, never after a second discard wait.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
This commit is contained in:
Ouroboros 2026-09-24 07:33:22 +03:00
parent cd04248b02
commit 3a31e955c8
3 changed files with 33 additions and 9 deletions

View file

@ -268,11 +268,17 @@ def _git(repo_dir: Path, *args: str, check: bool = True,
def _deletable(path: Path, root: Path) -> bool:
"""A checkout path this module may delete: non-empty, not a root spelling, strictly
inside the worktree root — the same refusal ``_remove_paths`` applies, because the
registry is durable state and a malformed row must never name an arbitrary path."""
"""A checkout path this module may delete: non-empty, not a root spelling, and
STRICTLY inside the worktree root (the root itself holds every live snapshot) —
the registry is durable state and a malformed row must never name an arbitrary
path. One guard for every delete this module performs."""
text = str(path).strip()
return bool(text) and text not in (".", "/", "//") and _is_within(path, root)
if not text or text in (".", "/", "//"):
return False
try:
return path.resolve() != Path(root).resolve() and _is_within(path, root)
except OSError:
return False
def _git_quiet(repo_dir: Path, *args: str) -> None:
@ -291,10 +297,7 @@ def _remove_paths(repo_dir: Path, wt_path: Path, branch: str, *, allowed_root: O
entry must never cause deletion of an arbitrary filesystem path.
"""
wt_path = Path(wt_path)
wt_text = str(wt_path).strip()
if allowed_root is not None and (
not wt_text or wt_text in (".", "/", "//") or not _is_within(wt_path, Path(allowed_root))
):
if allowed_root is not None and not _deletable(wt_path, Path(allowed_root)):
return
_git_quiet(repo_dir, "worktree", "remove", "--force", str(wt_path))
if wt_path.exists():
@ -1008,7 +1011,7 @@ def provision_payload_snapshot(
wt_path = (root / f"dlgp_{_safe_name(task_id)}_{safe_snap[:16]}").resolve()
_load_registry(data_dir, strict=True, op="provision_payload_snapshot") # refuse before the copy
root.mkdir(parents=True, exist_ok=True)
if wt_path.exists():
if _deletable(wt_path, root) and wt_path.exists():
_force_rmtree(wt_path) # idempotent re-provision of the SAME snapshot id
source_hash = payload_content_hash(target)
env = isolated_git_env()

View file

@ -151,6 +151,7 @@ def test_blocked_session_bootstrap_terminals_unrun_with_alternatives(monkeypatch
"reason": "subscription_window_exhausted",
"reset_at": "2030-01-01T00:00:00Z",
"requested": "harness",
"detail": "", # a blocked route carries no producer sentence; a refused provision does (#1241)
}
availability = task["subagent_availability"]
assert {key: availability[key] for key in (

View file

@ -174,8 +174,12 @@ def test_a_live_holder_is_a_typed_refusal_and_a_dead_holder_is_evicted(tmp_path,
monkeypatch.setattr(wt, "_LOCK_TIMEOUT_SEC", 0.5)
holder = _hold_lock(snaps, 30)
try:
started = time.monotonic()
with pytest.raises(wt.WorktreeOpsLockBusy) as info:
_provision(target, snaps, data)
# A busy FIRST section registered nothing, so nothing is discarded: the typed
# refusal arrives after the one wait, not after a second discard wait.
assert time.monotonic() - started < 3
finally:
holder.kill()
holder.wait()
@ -327,6 +331,22 @@ def test_acting_worktree_add_and_remove_keep_tree_work_outside_the_lock(tmp_path
assert not _lock_held(snaps)
def test_a_row_naming_the_snapshot_root_itself_deletes_nothing(tmp_path):
"""The registry is durable state; a malformed row whose path IS the root must not
wipe every live sibling snapshot (the root holds them all)."""
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"
keep = _provision(target, snaps, data, snapshot_id="keep")
rows = wt._load_registry(data)
rows.append({**rows[0], "snapshot_id": "bad", "path": str(snaps.resolve())})
wt._save_registry(rows, data)
assert wt.remove_execution_snapshot("bad", worktree_root=snaps, data_dir=data)
assert pathlib.Path(keep.path).is_dir() and wt.find_execution_snapshot("keep", data_dir=data) is not None
assert wt.find_execution_snapshot("bad", data_dir=data) is None
assert not wt._deletable(snaps, snaps) and not wt._deletable(pathlib.Path("/"), snaps)
assert wt._deletable(pathlib.Path(keep.path), snaps)
def test_removal_deletes_files_outside_the_lock_and_forgets_metadata_inside(tmp_path, monkeypatch):
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"