fix(kanban): count only real gc removals and resolve kanban home once

gc ran the full managed-root predicate before checking the dir exists, and
rmtree on a symlinked scratch path silently did nothing (ignore_errors) yet
still bumped the removed count. Check is_dir()/is_symlink() first (cheap,
and most archived rows were already cleaned at completion) and count a
removal only when the path is actually gone.

_managed_scratch_path_info re-resolved the same kanban home once per board
root; resolve it once and pass the real anchor into _add_root.

Co-authored-by: Kyle Caponi <94931731+kylecap9@users.noreply.github.com>
This commit is contained in:
kshitijk4poor
2026-09-27 15:56:53 +05:30
committed by kshitij
parent 7dfdef64bb
commit 818c096033
2 changed files with 23 additions and 8 deletions

View File

@@ -92,32 +92,45 @@ def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
# (resolved root, lexical spellings of the root, board) # (resolved root, lexical spellings of the root, board)
roots: list[tuple[Path, tuple[Path, ...], Optional[str]]] = [] roots: list[tuple[Path, tuple[Path, ...], Optional[str]]] = []
def _add_root(anchor: Path, parts: tuple[str, ...], board: Optional[str]) -> None: def _add_root(
anchor: Path, anchor_real: Path, parts: tuple[str, ...], board: Optional[str]
) -> None:
root = anchor.joinpath(*parts) root = anchor.joinpath(*parts)
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
roots.append(( roots.append((
root.resolve(strict=False), root.resolve(strict=False),
(_lexical_path(root), _lexical_path(anchor.resolve(strict=False).joinpath(*parts))), (_lexical_path(root), _lexical_path(anchor_real.joinpath(*parts))),
board, board,
)) ))
override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip() override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip()
if override: if override:
override_root = Path(override).expanduser() override_root = Path(override).expanduser()
_add_root(override_root.parent, (override_root.name,), None) with contextlib.suppress(OSError):
override_parent = override_root.parent
_add_root(
override_parent,
override_parent.resolve(strict=False),
(override_root.name,),
None,
)
try: try:
home = _kb.kanban_home() home = _kb.kanban_home()
# Resolve the shared anchor once, not once per board root.
home_real = home.resolve(strict=False)
except OSError: except OSError:
home = None home = None
if home is not None: if home is not None:
_add_root(home, ("kanban", "workspaces"), _kb.DEFAULT_BOARD) _add_root(home, home_real, ("kanban", "workspaces"), _kb.DEFAULT_BOARD)
entries: list[Path] = [] entries: list[Path] = []
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
entries = list((home / "kanban" / "boards").resolve(strict=False).iterdir()) entries = list((home / "kanban" / "boards").resolve(strict=False).iterdir())
for entry in entries: for entry in entries:
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
if entry.is_dir(): if entry.is_dir():
_add_root(home, ("kanban", "boards", entry.name, "workspaces"), entry.name) _add_root(
home, home_real, ("kanban", "boards", entry.name, "workspaces"), entry.name
)
for root, lexical_roots, board in roots: for root, lexical_roots, board in roots:
if p_abs == root: if p_abs == root:
continue continue

View File

@@ -331,10 +331,12 @@ def _cmd_gc(args: argparse.Namespace) -> int:
path = Path(row["workspace_path"] or (scratch_root / row["id"])) path = Path(row["workspace_path"] or (scratch_root / row["id"]))
# Same containment predicate as completion cleanup (#28818): strictly below a # Same containment predicate as completion cleanup (#28818): strictly below a
# managed root, never the root itself (which holds every task's scratch dir). # managed root, never the root itself (which holds every task's scratch dir).
if not kbw._is_managed_scratch_path(path): # Cheap existence/symlink check first: most rows were already cleaned at
# completion, and rmtree refuses a symlink (so it must not be counted).
if not path.is_dir() or path.is_symlink() or not kbw._is_managed_scratch_path(path):
continue continue
if path.is_dir(): shutil.rmtree(path, ignore_errors=True)
shutil.rmtree(path, ignore_errors=True) if not path.exists():
removed_ws += 1 removed_ws += 1
removed_events = 0 removed_events = 0