GodopuandClaude Opus 5.5
c0248b88be
_is_uuid took an uppercase or mixed-case uuid, and the paths built from it are the live lowercase session's own files on macOS's default case-insensitive APFS. The in-use checks compare exactly, though: session files and command lines carry the lowercase id, so `purge claude <UPPER>` found no holder and could remove a live session's files. _canonical now lowercases a canonical uuid, and the paths, the session-file and job matches, the in-use check and the residual all use it; a command line is searched for both spellings. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
453 lines
18 KiB
Python
453 lines
18 KiB
Python
import os, re, json, glob, math, stat, time, shutil, calendar, 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
|
||
|
||
|
||
_UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.IGNORECASE)
|
||
|
||
|
||
def _canonical(uuid) -> Optional[str]:
|
||
"""The id claude itself writes, in lowercase, for any spelling of a
|
||
canonical uuid; None for anything else. Any other name in a session's
|
||
place is not one: projects/<key>/memory/ (auto memory) sits beside the
|
||
session dirs. Another spelling must go through the same checks as
|
||
claude's own: on macOS's default case-insensitive APFS it names the
|
||
same files, while the ids in session files and on command lines are
|
||
compared exactly."""
|
||
return uuid.lower() if isinstance(uuid, str) and _UUID.fullmatch(uuid) else None
|
||
|
||
|
||
def _config_dir(ctx: DiscoveryContext) -> str:
|
||
"""claude's config root: CLAUDE_CONFIG_DIR moves all of it (sessions/,
|
||
file-history/, session-env/, jobs/, projects/)."""
|
||
return os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(ctx.home_dir, ".claude")
|
||
|
||
|
||
def _projects_dirs(ctx: DiscoveryContext) -> list:
|
||
"""ctx.claude_dir (CLAUDE_PROJECT_DIR, else the config root's projects/)
|
||
and the config root's own projects/, each once."""
|
||
dirs, seen = [], set()
|
||
for d in (ctx.claude_dir, os.path.join(_config_dir(ctx), "projects")):
|
||
real = os.path.realpath(d)
|
||
if real not in seen:
|
||
seen.add(real)
|
||
dirs.append(d)
|
||
return dirs
|
||
|
||
|
||
# Node's os.tmpdir() when no TMPDIR/TMP/TEMP is set; tests point it elsewhere.
|
||
_SYSTEM_TMP = "/tmp"
|
||
|
||
|
||
def _tmp_roots() -> list:
|
||
"""The per-uid temp roots claude may have used: ${CLAUDE_CODE_TMPDIR:-
|
||
os.tmpdir()}/claude-<uid>. A sandboxed tool shell sets CLAUDE_CODE_TMPDIR
|
||
to that per-uid root itself. The session that wrote the files may have
|
||
run under another env, so every candidate that exists is returned."""
|
||
name = f"claude-{os.getuid()}"
|
||
cands = []
|
||
override = os.environ.get("CLAUDE_CODE_TMPDIR", "")
|
||
if override:
|
||
cands.append(os.path.join(override, name))
|
||
if os.path.basename(override.rstrip("/")) == name:
|
||
cands.append(override)
|
||
# Node's os.tmpdir() on POSIX, not tempfile.gettempdir() (which probes
|
||
# writability and falls back to the cwd).
|
||
tmp = os.environ.get("TMPDIR") or os.environ.get("TMP") or os.environ.get("TEMP") or _SYSTEM_TMP
|
||
cands.append(os.path.join(tmp.rstrip("/") or "/", name))
|
||
cands.append(os.path.join(_SYSTEM_TMP, name))
|
||
roots, seen = [], set()
|
||
for c in cands:
|
||
real = os.path.realpath(c)
|
||
if os.path.isdir(c) and real not in seen:
|
||
seen.add(real)
|
||
roots.append(c)
|
||
return roots
|
||
|
||
|
||
def _session_file_live(data: dict, pid) -> bool:
|
||
"""The process that wrote this sessions/<pid>.json is still running.
|
||
|
||
A dead pid is stale. A live pid whose start (`ps -o lstart=`) is later
|
||
than the file's startedAt was reused by another process, so it is stale
|
||
too. Anything that cannot be told is taken as live: a purge is then
|
||
refused, never run over a live conversation."""
|
||
if not isinstance(pid, int) or isinstance(pid, bool) or pid <= 0:
|
||
return True
|
||
try:
|
||
os.kill(pid, 0)
|
||
except ProcessLookupError:
|
||
return False
|
||
except OSError:
|
||
pass
|
||
# startedAt is epoch ms: a finite positive number. A bool, a string, NaN
|
||
# (every comparison with it is false), an infinity or an int too large
|
||
# for a float is no time at all.
|
||
started = data.get("startedAt")
|
||
if isinstance(started, bool) or not isinstance(started, (int, float)):
|
||
return True
|
||
try:
|
||
started = float(started) / 1000.0
|
||
except OverflowError:
|
||
return True
|
||
if not math.isfinite(started) or started <= 0:
|
||
return True
|
||
# lstart in UTC: a local time is ambiguous in the hour a DST change repeats.
|
||
try:
|
||
out = subprocess.run(["ps", "-p", str(pid), "-o", "lstart="], capture_output=True, text=True,
|
||
timeout=5, env=dict(os.environ, LC_ALL="C", TZ="UTC0")).stdout
|
||
lstart = calendar.timegm(time.strptime(" ".join(out.split()), "%a %b %d %H:%M:%S %Y"))
|
||
except Exception:
|
||
return True
|
||
# startedAt is written 1.7-3.2 s after the process starts; 2 s covers
|
||
# lstart's one-second resolution.
|
||
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):
|
||
os.remove(path)
|
||
else:
|
||
shutil.rmtree(path, ignore_errors=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _key_files(session_file: str) -> list:
|
||
"""The key files of the pid a sessions/<pid>.json is named after,
|
||
matched by name only; their content is never read."""
|
||
d, name = os.path.split(session_file)
|
||
stem = name[:-len(".json")]
|
||
if not stem.isdigit():
|
||
return []
|
||
try:
|
||
names = sorted(os.listdir(d))
|
||
except OSError:
|
||
return []
|
||
return [os.path.join(d, n) for n in names if re.fullmatch(rf"{stem}\.[0-9a-f]{{64}}\.key", n)]
|
||
|
||
|
||
def _remove_job(job: str) -> None:
|
||
"""state.json goes last: it is the only proof that the dir is this
|
||
session's, so a dir that could not be emptied keeps it."""
|
||
if os.path.islink(job):
|
||
_remove(job)
|
||
return
|
||
try:
|
||
names = os.listdir(job)
|
||
except OSError:
|
||
return
|
||
for n in names:
|
||
if n != "state.json":
|
||
_remove(os.path.join(job, n))
|
||
try:
|
||
left = os.listdir(job)
|
||
except OSError:
|
||
return
|
||
if left in ([], ["state.json"]):
|
||
_remove(job)
|
||
|
||
|
||
def _label(path: str) -> str:
|
||
"""How a path is printed: a session key file is named by pid only, its
|
||
hex part is not shown."""
|
||
d, name = os.path.split(path)
|
||
m = re.fullmatch(r"(\d+)\.[0-9a-f]{64}\.key", name)
|
||
return os.path.join(d, f"{m.group(1)}.*.key") if m else path
|
||
|
||
|
||
class ClaudeAgentAdapter(BaseAgentAdapter):
|
||
@property
|
||
def name(self) -> str:
|
||
return 'claude'
|
||
|
||
@property
|
||
def own_key(self) -> str:
|
||
return 'claude_session_id_own'
|
||
|
||
@property
|
||
def ready_tokens(self) -> str:
|
||
return 'Anthropic|Assistant|Chat|Welcome|Claude Code|Opus|Sonnet|Haiku'
|
||
|
||
@property
|
||
def exit_key(self) -> str:
|
||
return '/exit'
|
||
|
||
@property
|
||
def delegate_agent_key(self) -> str:
|
||
return 'claude-code'
|
||
|
||
@property
|
||
def identity_cache_fields(self) -> tuple:
|
||
return ('session_id', 'session_jsonl', 'session_size_bytes', 'session_lines')
|
||
|
||
@property
|
||
def input_prompt(self) -> str:
|
||
return '❯'
|
||
|
||
@property
|
||
def input_placeholder(self) -> str:
|
||
return ''
|
||
|
||
@property
|
||
def input_rule_pattern(self) -> str:
|
||
return '─{10,}'
|
||
|
||
@property
|
||
def modal_tokens(self) -> Optional[str]:
|
||
return 'Try the new fullscreen renderer\\?'
|
||
|
||
@property
|
||
def startup_dialogs(self) -> list:
|
||
# The ladder handle_startup_dialogs used to hardcode, unchanged.
|
||
return [
|
||
('Do you trust the files|Yes, I trust this folder|Quick safety check', ['Enter'], 'trust'),
|
||
# Fullscreen-renderer upsell, not the idle /tui tip: Enter would
|
||
# accept and restart the session without the permission flags.
|
||
('Yes, try it', ['Escape'], 'notice'),
|
||
('Yes, proceed', ['Down', 'Enter'], 'trust'),
|
||
('Resuming the full session', ['Enter'], 'notice'),
|
||
('Press Enter to continue', ['Enter'], 'notice'),
|
||
]
|
||
|
||
def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
|
||
return f"{ctx.claude_dir}/{ctx.ws_key}/{uuid}.jsonl"
|
||
|
||
def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
|
||
path = self.artifact_path(uuid, ctx)
|
||
if not os.path.exists(path):
|
||
return False
|
||
if ctx.epoch and os.path.getmtime(path) < ctx.epoch:
|
||
return False
|
||
try:
|
||
valid_session = False
|
||
found_cwd = None
|
||
with open(path) as f:
|
||
for _ in range(50):
|
||
line = f.readline()
|
||
if not line:
|
||
break
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
payload = json.loads(line)
|
||
if payload.get("sessionId") == uuid:
|
||
valid_session = True
|
||
if payload.get("cwd"):
|
||
found_cwd = payload.get("cwd")
|
||
break
|
||
except Exception:
|
||
pass
|
||
if not valid_session:
|
||
return False
|
||
if found_cwd and workspace_key(found_cwd) != workspace_key(ctx.cwd):
|
||
return False
|
||
except Exception:
|
||
return False
|
||
return True
|
||
|
||
def _session_files(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
"""(path, pid, data) of each sessions/<pid>.json naming this session -
|
||
the registry of running claude processes, which a live `claude -c`
|
||
is found by even though its command line carries no uuid."""
|
||
found = []
|
||
for path in sorted(glob.glob(os.path.join(_config_dir(ctx), "sessions", "*.json"))):
|
||
try:
|
||
with open(path) as f:
|
||
data = json.load(f)
|
||
except Exception:
|
||
continue
|
||
if not isinstance(data, dict) or data.get("sessionId") != uuid:
|
||
continue
|
||
pid = data.get("pid")
|
||
if not isinstance(pid, int):
|
||
stem = os.path.basename(path)[:-5]
|
||
pid = int(stem) if stem.isdigit() else None
|
||
found.append((path, pid, data))
|
||
return found
|
||
|
||
def _owned_job(self, uuid: str, ctx: DiscoveryContext) -> Optional[str]:
|
||
"""jobs/<first 8 chars>/ when its state.json names this session; the
|
||
short name alone can be another session's."""
|
||
job = os.path.join(_config_dir(ctx), "jobs", uuid[:8])
|
||
try:
|
||
with open(os.path.join(job, "state.json")) as f:
|
||
if json.load(f).get("sessionId") == uuid:
|
||
return job
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def _candidates(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
"""(root, path) of every existing path that belongs to this session
|
||
alone (Issue #31), with the root (config, projects or tmp) it was
|
||
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
|
||
# pre-#31 reach: the `<id>.jsonl` file alone.
|
||
if not _plausible_id(uuid):
|
||
return []
|
||
canon = _canonical(uuid)
|
||
cfg = _config_dir(ctx)
|
||
found = []
|
||
for pdir in _projects_dirs(ctx):
|
||
base = os.path.join(pdir, ctx.ws_key, canon or uuid)
|
||
found += [(pdir, p) for p in ((base + ".jsonl", base) if canon else (base + ".jsonl",))
|
||
if os.path.lexists(p)]
|
||
if not canon:
|
||
return found
|
||
uuid = canon
|
||
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)]
|
||
job = self._owned_job(uuid, ctx)
|
||
if 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.
|
||
found += [(cfg, p) for p in _key_files(path) + [path]]
|
||
for root in _tmp_roots():
|
||
try:
|
||
keys = sorted(os.listdir(root))
|
||
except OSError:
|
||
continue
|
||
found += [(root, p) for p in (os.path.join(root, k, uuid) for k in keys) if os.path.lexists(p)]
|
||
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).
|
||
return bool(self._owned_paths(uuid, ctx))
|
||
|
||
def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
|
||
# No claude process holds a name that is not a uuid; searching
|
||
# command lines for one ("a/b", "memory") only finds strangers.
|
||
canon = _canonical(uuid)
|
||
if not canon:
|
||
return False
|
||
# claude's own spelling and the one asked about, on a command line.
|
||
if any(conversation_in_use(u, [self.artifact_path(canon, ctx)]) for u in {canon, uuid}):
|
||
return True
|
||
return any(_session_file_live(data, pid) for _, pid, data in self._session_files(canon, ctx))
|
||
|
||
def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
return [f"a claude process whose session file under {os.path.join(_config_dir(ctx), 'sessions')}/ names it"] \
|
||
+ super().in_use_hints(_canonical(uuid) or uuid, ctx)
|
||
|
||
def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
# Best effort: never raises (stop runs this inside its registry
|
||
# transaction). What could not be removed is not listed here and
|
||
# shows up in residual_artifacts instead.
|
||
# Ownership evidence (a job's state.json, a session file) is removed
|
||
# only after what it vouches for, so a retry can still find the rest.
|
||
uuid = _canonical(uuid) or uuid
|
||
purged = []
|
||
job = self._owned_job(uuid, ctx)
|
||
for path in self._owned_paths(uuid, ctx):
|
||
if path == job:
|
||
_remove_job(path)
|
||
elif path.endswith(".json") and any(os.path.lexists(k) for k in _key_files(path)):
|
||
continue
|
||
else:
|
||
_remove(path)
|
||
if not os.path.lexists(path) and _label(path) not in purged:
|
||
purged.append(_label(path))
|
||
return purged
|
||
|
||
def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
"""What still carries this session: names holding the uuid under the
|
||
config root, in projects (two levels: key, then entry) and in the tmp
|
||
roots (two levels), plus the content-matched jobs dir and session
|
||
files with their (masked) key files. projects is not walked deeper;
|
||
it can be very large."""
|
||
if not _canonical(uuid):
|
||
return sorted(p for _, p in self._candidates(uuid, ctx))
|
||
uuid = _canonical(uuid)
|
||
cfg = _config_dir(ctx)
|
||
pdirs = _projects_dirs(ctx)
|
||
found = set(scan_names(cfg, uuid, skip=pdirs))
|
||
for pdir in pdirs:
|
||
try:
|
||
keys = sorted(os.listdir(pdir))
|
||
except OSError:
|
||
keys = []
|
||
for key in keys:
|
||
kdir = os.path.join(pdir, key)
|
||
if uuid in key:
|
||
found.add(kdir)
|
||
elif os.path.isdir(kdir):
|
||
try:
|
||
found.update(os.path.join(kdir, n) for n in os.listdir(kdir) if uuid in n)
|
||
except OSError:
|
||
pass
|
||
# Every candidate, inside its root or behind a link (never purged).
|
||
found.update(_label(p) for _, p in self._candidates(uuid, ctx))
|
||
for path, _, _ in self._session_files(uuid, ctx):
|
||
found.add(path)
|
||
found.update(_label(k) for k in _key_files(path))
|
||
return sorted(found)
|
||
|
||
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
|
||
if use_wrapper or not session_uuid:
|
||
return f"{binary} --dangerously-skip-permissions"
|
||
return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"
|
||
|
||
def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
|
||
if materialized and session_uuid:
|
||
return f"{binary} --dangerously-skip-permissions -r {session_uuid}"
|
||
elif session_uuid:
|
||
return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"
|
||
return f"{binary} --dangerously-skip-permissions"
|
||
|
||
def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
|
||
if run_cmd is not None:
|
||
try:
|
||
rc, stdout, stderr = run_cmd(['claude', 'auth', 'status'])
|
||
return rc == 0 and ('"loggedIn":true' in (stdout or '').replace(' ', ''))
|
||
except Exception:
|
||
return False
|
||
try:
|
||
res = subprocess.run(['claude', 'auth', 'status'], capture_output=True, text=True)
|
||
return res.returncode == 0 and ('"loggedIn":true' in (res.stdout or '').replace(' ', ''))
|
||
except Exception:
|
||
return False
|
||
|
||
def discover(self, ctx: DiscoveryContext) -> list:
|
||
key = workspace_key(ctx.workspace)
|
||
proj = f"{ctx.claude_dir}/{key}"
|
||
candidates = []
|
||
if os.path.isdir(proj):
|
||
for j in sorted(glob.glob(f"{proj}/*.jsonl"), key=os.path.getmtime, reverse=True):
|
||
cand = os.path.basename(j)[:-6]
|
||
if cand and self.verify_artifact(cand, ctx):
|
||
candidates.append(cand)
|
||
return candidates
|