GodopuandClaude Opus 5.5
366da3b3d6
claude had no residual_artifacts, so a purge always printed `residual: none`. It now reports names holding the uuid under ~/.claude (projects excluded), in projects two levels deep, in the tmp roots, plus the owned jobs dir and session files. Shared files (history.jsonl, ~/.claude.json) are not reported: they always mention every session. scan_names goes to base.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
328 lines
13 KiB
Python
328 lines
13 KiB
Python
import os, re, json, glob, time, shutil, subprocess
|
||
from typing import Optional, Any
|
||
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, conversation_in_use, scan_names, _plausible_id
|
||
from lib_py.verify_session import workspace_key
|
||
|
||
|
||
def _config_dir(ctx: DiscoveryContext) -> str:
|
||
return os.path.join(ctx.home_dir, ".claude")
|
||
|
||
|
||
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 "/tmp"
|
||
cands.append(os.path.join(tmp.rstrip("/") or "/", name))
|
||
cands.append(os.path.join("/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
|
||
try:
|
||
out = subprocess.run(["ps", "-p", str(pid), "-o", "lstart="], capture_output=True, text=True,
|
||
timeout=5, env=dict(os.environ, LC_ALL="C")).stdout
|
||
lstart = time.mktime(time.strptime(" ".join(out.split()), "%a %b %d %H:%M:%S %Y"))
|
||
started = float(data["startedAt"]) / 1000.0
|
||
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 _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 _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 _owned_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
|
||
"""Every existing path that belongs to this session alone (Issue #31).
|
||
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.
|
||
if not _plausible_id(uuid):
|
||
return []
|
||
cfg = _config_dir(ctx)
|
||
jsonl = self.artifact_path(uuid, ctx)
|
||
paths = [p for p in (jsonl, jsonl[:-len(".jsonl")],
|
||
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:
|
||
paths.append(job)
|
||
for path, pid, data in self._session_files(uuid, ctx):
|
||
if _session_file_live(data, pid):
|
||
continue
|
||
paths.append(path)
|
||
stem = os.path.basename(path)[:-len(".json")]
|
||
sessions = os.path.dirname(path)
|
||
if stem.isdigit():
|
||
# The key file is matched by name only; its content is never read.
|
||
paths.extend(os.path.join(sessions, n) for n in sorted(os.listdir(sessions))
|
||
if re.fullmatch(rf"{stem}\.[0-9a-f]{{64}}\.key", n))
|
||
for root in _tmp_roots():
|
||
try:
|
||
keys = sorted(os.listdir(root))
|
||
except OSError:
|
||
continue
|
||
paths.extend(p for p in (os.path.join(root, k, uuid) for k in keys) if os.path.lexists(p))
|
||
return paths
|
||
|
||
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:
|
||
if conversation_in_use(uuid, [self.artifact_path(uuid, ctx)]):
|
||
return True
|
||
return any(_session_file_live(data, pid) for _, pid, data in self._session_files(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.
|
||
purged = []
|
||
for path in self._owned_paths(uuid, ctx):
|
||
_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. projects is not walked deeper; it can be very large."""
|
||
if not _plausible_id(uuid):
|
||
return []
|
||
cfg = _config_dir(ctx)
|
||
found = set(scan_names(cfg, uuid, skip=(os.path.join(cfg, "projects"), ctx.claude_dir)))
|
||
try:
|
||
keys = sorted(os.listdir(ctx.claude_dir))
|
||
except OSError:
|
||
keys = []
|
||
for key in keys:
|
||
kdir = os.path.join(ctx.claude_dir, 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
|
||
job = self._owned_job(uuid, ctx)
|
||
if job:
|
||
found.add(job)
|
||
found.update(path for path, _, _ in self._session_files(uuid, ctx))
|
||
for root in _tmp_roots():
|
||
try:
|
||
tkeys = os.listdir(root)
|
||
except OSError:
|
||
continue
|
||
found.update(p for p in (os.path.join(root, k, uuid) for k in tkeys) if os.path.lexists(p))
|
||
return sorted(found)
|
||
|
||
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
|
||
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
|