GodopuandClaude Opus 5.5
6e0361cf43
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>
853 lines
35 KiB
Python
853 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""Issue #31 — a claude purge removes every path the session owns and reports
|
|
what still carries the uuid; stop and orc_onboard both go through it.
|
|
|
|
Before #31 the claude adapter removed only `<uuid>.jsonl`. file-history,
|
|
session-env, `projects/<key>/<uuid>/`, the daemon job dir, a stale
|
|
`sessions/<pid>.json` and the tmp scratch dir survived, and with no
|
|
`residual_artifacts` the purge printed `residual: none`. A live `claude -c`
|
|
names no uuid on its command line, so only `sessions/<pid>.json` shows it is
|
|
running: the purge CLI and stop both refuse while that process lives.
|
|
codex, grok and opencode report their leftovers (their purge is unchanged).
|
|
|
|
Every test makes fresh ids: conversation_in_use scans every process on the
|
|
host, so a fixed id here is named by anything carrying this file's text.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid as uuidlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
ONBOARD_SH = SKILLS / "multi-agent-mux-orc-onboard" / "scripts" / "orc_onboard.sh"
|
|
|
|
sys.path.insert(0, str(SKILLS))
|
|
from lib_py.agents.registry import get_adapter # noqa: E402
|
|
from lib_py.agents.base import DiscoveryContext # noqa: E402
|
|
from lib_py.verify_session import workspace_key # noqa: E402
|
|
|
|
SESSION_ENV = ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID",
|
|
"HERMES_SESSION_ID", "GROK_SESSION_ID", "OPENCODE_SESSION_ID")
|
|
UID = os.getuid()
|
|
|
|
|
|
def _fresh_id() -> str:
|
|
ident = str(uuidlib.uuid4())
|
|
out = subprocess.run(["ps", "-axo", "pid=,command="], capture_output=True, text=True, timeout=10).stdout
|
|
assert ident not in out, f"fresh id {ident} is already named by a live process"
|
|
return ident
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def own_tmp_roots(tmp_path, monkeypatch):
|
|
"""Every tmp root the claude adapter may look in stays inside the test:
|
|
TMP/TEMP/TMPDIR and the fixed /tmp fallback, here and in every python a
|
|
test starts (a sitecustomize on PYTHONPATH sets the fallback there)."""
|
|
from lib_py.agents.adapters import claude as claude_mod
|
|
sys_tmp, tmpdir, shim = tmp_path / "systmp", tmp_path / "tmpdir", tmp_path / "pyshim"
|
|
for d in (sys_tmp, tmpdir, shim):
|
|
d.mkdir()
|
|
monkeypatch.setattr(claude_mod, "_SYSTEM_TMP", str(sys_tmp))
|
|
monkeypatch.setenv("TMPDIR", str(tmpdir))
|
|
monkeypatch.delenv("TMP", raising=False)
|
|
monkeypatch.delenv("TEMP", raising=False)
|
|
(shim / "sitecustomize.py").write_text(
|
|
"try:\n"
|
|
" from lib_py.agents.adapters import claude\n"
|
|
f" claude._SYSTEM_TMP = {str(sys_tmp)!r}\n"
|
|
"except Exception:\n"
|
|
" pass\n")
|
|
monkeypatch.setenv("PYTHONPATH", f"{shim}:{os.environ.get('PYTHONPATH', '')}")
|
|
return sys_tmp
|
|
|
|
|
|
@pytest.fixture
|
|
def conv() -> str:
|
|
return _fresh_id()
|
|
|
|
|
|
@pytest.fixture
|
|
def other() -> str:
|
|
return _fresh_id()
|
|
|
|
|
|
@pytest.fixture
|
|
def home(tmp_path, monkeypatch):
|
|
"""A claude home of its own: HOME, projects and the claude tmp root."""
|
|
h = tmp_path / "home"
|
|
(h / ".claude" / "projects").mkdir(parents=True)
|
|
monkeypatch.setenv("HOME", str(h))
|
|
monkeypatch.setenv("HOME_DIR", str(h))
|
|
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(h / ".claude" / "projects"))
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "ctmp"))
|
|
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
|
for k in SESSION_ENV:
|
|
monkeypatch.delenv(k, raising=False)
|
|
return h
|
|
|
|
|
|
def _dead_pid() -> int:
|
|
p = subprocess.Popen(["true"])
|
|
p.wait()
|
|
return p.pid
|
|
|
|
|
|
@pytest.fixture
|
|
def live():
|
|
"""A running process standing in for claude; its session file must be
|
|
written after it started (startedAt >= its lstart)."""
|
|
p = subprocess.Popen(["sleep", "120"])
|
|
time.sleep(0.05)
|
|
yield p
|
|
p.kill()
|
|
p.wait()
|
|
|
|
|
|
def _session_file(cfg: Path, pid: int, sid: str, started_ms=None, key=True) -> list:
|
|
d = cfg / "sessions"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
started_ms = int(time.time() * 1000) if started_ms is None else started_ms
|
|
f = d / f"{pid}.json"
|
|
f.write_text(json.dumps({"pid": pid, "sessionId": sid, "cwd": "/x", "startedAt": started_ms}))
|
|
made = [f]
|
|
if key:
|
|
k = d / f"{pid}.{uuidlib.uuid4().hex}{uuidlib.uuid4().hex}.key"
|
|
k.write_bytes(os.urandom(140))
|
|
made.append(k)
|
|
return made
|
|
|
|
|
|
def _tree(h: Path, ws: Path, sid: str, tmp_root: Path, pid=None, cfg=None) -> dict:
|
|
"""Everything a claude session `sid` owns, as claude 2.1 lays it out
|
|
(`cfg`: a CLAUDE_CONFIG_DIR root instead of <home>/.claude)."""
|
|
cfg = cfg or h / ".claude"
|
|
key = workspace_key(str(ws))
|
|
proj = cfg / "projects" / key
|
|
proj.mkdir(parents=True, exist_ok=True)
|
|
t = {}
|
|
t["jsonl"] = proj / f"{sid}.jsonl"
|
|
t["jsonl"].write_text(json.dumps({"sessionId": sid, "cwd": os.path.realpath(str(ws))}) + "\n")
|
|
t["proj_dir"] = proj / sid
|
|
(t["proj_dir"] / "tool-results").mkdir(parents=True)
|
|
(t["proj_dir"] / "tool-results" / "a.txt").write_text("r")
|
|
(t["proj_dir"] / "subagents").mkdir()
|
|
(t["proj_dir"] / "subagents" / "agent-1.jsonl").write_text("{}\n")
|
|
t["file_history"] = cfg / "file-history" / sid
|
|
t["file_history"].mkdir(parents=True)
|
|
(t["file_history"] / "x@v1").write_text("edited")
|
|
t["session_env"] = cfg / "session-env" / sid
|
|
t["session_env"].mkdir(parents=True)
|
|
t["job"] = cfg / "jobs" / sid[:8]
|
|
t["job"].mkdir(parents=True)
|
|
(t["job"] / "state.json").write_text(json.dumps({"sessionId": sid, "backend": "daemon"}))
|
|
(t["job"] / "timeline.jsonl").write_text("{}\n")
|
|
t["session_json"], t["session_key"] = _session_file(cfg, pid or _dead_pid(), sid)
|
|
t["tmp"] = tmp_root / f"claude-{UID}" / key / sid
|
|
(t["tmp"] / "scratchpad").mkdir(parents=True)
|
|
(t["tmp"] / "tasks").mkdir()
|
|
(t["tmp"] / "tasks" / "b.output").write_text("out")
|
|
return t
|
|
|
|
|
|
def _ctx(ws: Path, h: Path) -> DiscoveryContext:
|
|
return DiscoveryContext(workspace=str(ws), agent_name="claude", home_dir=str(h))
|
|
|
|
|
|
def _masked(p: Path) -> str:
|
|
return os.path.join(str(p.parent), p.name.split(".")[0] + ".*.key")
|
|
|
|
|
|
def _env(extra=None) -> dict:
|
|
env = dict(os.environ)
|
|
for k in SESSION_ENV + ("MAM_ORCHESTRATOR_UUIDS",):
|
|
env.pop(k, None)
|
|
env["PYTHONPATH"] = f"{SKILLS}:{env.get('PYTHONPATH', '')}"
|
|
if extra:
|
|
env.update(extra)
|
|
return env
|
|
|
|
|
|
def _agents_cli(args, cwd) -> subprocess.CompletedProcess:
|
|
return subprocess.run([sys.executable, "-m", "lib_py.agents"] + list(args), capture_output=True,
|
|
text=True, env=_env(), cwd=str(cwd), timeout=60)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# claude adapter
|
|
# ---------------------------------------------------------------------------
|
|
def test_claude_purge_removes_every_owned_path(home, tmp_path, conv):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
|
|
purged = get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
|
|
|
|
assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
|
|
expected = {str(p) for k, p in t.items() if k != "session_key"} | {_masked(t["session_key"])}
|
|
assert set(purged) == expected
|
|
assert len(purged) == len(expected)
|
|
assert not any(t["session_key"].name.split(".")[1] in p for p in purged), "the key's hex is never printed"
|
|
|
|
|
|
def test_claude_purge_leaves_other_sessions_and_shared_files(home, tmp_path, conv, other, live):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
cfg = home / ".claude"
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
o = _tree(home, ws, other, tmp_path / "ctmp")
|
|
o_live = _session_file(cfg, live.pid, other)
|
|
history = cfg / "history.jsonl"
|
|
history.write_bytes((json.dumps({"sessionId": conv, "display": "hi"}) + "\n").encode())
|
|
before = history.read_bytes()
|
|
memory = cfg / "projects" / workspace_key(str(ws)) / "memory" / "MEMORY.md"
|
|
memory.parent.mkdir()
|
|
memory.write_text("m")
|
|
|
|
get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
|
|
|
|
assert all(p.exists() for p in o.values())
|
|
assert all(p.exists() for p in o_live)
|
|
assert history.read_bytes() == before
|
|
assert memory.exists()
|
|
|
|
|
|
def test_claude_purge_keeps_a_job_dir_of_another_session_with_the_same_prefix(home, tmp_path, conv):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
twin = conv[:8] + str(uuidlib.uuid4())[8:]
|
|
(t["job"] / "state.json").write_text(json.dumps({"sessionId": twin}))
|
|
adapter = get_adapter("claude")
|
|
|
|
adapter.purge_artifacts(conv, _ctx(ws, home))
|
|
|
|
assert (t["job"] / "state.json").exists()
|
|
assert str(t["job"]) not in adapter.residual_artifacts(conv, _ctx(ws, home))
|
|
|
|
|
|
def test_claude_residual_lists_leftovers_then_nothing(home, tmp_path, conv, other):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
o = _tree(home, ws, other, tmp_path / "ctmp")
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
before = adapter.residual_artifacts(conv, ctx)
|
|
|
|
names = {str(t[k]) for k in ("jsonl", "proj_dir", "file_history", "session_env", "job", "session_json", "tmp")}
|
|
assert names <= set(before), names - set(before)
|
|
assert not any(other in p for p in before)
|
|
assert not any(str(o["job"]) == p or str(o["session_json"]) == p for p in before)
|
|
|
|
adapter.purge_artifacts(conv, ctx)
|
|
assert adapter.residual_artifacts(conv, ctx) == []
|
|
|
|
# A transcript under another workspace key is not this purge's to remove,
|
|
# but it still carries the uuid.
|
|
stray = home / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
|
|
stray.parent.mkdir()
|
|
stray.write_text("{}\n")
|
|
adapter.purge_artifacts(conv, ctx)
|
|
assert stray.exists()
|
|
assert adapter.residual_artifacts(conv, ctx) == [str(stray)]
|
|
|
|
|
|
@pytest.mark.parametrize("bad", ["", ".", "..", "*", "../x", "a/b", "memory", " ", "notes",
|
|
"12345678-1234-1234-1234-12345678901g", "{conv} ", "x{conv}"])
|
|
def test_claude_rejects_malformed_uuid(home, tmp_path, conv, bad):
|
|
bad = bad.format(conv=conv)
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
# Claude Code's auto memory sits beside the session dirs, under the one
|
|
# non-uuid name there.
|
|
memory = home / ".claude" / "projects" / workspace_key(str(ws)) / "memory" / "MEMORY.md"
|
|
memory.parent.mkdir()
|
|
memory.write_text("m")
|
|
named = [home / ".claude" / d / bad / "x" for d in ("file-history", "session-env")
|
|
if bad.strip(". ") and "/" not in bad and "*" not in bad]
|
|
for p in named:
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text("x")
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.purge_artifacts(bad, ctx) == []
|
|
assert adapter.residual_artifacts(bad, ctx) == []
|
|
assert adapter.has_artifacts(bad, ctx) is False
|
|
assert adapter.conversation_in_use(bad, ctx) is False
|
|
assert all(p.exists() for p in t.values())
|
|
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."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
proj = home / ".claude" / "projects" / workspace_key(str(ws))
|
|
proj.mkdir()
|
|
jsonl = proj / "uuid-c.jsonl"
|
|
jsonl.write_text("{}\n")
|
|
kept = [proj / "uuid-c" / "a", home / ".claude" / "file-history" / "uuid-c" / "b"]
|
|
for p in kept:
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text("x")
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.has_artifacts("uuid-c", ctx) is True
|
|
assert adapter.residual_artifacts("uuid-c", ctx) == [str(jsonl)]
|
|
assert adapter.purge_artifacts("uuid-c", ctx) == [str(jsonl)]
|
|
assert not jsonl.exists() and all(p.exists() for p in kept)
|
|
assert adapter.residual_artifacts("uuid-c", ctx) == []
|
|
|
|
|
|
def test_purge_cli_resolves_claude_from_leftovers_only(home, tmp_path, conv):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
fh = home / ".claude" / "file-history" / conv
|
|
se = home / ".claude" / "session-env" / conv
|
|
fh.mkdir(parents=True)
|
|
se.mkdir(parents=True)
|
|
|
|
res = _agents_cli(["purge", "auto", conv, str(ws)], ws)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
out = json.loads(res.stdout)
|
|
assert out["agent"] == "claude"
|
|
assert set(out["purged"]) == {str(fh), str(se)}
|
|
assert out["residual"] == []
|
|
assert not fh.exists() and not se.exists()
|
|
|
|
|
|
def test_purge_cli_refuses_a_live_claude_without_uuid_on_argv(home, tmp_path, conv, live):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp", pid=live.pid)
|
|
|
|
res = _agents_cli(["purge", "claude", conv, str(ws)], ws)
|
|
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert json.loads(res.stdout)["in_use"] is True
|
|
assert all(p.exists() for p in t.values())
|
|
|
|
|
|
def test_a_reused_pid_does_not_hold_the_conversation(home, tmp_path, conv, live):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
cfg = home / ".claude"
|
|
# The file was written an hour before the process now holding that pid
|
|
# started: the pid was reused, the session is gone.
|
|
made = _session_file(cfg, live.pid, conv, started_ms=int((time.time() - 3600) * 1000))
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.conversation_in_use(conv, ctx) is False
|
|
purged = adapter.purge_artifacts(conv, ctx)
|
|
assert not any(p.exists() for p in made)
|
|
assert set(purged) == {str(made[0]), _masked(made[1])}
|
|
|
|
|
|
def test_session_file_liveness_fails_safe(monkeypatch, live):
|
|
from lib_py.agents.adapters import claude as claude_mod
|
|
now_ms = int(time.time() * 1000)
|
|
assert claude_mod._session_file_live({"startedAt": now_ms}, live.pid) is True
|
|
assert claude_mod._session_file_live({}, live.pid) is True, "no startedAt: cannot tell, so live"
|
|
assert claude_mod._session_file_live({"startedAt": now_ms}, _dead_pid()) is False
|
|
assert claude_mod._session_file_live({"startedAt": now_ms}, None) is True
|
|
assert claude_mod._session_file_live({"startedAt": now_ms}, 0) is True
|
|
assert claude_mod._session_file_live({"startedAt": 0}, live.pid) is True, "no usable startedAt: live"
|
|
hour_ago = int((time.time() - 3600) * 1000)
|
|
assert claude_mod._session_file_live({"startedAt": hour_ago}, live.pid) is False, "reused pid"
|
|
monkeypatch.setenv("PATH", "/nonexistent")
|
|
assert claude_mod._session_file_live({"startedAt": hour_ago}, live.pid) is True, "ps unavailable: live"
|
|
|
|
|
|
def test_claude_purge_takes_the_stale_key_with_its_session_file(home, tmp_path, conv, other, live):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
cfg = home / ".claude"
|
|
stale = _session_file(cfg, _dead_pid(), conv)
|
|
held = _session_file(cfg, live.pid, other)
|
|
lone_key = cfg / "sessions" / f"{_dead_pid()}.{'a' * 64}.key"
|
|
lone_key.write_bytes(b"k")
|
|
|
|
purged = get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
|
|
|
|
assert not any(p.exists() for p in stale)
|
|
assert all(p.exists() for p in held)
|
|
assert lone_key.exists(), "a key with no session file of this conversation is not touched"
|
|
assert set(purged) == {str(stale[0]), _masked(stale[1])}
|
|
|
|
|
|
@pytest.mark.parametrize("project_override", [False, True])
|
|
def test_claude_config_dir_moves_the_config_root(home, tmp_path, monkeypatch, conv, live, project_override):
|
|
"""CLAUDE_CONFIG_DIR moves sessions/, file-history/, ... and projects/:
|
|
a live session there is refused, a stopped one is cleaned up. With
|
|
CLAUDE_PROJECT_DIR naming the same projects/ the result is the same."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
cfg = tmp_path / "cfg"
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(cfg))
|
|
if project_override:
|
|
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(cfg / "projects"))
|
|
else:
|
|
monkeypatch.delenv("CLAUDE_PROJECT_DIR")
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp", pid=live.pid, cfg=cfg)
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.conversation_in_use(conv, ctx) is True
|
|
res = _agents_cli(["purge", "claude", conv, str(ws)], ws)
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert all(p.exists() for p in t.values())
|
|
|
|
live.kill()
|
|
live.wait()
|
|
purged = 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 str(t["file_history"]) in purged and str(t["jsonl"]) in purged
|
|
assert adapter.residual_artifacts(conv, ctx) == []
|
|
assert not (home / ".claude" / "file-history").exists(), "the default root is not used"
|
|
|
|
|
|
def _failing_remove(monkeypatch, victim: Path):
|
|
from lib_py.agents.adapters import claude as claude_mod
|
|
real = os.remove
|
|
|
|
def remove(path, *a, **kw):
|
|
if os.path.realpath(path) == os.path.realpath(str(victim)):
|
|
raise PermissionError(13, "denied", str(path))
|
|
return real(path, *a, **kw)
|
|
monkeypatch.setattr(claude_mod.os, "remove", remove)
|
|
|
|
|
|
def test_a_key_that_cannot_be_removed_keeps_its_session_file(home, tmp_path, monkeypatch, conv):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
sfile, key = _session_file(home / ".claude", _dead_pid(), conv)
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
with monkeypatch.context() as m:
|
|
_failing_remove(m, key)
|
|
purged = adapter.purge_artifacts(conv, ctx)
|
|
|
|
assert key.exists() and sfile.exists(), "the session file is what ties the key to this session"
|
|
assert purged == []
|
|
assert adapter.residual_artifacts(conv, ctx) == sorted([str(sfile), _masked(key)])
|
|
|
|
assert set(adapter.purge_artifacts(conv, ctx)) == {_masked(key), str(sfile)}
|
|
assert adapter.residual_artifacts(conv, ctx) == []
|
|
|
|
|
|
def test_a_job_dir_that_cannot_be_emptied_keeps_its_state_json(home, tmp_path, monkeypatch, conv):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp")
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
with monkeypatch.context() as m:
|
|
_failing_remove(m, t["job"] / "timeline.jsonl")
|
|
purged = adapter.purge_artifacts(conv, ctx)
|
|
|
|
assert (t["job"] / "state.json").exists(), "state.json proves the dir is this session's"
|
|
assert str(t["job"]) not in purged
|
|
assert not t["file_history"].exists()
|
|
assert adapter.residual_artifacts(conv, ctx) == [str(t["job"])]
|
|
|
|
assert adapter.purge_artifacts(conv, ctx) == [str(t["job"])]
|
|
assert not t["job"].exists()
|
|
assert adapter.residual_artifacts(conv, ctx) == []
|
|
|
|
|
|
def test_claude_tmp_root_rules(tmp_path, monkeypatch, own_tmp_roots):
|
|
from lib_py.agents.adapters.claude import _tmp_roots
|
|
name = f"claude-{UID}"
|
|
x, y, z = tmp_path / "x", tmp_path / "y", tmp_path / "z"
|
|
for d in (x / name, y / name, z / name, own_tmp_roots / name):
|
|
d.mkdir(parents=True)
|
|
real = lambda roots: [os.path.realpath(r) for r in roots] # noqa: E731
|
|
sys_tmp = [os.path.realpath(str(own_tmp_roots / name))]
|
|
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(x))
|
|
monkeypatch.setenv("TMPDIR", str(z) + "/")
|
|
assert real(_tmp_roots()) == [str(x / name), str(z / name)] + sys_tmp
|
|
|
|
# A sandboxed tool shell: CLAUDE_CODE_TMPDIR is the per-uid root itself.
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(y / name))
|
|
assert real(_tmp_roots())[0] == str(y / name)
|
|
|
|
# Missing candidates are dropped, duplicates collapse.
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(z))
|
|
monkeypatch.delenv("TMPDIR", raising=False)
|
|
monkeypatch.delenv("TMP", raising=False)
|
|
monkeypatch.setenv("TEMP", str(z))
|
|
assert real(_tmp_roots()) == [str(z / name)] + sys_tmp
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "absent"))
|
|
assert real(_tmp_roots()) == [str(z / name)] + sys_tmp
|
|
|
|
|
|
def test_tests_never_reach_the_real_tmp_root(tmp_path, own_tmp_roots):
|
|
from lib_py.agents.adapters.claude import _tmp_roots
|
|
assert all(os.path.realpath(r).startswith(os.path.realpath(str(tmp_path))) for r in _tmp_roots())
|
|
res = subprocess.run([sys.executable, "-c", "from lib_py.agents.adapters import claude; print(claude._SYSTEM_TMP)"],
|
|
capture_output=True, text=True, env=_env(), timeout=30)
|
|
assert res.stdout.strip() == str(own_tmp_roots), res.stdout + res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# in-use CLI and the stop gate
|
|
# ---------------------------------------------------------------------------
|
|
def test_in_use_cli_waits_then_answers(home, tmp_path, conv, live):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
|
|
t0 = time.monotonic()
|
|
res = _agents_cli(["in-use", "claude", conv, str(ws), "--wait", "1"], ws)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert json.loads(res.stdout) == {"agent": "claude", "in_use": False}
|
|
assert time.monotonic() - t0 < 5
|
|
|
|
_session_file(home / ".claude", live.pid, conv)
|
|
t0 = time.monotonic()
|
|
res = _agents_cli(["in-use", "claude", conv, str(ws), "--wait", "1"], ws)
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert json.loads(res.stdout)["in_use"] is True
|
|
assert time.monotonic() - t0 >= 1
|
|
|
|
assert _agents_cli(["in-use", "nosuchagent", conv, str(ws)], ws).returncode == 1
|
|
assert _agents_cli(["in-use", "claude"], ws).returncode == 1
|
|
|
|
|
|
def _stop_row(sandbox: Path, name: str, sid: str):
|
|
lib = sandbox / ".agents" / "skills" / "lib.sh"
|
|
yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
|
|
mutation = (f"d['herdr_sessions'] = [{{'name': {name!r}, 'status': 'running', 'agent': 'claude', "
|
|
f"'claude_session_id_own': {sid!r}, 'pane': {{'cwd': {os.path.realpath(str(sandbox))!r}}}}}]\n")
|
|
res = subprocess.run(["bash", "-c", f"source {lib} && atomic_dump_yaml {yaml_path}"], input=mutation,
|
|
capture_output=True, text=True, env=_env())
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
|
|
def _stop(sandbox: Path, name: str) -> subprocess.CompletedProcess:
|
|
script = sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
return subprocess.run(["bash", str(script), "--session", name, "--agent", "claude",
|
|
"--purge-conversation", "--yes"], capture_output=True, text=True,
|
|
env=_env(), cwd=str(sandbox), timeout=120)
|
|
|
|
|
|
def _row_left(sandbox: Path, name: str) -> tuple:
|
|
in_yaml = name in (sandbox / ".mam" / "agent-sessions.yaml").read_text()
|
|
conn = sqlite3.connect(str(sandbox / ".mam" / "agent-sessions.db"))
|
|
try:
|
|
in_db = conn.execute("SELECT COUNT(*) FROM sessions WHERE name=?", (name,)).fetchone()[0] > 0
|
|
finally:
|
|
conn.close()
|
|
return in_yaml, in_db
|
|
|
|
|
|
@pytest.fixture
|
|
def stop_box(mam_sandbox, mock_herdr, monkeypatch):
|
|
monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(mam_sandbox / "ctmp"))
|
|
return mam_sandbox
|
|
|
|
|
|
def test_stop_purge_reports_residual_and_removes_the_row(stop_box, conv):
|
|
name = "issue31-creator-claude"
|
|
t = _tree(stop_box, stop_box, conv, stop_box / "ctmp")
|
|
_stop_row(stop_box, name, conv)
|
|
assert _row_left(stop_box, name) == (True, True)
|
|
|
|
res = _stop(stop_box, name)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "herdr already dead" in res.stdout
|
|
assert res.stdout.count("purged: ") == 8, res.stdout
|
|
assert "residual: none (agent: claude)" in res.stdout
|
|
assert not any(p.exists() for p in t.values())
|
|
assert _row_left(stop_box, name) == (False, False)
|
|
|
|
|
|
def test_stop_purge_warns_about_a_transcript_under_another_key(stop_box, conv):
|
|
name = "issue31-creator-claude"
|
|
_tree(stop_box, stop_box, conv, stop_box / "ctmp")
|
|
stray = stop_box / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
|
|
stray.parent.mkdir()
|
|
stray.write_text("{}\n")
|
|
_stop_row(stop_box, name, conv)
|
|
|
|
res = _stop(stop_box, name)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert f"WARN residual: {stray}" in res.stdout
|
|
assert "residual: none" not in res.stdout
|
|
assert _row_left(stop_box, name) == (False, False)
|
|
|
|
|
|
def test_stop_purge_refuses_while_the_claude_process_lives(stop_box, conv):
|
|
name = "issue31-creator-claude"
|
|
holder = subprocess.Popen(["sleep", "120"])
|
|
try:
|
|
time.sleep(0.05)
|
|
t = _tree(stop_box, stop_box, conv, stop_box / "ctmp", pid=holder.pid)
|
|
_stop_row(stop_box, name, conv)
|
|
|
|
res = _stop(stop_box, name)
|
|
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert "herdr already dead" in res.stdout
|
|
assert "records preserved" in res.stderr
|
|
assert "purged: " not in res.stdout
|
|
assert all(p.exists() for p in t.values())
|
|
assert _row_left(stop_box, name) == (True, True)
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|
|
|
|
# Not a permanent block: once the process is gone the same stop purges.
|
|
res = _stop(stop_box, name)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert not any(p.exists() for p in t.values())
|
|
assert _row_left(stop_box, name) == (False, False)
|
|
|
|
|
|
@pytest.mark.parametrize("project_override", [False, True])
|
|
def test_stop_purge_follows_claude_config_dir(stop_box, conv, monkeypatch, project_override):
|
|
"""stop resolves the uuid before it purges: with CLAUDE_CONFIG_DIR alone
|
|
the transcript under <cfg>/projects must be found too, or the row goes
|
|
and every artifact stays ("nothing purged")."""
|
|
cfg = stop_box / "custom-config"
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(cfg))
|
|
if project_override:
|
|
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(cfg / "projects"))
|
|
else:
|
|
monkeypatch.delenv("CLAUDE_PROJECT_DIR", raising=False)
|
|
name = "issue31-creator-claude"
|
|
t = _tree(stop_box, stop_box, conv, stop_box / "ctmp", cfg=cfg)
|
|
_stop_row(stop_box, name, conv)
|
|
|
|
res = _stop(stop_box, name)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "nothing purged" not in res.stdout
|
|
assert "residual: none (agent: claude)" in res.stdout
|
|
assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
|
|
assert _row_left(stop_box, name) == (False, False)
|
|
|
|
|
|
def test_discovery_context_projects_follow_claude_config_dir(tmp_path, monkeypatch):
|
|
from lib_py.agents.base import DiscoveryContext as Ctx
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cfg"))
|
|
assert Ctx(str(tmp_path), "claude", home_dir=str(tmp_path / "h")).claude_dir == str(tmp_path / "cfg" / "projects")
|
|
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path / "p"))
|
|
assert Ctx(str(tmp_path), "claude", home_dir=str(tmp_path / "h")).claude_dir == str(tmp_path / "p")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# orc_onboard --remove --purge-conversation
|
|
# ---------------------------------------------------------------------------
|
|
def _run_onboard(args, sandbox: Path) -> subprocess.CompletedProcess:
|
|
env = _env({
|
|
"AGENT_SESSIONS_YAML": str(sandbox / ".mam" / "agent-sessions.yaml"),
|
|
"YAML_PATH": str(sandbox / ".mam" / "agent-sessions.yaml"),
|
|
"HOME_DIR": str(sandbox),
|
|
"AGENT_PYTHON_BIN": sys.executable,
|
|
"DELEGATE_JOB_PYTHON": sys.executable,
|
|
})
|
|
return subprocess.run(["bash", str(ONBOARD_SH)] + list(args), capture_output=True, text=True,
|
|
env=env, cwd=str(sandbox), timeout=120)
|
|
|
|
|
|
@pytest.mark.parametrize("stray", [False, True])
|
|
def test_orc_remove_purge_reports_claude_leftovers(stop_box, conv, stray):
|
|
t = _tree(stop_box, stop_box, conv, stop_box / "ctmp")
|
|
other_key = stop_box / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
|
|
if stray:
|
|
other_key.parent.mkdir()
|
|
other_key.write_text("{}\n")
|
|
assert _run_onboard(["--uuid", conv], stop_box).returncode == 0
|
|
|
|
res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes", "--agent", "claude"], stop_box)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert not any(p.exists() for p in t.values())
|
|
if stray:
|
|
assert f"WARN residual: {other_key}" in res.stdout
|
|
else:
|
|
assert "residual: none (agent: claude)" in res.stdout
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# codex, grok, opencode: residual only, purge unchanged
|
|
# ---------------------------------------------------------------------------
|
|
def _db(path: Path, ddl: list, rows: list):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(path))
|
|
for stmt in ddl:
|
|
conn.execute(stmt)
|
|
for stmt, args in rows:
|
|
conn.execute(stmt, args)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def test_opencode_residual_reports_event_rows(home, tmp_path, monkeypatch):
|
|
u, o = f"ses_{uuidlib.uuid4().hex}", f"ses_{uuidlib.uuid4().hex}"
|
|
db = home / ".local" / "share" / "opencode" / "opencode.db"
|
|
ddl = ["CREATE TABLE session (id TEXT PRIMARY KEY)",
|
|
"CREATE TABLE message (id TEXT, session_id TEXT)",
|
|
"CREATE TABLE part (id TEXT, session_id TEXT)",
|
|
"CREATE TABLE event (id TEXT, aggregate_id TEXT)",
|
|
"CREATE TABLE event_sequence (aggregate_id TEXT, seq INTEGER)",
|
|
"CREATE TABLE todo (id TEXT, session_id TEXT)"]
|
|
rows = []
|
|
for s in (u, o):
|
|
rows += [("INSERT INTO session VALUES (?)", (s,)),
|
|
("INSERT INTO message VALUES ('m', ?)", (s,)),
|
|
("INSERT INTO part VALUES ('p', ?)", (s,)),
|
|
("INSERT INTO event VALUES ('e1', ?)", (s,)),
|
|
("INSERT INTO event VALUES ('e2', ?)", (s,)),
|
|
("INSERT INTO event_sequence VALUES (?, 2)", (s,)),
|
|
("INSERT INTO todo VALUES ('t', ?)", (s,))]
|
|
_db(db, ddl, rows)
|
|
adapter = get_adapter("opencode")
|
|
monkeypatch.setattr(adapter, "_resolve_db", lambda ctx: str(db)) # never run a real opencode
|
|
ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="opencode", home_dir=str(home))
|
|
|
|
adapter.purge_artifacts(u, ctx)
|
|
|
|
assert adapter.residual_artifacts(u, ctx) == [f"{db} (event.aggregate_id: 2 rows)",
|
|
f"{db} (event_sequence.aggregate_id: 1 rows)",
|
|
f"{db} (todo.session_id: 1 rows)"]
|
|
|
|
|
|
def test_codex_residual_reports_name_and_thread_rows(home, tmp_path, monkeypatch, conv, other):
|
|
codex_home = tmp_path / "codex"
|
|
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
|
rollout = codex_home / "sessions" / "2026" / "10" / "15" / f"rollout-2026-10-15T00-00-00-{conv}.jsonl"
|
|
rollout.parent.mkdir(parents=True)
|
|
rollout.write_text("{}\n")
|
|
marker = codex_home / "tui-thread-reference-capabilities" / conv
|
|
marker.parent.mkdir()
|
|
marker.write_text("")
|
|
(codex_home / "tui-thread-reference-capabilities" / other).write_text("")
|
|
state = codex_home / "state_5.sqlite"
|
|
_db(state, ["CREATE TABLE threads (id TEXT PRIMARY KEY)"],
|
|
[("INSERT INTO threads VALUES (?)", (conv,)), ("INSERT INTO threads VALUES (?)", (other,))])
|
|
hist = codex_home / "thread_history_1.sqlite"
|
|
_db(hist, ["CREATE TABLE thread_items (thread_id TEXT, n INTEGER)"],
|
|
[("INSERT INTO thread_items VALUES (?, 1)", (conv,)), ("INSERT INTO thread_items VALUES (?, 2)", (conv,))])
|
|
ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="codex", home_dir=str(home))
|
|
|
|
# purge is not run: it would execute the real `codex delete` on PATH.
|
|
residual = get_adapter("codex").residual_artifacts(conv, ctx)
|
|
|
|
assert residual == sorted([str(marker), str(rollout)]) + [
|
|
f"{state} (threads.id: 1 rows)", f"{hist} (thread_items.thread_id: 2 rows)"]
|
|
|
|
|
|
def test_grok_residual_reports_memory_observations(home, tmp_path, conv, other):
|
|
from urllib.parse import quote
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
grok = home / ".grok"
|
|
sess = grok / "sessions" / quote(os.path.realpath(str(ws)), safe="") / conv
|
|
sess.mkdir(parents=True)
|
|
(sess / "chat_history.jsonl").write_text(json.dumps({"session_id": conv}) + "\n")
|
|
inbox = grok / "memory-v2" / "workspaces" / "w" / "observations" / "_inbox"
|
|
inbox.mkdir(parents=True)
|
|
mine = inbox / f"{conv}__t1.md"
|
|
mine.write_text("obs")
|
|
(inbox / f"{other}__t1.md").write_text("obs")
|
|
adapter = get_adapter("grok")
|
|
ctx = DiscoveryContext(workspace=str(ws), agent_name="grok", home_dir=str(home))
|
|
|
|
adapter.purge_artifacts(conv, ctx)
|
|
|
|
assert not sess.exists()
|
|
assert adapter.residual_artifacts(conv, ctx) == [str(mine)]
|