Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions src/forge_loop/runner/tick_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -535,6 +535,38 @@ def _inflight_worktrees(cfg: Config) -> set[str]:
return set()


_SYSTEM_ROOTS = {"/", "/tmp", "/var/tmp", "/home"}


def _clamp_overbroad_root(cfg: Config, root, tick: int):
"""#451 — never honor a worktree_root that is a SYSTEM directory.

A live run configured ``worktree_root: /tmp``; the sweep then treated
every /tmp worktree as loop-owned and reaped two OPERATOR worktrees
(one seconds old, uncommitted work destroyed). Roots resolving to /,
/tmp, /var/tmp, /home or $HOME are clamped to the loop's own per-repo
namespace (``worktree_base(repo)``) and a typed event is emitted so
the operator sees the config smell."""
import os
from pathlib import Path as _P

from forge_loop.worker_worktree import worktree_base

resolved = str(_P(root).resolve())
home = os.path.expanduser("~")
if resolved in _SYSTEM_ROOTS or resolved == home:
clamped = worktree_base(cfg.repo)
append_event(
cfg.events_file,
"worktree_root_clamped",
tick=tick,
configured=str(root),
clamped_to=str(clamped),
)
return clamped
return root


def run_worktree_sweep(
cfg: Config,
tick: int,
Expand All @@ -558,6 +590,7 @@ def run_worktree_sweep(
root = getattr(cfg, "worktree_root", None)
if not root:
return None
root = _clamp_overbroad_root(cfg, root, tick)
wts = worktrees if worktrees is not None else _list_worktrees(cfg.repo)
live = (
live_paths
Expand Down
32 changes: 32 additions & 0 deletions tests/test_worktree_sweep_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,3 +204,35 @@ def test_e2e_reaps_intact_stale_agent_worktree(tmp_path: Path) -> None:
assert not wt.exists() # orphan gone
assert (repo / ".git").exists() # checkout intact
assert (repo / "f.txt").read_text() == "x\n"


def test_run_worktree_sweep_clamps_system_temp_root(tmp_path: Path) -> None:
"""#451 — `worktree_root: /tmp` (a system temp dir) must NOT make every
/tmp worktree reapable. A live run reaped two OPERATOR worktrees
seconds after creation because the config declared all of /tmp as the
loop's namespace. Over-broad roots get clamped to the loop's own
per-repo base (worktree_base) and a clamp event is emitted."""
cfg = Config(
repo=tmp_path,
github_repo="o/r",
labels=Labels(),
briefs=Briefs(),
worktree_root=Path("/tmp"), # the footgun config
maintenance_every_n_ticks=5,
)
removed: list[str] = []
report = tc.run_worktree_sweep(
cfg,
tick=5,
worktrees=[
"/tmp/wt-operator", # operator worktree → MUST survive
"/tmp/clone-x", # operator clone-ish dir → MUST survive
],
live_paths=set(),
remove=lambda p: removed.append(p) or True,
)
assert removed == []
assert report is None or report.reaped == []
evts = [e for e in _read_events(cfg) if e["kind"] == "worktree_root_clamped"]
assert len(evts) == 1
assert "/tmp" in evts[0]["configured"]
Loading