GodopuandClaude Opus 5.5
de6f3fc81f
macOS /bin/sh (bash 3.2) execs the lone `sleep 120` of `sh -c "sleep 120" <uuid>`, so the uuid left the process list and both [upper] and [mixed] failed on the Mac. A Python holder keeps the uuid in argv, like _holder() in test_issue10_13_purge.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
973 lines
41 KiB
Python
973 lines
41 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)
|
|
|
|
|
|
@pytest.mark.parametrize("spell", [str.upper, lambda u: u[:18].upper() + u[18:]], ids=["upper", "mixed"])
|
|
def test_any_spelling_of_the_uuid_is_the_lowercase_session(home, tmp_path, conv, live, spell):
|
|
"""claude writes its ids in lowercase, and on macOS's default
|
|
case-insensitive APFS another spelling names the very same files. Every
|
|
check then uses the lowercase id: the live holder is seen, by its session
|
|
file or its command line, and a purge takes the session's own paths."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
t = _tree(home, ws, conv, tmp_path / "ctmp", pid=live.pid)
|
|
other = spell(conv)
|
|
assert other != conv
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.conversation_in_use(other, ctx) is True
|
|
for args in (["purge", "claude", other, str(ws)], ["in-use", "claude", other, str(ws)]):
|
|
res = _agents_cli(args, ws)
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert all(p.exists() for p in t.values())
|
|
|
|
live.kill()
|
|
live.wait()
|
|
# Not `sh -c "sleep 120" <uuid>`: a shell execs its last simple command
|
|
# and the uuid would vanish from the process list (macOS /bin/sh).
|
|
named = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)", conv])
|
|
try:
|
|
time.sleep(0.05)
|
|
assert adapter.conversation_in_use(other, ctx) is True, "a command line naming the lowercase id"
|
|
finally:
|
|
named.kill()
|
|
named.wait()
|
|
|
|
assert adapter.has_artifacts(other, ctx) is True
|
|
purged = adapter.purge_artifacts(other, ctx)
|
|
assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
|
|
assert purged and all(other not in p for p in purged)
|
|
assert adapter.residual_artifacts(other, ctx) == []
|
|
|
|
|
|
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_purge_cli_reports_a_linked_orphan_it_will_not_remove(home, tmp_path, conv):
|
|
"""A candidate behind a link is all that is left: the CLI still picks
|
|
claude, so its residual names it, and removes nothing."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
keep = tmp_path / "outside" / conv / "keep.txt"
|
|
keep.parent.mkdir(parents=True)
|
|
keep.write_text("keep")
|
|
troot = tmp_path / "ctmp" / f"claude-{UID}"
|
|
troot.mkdir(parents=True)
|
|
(troot / "linked-workspace").symlink_to(tmp_path / "outside")
|
|
|
|
res = _agents_cli(["purge", "claude", conv, str(ws)], ws)
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert json.loads(res.stdout) == {"agent": "claude", "purged": [],
|
|
"residual": [str(troot / "linked-workspace" / conv)], "in_use": False}
|
|
assert keep.exists()
|
|
|
|
|
|
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) == []
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["uuid-c", "memory", "notes"])
|
|
def test_a_non_uuid_id_reaches_nothing(home, tmp_path, name):
|
|
"""Only a uuid names a claude session: no path is built for any other id,
|
|
not even the `<id>.jsonl` the adapter took before #31."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
proj = home / ".claude" / "projects" / workspace_key(str(ws))
|
|
proj.mkdir()
|
|
jsonl = proj / f"{name}.jsonl"
|
|
jsonl.write_text(json.dumps({"sessionId": name, "cwd": os.path.realpath(str(ws))}) + "\n")
|
|
kept = [jsonl, proj / name / "a", home / ".claude" / "file-history" / name / "b"]
|
|
for p in kept[1:]:
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text("x")
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.has_artifacts(name, ctx) is False
|
|
assert adapter.residual_artifacts(name, ctx) == []
|
|
assert adapter.purge_artifacts(name, ctx) == []
|
|
res = _agents_cli(["purge", "claude", name, str(ws)], ws)
|
|
assert json.loads(res.stdout)["purged"] == [], res.stdout + res.stderr
|
|
assert all(p.exists() for p in kept)
|
|
|
|
|
|
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_liveness_reads_lstart_as_utc(tmp_path, monkeypatch, live):
|
|
"""2026-11-01 01:30 happens twice in EST5EDT. A process that started at
|
|
the second one (06:30 UTC) shows lstart 01:30 to a local-time ps; read as
|
|
the first (05:30 UTC) it predates a session file written at 05:45. ps
|
|
runs with TZ=UTC0 instead, which BSD ps and procps both honour."""
|
|
import calendar
|
|
from lib_py.agents.adapters import claude as claude_mod
|
|
fake = tmp_path / "fakebin"
|
|
fake.mkdir()
|
|
(fake / "ps").write_text('#!/bin/sh\nif [ "$TZ" = UTC0 ]; then echo "Sun Nov 1 06:30:00 2026"; '
|
|
'else echo "Sun Nov 1 01:30:00 2026"; fi\n')
|
|
(fake / "ps").chmod(0o755)
|
|
utc = lambda hhmmss: calendar.timegm(time.strptime(f"2026-11-01 {hhmmss}", "%Y-%m-%d %H:%M:%S")) * 1000 # noqa: E731
|
|
try:
|
|
with monkeypatch.context() as m:
|
|
m.setenv("PATH", f"{fake}:{os.environ['PATH']}")
|
|
m.setenv("TZ", "EST5EDT,M3.2.0,M11.1.0")
|
|
time.tzset()
|
|
assert claude_mod._session_file_live({"startedAt": utc("05:45:00")}, live.pid) is False, "reused pid"
|
|
assert claude_mod._session_file_live({"startedAt": utc("06:30:02")}, live.pid) is True
|
|
finally:
|
|
time.tzset()
|
|
|
|
|
|
@pytest.mark.parametrize("started", ["NaN", float("nan"), float("inf"), float("-inf"), "inf", "1760000000000",
|
|
"soon", True, False, None, [1], pytest.param(10 ** 400, id="10**400"),
|
|
"<missing>"])
|
|
def test_an_unusable_started_at_means_live(live, started):
|
|
from lib_py.agents.adapters import claude as claude_mod
|
|
data = {} if started == "<missing>" else {"startedAt": started}
|
|
assert claude_mod._session_file_live(data, live.pid) is True
|
|
assert claude_mod._session_file_live(data, _dead_pid()) is False, "a dead pid holds nothing"
|
|
|
|
|
|
@pytest.mark.parametrize("raw", ["NaN", '"NaN"', "Infinity", "true",
|
|
pytest.param("1" + "0" * 400, id="10**400")])
|
|
def test_a_live_holder_with_an_unusable_started_at_blocks_the_purge(home, tmp_path, conv, live, raw):
|
|
"""json.load turns a bare NaN or Infinity into a float that every
|
|
comparison fails, which once made a live holder look stale, and a huge
|
|
integer into an int no float can hold."""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
sfile, key = _session_file(home / ".claude", live.pid, conv)
|
|
sfile.write_text(f'{{"pid": {live.pid}, "sessionId": "{conv}", "cwd": "/x", "startedAt": {raw}}}')
|
|
adapter = get_adapter("claude")
|
|
ctx = _ctx(ws, home)
|
|
|
|
assert adapter.conversation_in_use(conv, ctx) is True
|
|
res = _agents_cli(["in-use", "claude", conv, str(ws)], ws)
|
|
assert res.returncode == 4, res.stdout + res.stderr
|
|
assert adapter.purge_artifacts(conv, ctx) == []
|
|
assert sfile.exists() and key.exists()
|
|
|
|
|
|
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
|
|
assert adapter.in_use_hints(conv, ctx)[0] == f"a claude process whose session file under {cfg / 'sessions'}/ names it"
|
|
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 f"session file under {home / '.claude' / 'sessions'}/ names it" in res.stderr
|
|
assert f"ps -axww | grep {conv}" in res.stderr
|
|
|
|
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 f"session file under {stop_box / '.claude' / 'sessions'}/ names it" in res.stderr
|
|
assert f"ps -axww | grep {conv}" 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)]
|