fix(agents): a claude purge never follows a linked dir below its root (Issue #31, Mac review)

A tmp-root entry <key> linking outside made `<key>/<uuid>` look like a plain
directory, and rmtree removed what lay behind the link; projects/<key>,
file-history/ and the other config dirs had no such check either. Each
candidate now keeps the root it was found under (config, projects or tmp),
and is purged only when every component between that root and the leaf is a
real directory and its parent resolves inside the resolved root. A root may
itself be a link (/tmp -> /private/tmp), a leaf link is still removed as the
link alone, and a candidate behind a link is reported as residual instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/skills/lib_py/agents/adapters/claude.py

@@ -1,4 +1,4 @@
-import os, re, json, glob, time, shutil, subprocess
+import os, re, json, glob, stat, time, shutil, subprocess
 from typing import Optional, Any
 from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, conversation_in_use, scan_names, _plausible_id
 from lib_py.verify_session import workspace_key
@@ -91,6 +91,23 @@ def _session_file_live(data: dict, pid) -> bool:
     return lstart <= started + 2
 
 
+def _inside(root: str, path: str) -> bool:
+    """`path` is reached from `root` through real directories only: no
+    component between the two is a symlink, and its parent resolves inside
+    the resolved root. The root may be a link itself (/tmp -> /private/tmp,
+    a linked ~/.claude), and so may `path`: _remove then takes the link alone."""
+    cur = root
+    for part in os.path.relpath(os.path.dirname(path), root).split(os.sep):
+        cur = os.path.join(cur, part)
+        try:
+            if not stat.S_ISDIR(os.lstat(cur).st_mode):
+                return False
+        except OSError:
+            return False
+    real_root = os.path.realpath(root)
+    return os.path.realpath(os.path.dirname(path)).startswith(real_root.rstrip(os.sep) + os.sep)
+
+
 def _remove(path: str) -> None:
     try:
         if os.path.islink(path) or not os.path.isdir(path):
@@ -267,10 +284,12 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
             pass
         return None
 
-    def _owned_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
+    def _candidates(self, uuid: str, ctx: DiscoveryContext) -> list:
-        """Every existing path that belongs to this session alone (Issue #31).
+        """(root, path) of every existing path that belongs to this session
-        Shared files that merely mention it (history.jsonl, ~/.claude.json,
+        alone (Issue #31), with the root (config, projects or tmp) it was
-        backups, paste-cache) are not the session's and are never listed."""
+        found under. Shared files that merely mention it (history.jsonl,
+        ~/.claude.json, backups, paste-cache) are not the session's and are
+        never listed."""
         # Checked before any path is built: whole directories named by the id
         # are removed, so "", "." or ".." would reach file-history/ itself,
         # and "memory" the auto memory. An id that is not a uuid keeps the
@@ -278,30 +297,35 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
         if not _plausible_id(uuid):
             return []
         cfg = _config_dir(ctx)
-        paths = []
+        found = []
         for pdir in _projects_dirs(ctx):
             base = os.path.join(pdir, ctx.ws_key, uuid)
-            paths += [p for p in ((base + ".jsonl", base) if _is_uuid(uuid) else (base + ".jsonl",))
+            found += [(pdir, p) for p in ((base + ".jsonl", base) if _is_uuid(uuid) else (base + ".jsonl",))
                       if os.path.lexists(p)]
         if not _is_uuid(uuid):
-            return paths
+            return found
-        paths += [p for p in (os.path.join(cfg, "file-history", uuid),
+        found += [(cfg, p) for p in (os.path.join(cfg, "file-history", uuid),
-                              os.path.join(cfg, "session-env", uuid)) if os.path.lexists(p)]
+                                     os.path.join(cfg, "session-env", uuid)) if os.path.lexists(p)]
         job = self._owned_job(uuid, ctx)
         if job:
-            paths.append(job)
+            found.append((cfg, job))
         for path, pid, data in self._session_files(uuid, ctx):
             if not _session_file_live(data, pid):
                 # Keys before the session file: the file is what ties them to
                 # this session, so it must outlive them.
-                paths += _key_files(path) + [path]
+                found += [(cfg, p) for p in _key_files(path) + [path]]
         for root in _tmp_roots():
             try:
                 keys = sorted(os.listdir(root))
             except OSError:
                 continue
-            paths.extend(p for p in (os.path.join(root, k, uuid) for k in keys) if os.path.lexists(p))
+            found += [(root, p) for p in (os.path.join(root, k, uuid) for k in keys) if os.path.lexists(p)]
-        return paths
+        return found
+
+    def _owned_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
+        # What lies behind a linked dir below its root is not the session's:
+        # it is kept, and residual_artifacts reports it.
+        return [p for root, p in self._candidates(uuid, ctx) if _inside(root, p)]
 
     def has_artifacts(self, uuid: str, ctx: DiscoveryContext) -> bool:
         # Leftovers count without the transcript (a pre-#31 purge left them).
@@ -342,7 +366,7 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
         files with their (masked) key files. projects is not walked deeper;
         it can be very large."""
         if not _is_uuid(uuid):
-            return self._owned_paths(uuid, ctx)
+            return sorted(p for _, p in self._candidates(uuid, ctx))
         cfg = _config_dir(ctx)
         pdirs = _projects_dirs(ctx)
         found = set(scan_names(cfg, uuid, skip=pdirs))
@@ -360,18 +384,11 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
                         found.update(os.path.join(kdir, n) for n in os.listdir(kdir) if uuid in n)
                     except OSError:
                         pass
-        job = self._owned_job(uuid, ctx)
+        # Every candidate, inside its root or behind a link (never purged).
-        if job:
+        found.update(_label(p) for _, p in self._candidates(uuid, ctx))
-            found.add(job)
         for path, _, _ in self._session_files(uuid, ctx):
             found.add(path)
             found.update(_label(k) for k in _key_files(path))
-        for root in _tmp_roots():
-            try:
-                tkeys = os.listdir(root)
-            except OSError:
-                continue
-            found.update(p for p in (os.path.join(root, k, uuid) for k in tkeys) if os.path.lexists(p))
         return sorted(found)
 
     def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:

tests/test_issue31_purge_residual.py

@@ -290,6 +290,79 @@ def test_claude_rejects_malformed_uuid(home, tmp_path, conv, bad):
     assert memory.exists() and all(p.exists() for p in named)
 
 
+def test_claude_purge_never_follows_a_linked_dir_below_a_root(home, tmp_path, conv):
+    """A symlinked <key> (tmp, projects) or file-history/ below its root is
+    not followed: what lies behind it stays and is reported as residual."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    cfg = home / ".claude"
+    key = workspace_key(str(ws))
+    out = tmp_path / "outside"
+    victims = [out / "tmp" / conv / "keep.txt", out / "proj" / f"{conv}.jsonl", out / "proj" / conv / "keep.txt",
+               out / "fh" / conv / "keep.txt"]
+    for v in victims:
+        v.parent.mkdir(parents=True, exist_ok=True)
+        v.write_text("keep")
+    troot = tmp_path / "ctmp" / f"claude-{UID}"
+    troot.mkdir(parents=True)
+    links = [troot / "linked-workspace", cfg / "projects" / key, cfg / "file-history"]
+    for link, target in zip(links, (out / "tmp", out / "proj", out / "fh")):
+        link.symlink_to(target)
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    assert adapter.purge_artifacts(conv, ctx) == []
+    assert all(v.exists() for v in victims) and all(link.is_symlink() for link in links)
+    assert adapter.residual_artifacts(conv, ctx) == sorted(
+        [str(troot / "linked-workspace" / conv), str(cfg / "projects" / key / f"{conv}.jsonl"),
+         str(cfg / "projects" / key / conv), str(cfg / "file-history" / conv)])
+
+
+def test_claude_purge_removes_only_the_link_at_the_leaf(home, tmp_path, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    key = workspace_key(str(ws))
+    out = tmp_path / "outside"
+    keep = [out / "tmp" / "keep.txt", out / "proj" / "keep.txt"]
+    for k in keep:
+        k.parent.mkdir(parents=True)
+        k.write_text("keep")
+    leaves = [tmp_path / "ctmp" / f"claude-{UID}" / key / conv, home / ".claude" / "projects" / key / conv]
+    for leaf, target in zip(leaves, (out / "tmp", out / "proj")):
+        leaf.parent.mkdir(parents=True)
+        leaf.symlink_to(target)
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    assert set(adapter.purge_artifacts(conv, ctx)) == {str(p) for p in leaves}
+    assert not any(os.path.lexists(p) for p in leaves)
+    assert all(k.exists() for k in keep)
+    assert adapter.residual_artifacts(conv, ctx) == []
+
+
+def test_claude_roots_may_be_links(home, tmp_path, monkeypatch, conv):
+    """A root that is itself a link (/tmp -> /private/tmp on macOS, a linked
+    ~/.claude) is the boundary, not a way out of it."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    real_cfg, real_tmp = tmp_path / "real-cfg", tmp_path / "real-tmp"
+    real_cfg.mkdir()
+    real_tmp.mkdir()
+    (tmp_path / "cfg-link").symlink_to(real_cfg)
+    (tmp_path / "tmp-link").symlink_to(real_tmp)
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cfg-link"))
+    monkeypatch.delenv("CLAUDE_PROJECT_DIR")
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "tmp-link"))
+    t = _tree(home, ws, conv, tmp_path / "tmp-link", cfg=tmp_path / "cfg-link")
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    adapter.purge_artifacts(conv, ctx)
+
+    assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
+    assert adapter.residual_artifacts(conv, ctx) == []
+
+
 def test_a_non_uuid_id_takes_only_its_transcript_file(home, tmp_path):
     """The pre-#31 reach, kept for ids that are not uuids (test_a4 purges
     "uuid-c"): the `<id>.jsonl` file, never a directory of that name."""