diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 3812dc5..46c79a3 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -453,6 +453,35 @@ from the desired release checkout to refresh both standalone program and skill. Local `session-peer update --check` and `session-peer update` also populate the same cache used by automatic notices. +### Claude profiles managed by Orca + +On macOS, when neither `CLAUDE_CONFIG_DIR` nor `ANTHROPIC_CONFIG_DIR` selects a +directory, discovery includes `~/.claude/sessions` and known same-user Orca +`~/Library/Application Support/orca/claude-profiles/*/home/sessions` directories. +An explicit environment directory retains its original scope and precedence; +it never expands into other profiles. For example, use +`CLAUDE_CONFIG_DIR="/path/to/profile/home" session-peer list --agent claude --json` +to inspect one profile only. Installation paths are unchanged. + +On all platforms, default and explicit registry scans share a total limit of +4,096 directory entries and bounded 1 MiB registration files. Only the additional +Orca profile inventory is macOS-specific, limited to 128 immediate entries. +Automatic profile directories reject symlinks, reparse points and foreign-user +ownership. Discovery and doctor never read credentials, inbox keys or transcripts. +Canonical aliases of one registry are scanned once. Distinct records claiming +the same live PID or inbox conflict even when their public metadata is identical. + +Partial errors preserve healthy listing rows, but **any inventory error, including +a live conflict, refuses every target selection**, even an unrelated healthy PID, +for dry-run, send and handoff. An explicit directory can isolate errors in other +profiles; it cannot repair conflicts or limits within the selected directory. +Legacy single-directory handling of malformed records is unchanged. +Selected record and endpoint evidence is rechecked without rescanning other +profiles. Ordinary Unix socket aliases remain supported; their existing optional +generation limitation remains. A generation token is not send permission or ACK. +These checks do not authenticate a model, remove all filesystem races or prove +receipt. Other Orca layouts and arbitrary configuration roots are not scanned. + ### Environment variables | Variable | Effect | diff --git a/docs/i18n-manifest.json b/docs/i18n-manifest.json index 6d78726..f53a8b1 100644 --- a/docs/i18n-manifest.json +++ b/docs/i18n-manifest.json @@ -15,7 +15,7 @@ "docs/ai-assistant-guide.md": "9624ec96b92fb688fae3be3c94cb3564c77985091be451fbf1f056a1af9b9d9c", "docs/antigravity.md": "f2b5122a0ce6e6fb5bd4d8ed50ce2c8bd2dc9c98ec8ce328405cf7c2cd24229d", "docs/architecture/agent-transports.md": "01a4e3942e87acb476595859f2e37f9653ea1e3bc19ebc68dced1adf20a6966f", - "docs/cli-reference.md": "a1aae5b881f4aa939c995d7bc06819d04f808da3b31fd0d5da66c817e1b47e82", + "docs/cli-reference.md": "43e9e3232d8975d3e6b17065f40f43ff590184cac503d2c8fdae05142668334b", "docs/compatibility-v1.md": "7e551aad37a369faff7e370728376763efb54c2c9e9cfd8556a0c19aabc3d103", "docs/contracts/handoff-v1.md": "46a953010c49214bd8f969412f26713384fbe09a0ba78ebf0fef75da40d0d889", "docs/diagnostics.md": "0949e53612eeab2cd7c7bbda498e14d71e9cb1b1189aa7722030cb8702cd8481", diff --git a/docs/ja/cli-reference.md b/docs/ja/cli-reference.md index c9628f4..590eff6 100644 --- a/docs/ja/cli-reference.md +++ b/docs/ja/cli-reference.md @@ -442,6 +442,32 @@ pip/pipx/uv のインストールは検査しません。リモート更新は ローカルの `session-peer update --check` および `session-peer update` は、自動通知で 使用されるのと同じキャッシュも設定します。 +### Orca が管理する Claude プロファイル + +macOS で `CLAUDE_CONFIG_DIR` と `ANTHROPIC_CONFIG_DIR` を指定しない場合、 +`~/.claude/sessions` と同じユーザーが所有する既知の Orca ディレクトリ +`~/Library/Application Support/orca/claude-profiles/*/home/sessions` を検索します。 +環境変数による明示指定は従来の優先順位と範囲を維持し、他のプロファイルへ広げません。 +1 つだけ確認するには +`CLAUDE_CONFIG_DIR="/path/to/profile/home" session-peer list --agent claude --json` +を使います。インストール先は変わりません。 + +すべてのプラットフォームで、既定・明示指定の登録先検索は全体で 4,096 項目、 +登録ファイルごとに 1 MiB に制限されます。追加の Orca プロファイル検索だけが macOS 専用で、 +直下 128 項目が上限です。自動検索するプロファイルのシンボリックリンク、再解析ポイント、 +他のユーザーの所有権は拒否します。検索と doctor は認証情報、受信箱の鍵、会話履歴を読みません。 +同じ実体の登録ディレクトリの別名は 1 回だけ検索します。別の記録が同じ生存 PID または +受信箱を主張すれば、公開メタデータが同一でも矛盾です。 + +正常な一覧は残しますが、**生存セッションの矛盾を含む検索エラーが 1 件でもあれば、 +無関係な正常 PID も含めてすべての送信先選択を拒否**します。dry-run・送信・handoff が対象です。 +明示指定で他のプロファイルのエラーを分離できますが、指定先内部の矛盾や検索上限は解消しません。 +従来の単一ディレクトリでの不正な記録の扱いは変更しません。 +選択した記録・受信箱を、他のプロファイルへ再検索せず確認します。通常の Unix ソケット別名は +維持しますが、既存の任意世代検査の制限も維持します。世代トークンは送信権限や ACK ではありません。 +モデル認証、すべてのファイルシステム競合の排除、受信証明は提供しません。 +他の Orca 構成や任意の設定ディレクトリは自動検索しません。 + ### 環境変数 | 変数 | 効果 | diff --git a/docs/ko/cli-reference.md b/docs/ko/cli-reference.md index 2ab8340..bc9e92e 100644 --- a/docs/ko/cli-reference.md +++ b/docs/ko/cli-reference.md @@ -440,6 +440,33 @@ pip/pipx/uv 설치가 아니라 `~/.local/share/session-peer/session_peer.py`만 로컬 `session-peer update --check` 및 `session-peer update`는 자동 알림에 사용되는 동일한 캐시도 채웁니다. +### Orca에서 관리하는 Claude 프로필 + +macOS에서 `CLAUDE_CONFIG_DIR`과 `ANTHROPIC_CONFIG_DIR`을 지정하지 않으면 +`~/.claude/sessions`와 같은 사용자 소유의 알려진 Orca 경로 +`~/Library/Application Support/orca/claude-profiles/*/home/sessions`를 검색합니다. +환경변수로 지정하면 기존 우선순위와 범위를 유지하며 다른 프로필로 검색을 넓히지 않습니다. +한 프로필만 확인하려면 +`CLAUDE_CONFIG_DIR="/path/to/profile/home" session-peer list --agent claude --json`을 +사용하세요. 설치 경로는 바뀌지 않습니다. + +모든 플랫폼의 기본·명시적 등록 디렉터리 검색에는 전체 항목 4,096개, 등록 파일당 +1 MiB 제한이 적용됩니다. 추가 Orca 프로필 검색만 macOS 전용이며 바로 아래 항목 128개로 +제한합니다. 자동 검색하는 프로필 디렉터리의 심볼릭 링크·재분석 지점·다른 사용자 소유권은 +거부합니다. 검색과 doctor는 인증정보·수신함 키·대화 기록을 읽지 않습니다. +같은 실제 등록 디렉터리의 별칭은 한 번만 검색합니다. 별도 기록이 같은 살아 있는 PID나 +수신함을 주장하면 공개 메타데이터가 같아도 충돌입니다. + +일부 오류가 있어도 정상 목록은 보존하지만, **살아 있는 세션 충돌을 포함한 검색 오류가 +하나라도 있으면 관련 없는 정상 PID도 포함하여 모든 대상 선택을 거부**합니다. +이는 dry-run·전송·handoff에 모두 적용됩니다. 명시적 디렉터리는 다른 프로필의 오류를 +분리할 수 있지만 선택한 디렉터리 내부 충돌이나 검색 한도를 해결하지는 못합니다. +기존 단일 디렉터리의 잘못된 기록 처리 방식은 유지합니다. +선택한 기록·수신함을 다른 프로필로 재검색하지 않고 확인합니다. 일반 Unix 소켓 별칭은 +유지하지만 기존의 선택적 세대 확인 제한도 유지합니다. 세대 토큰은 전송 권한이나 ACK가 아닙니다. +모델 인증·모든 파일시스템 경쟁 제거·수신 증명을 제공하지 않으며, 다른 Orca 구조나 임의 설정 +경로를 자동 검색하지 않습니다. + ### 환경 변수 | Variable | Effect | diff --git a/docs/zh-CN/cli-reference.md b/docs/zh-CN/cli-reference.md index e4b12e2..e042fcf 100644 --- a/docs/zh-CN/cli-reference.md +++ b/docs/zh-CN/cli-reference.md @@ -314,6 +314,29 @@ session-peer update --host web-01 # push this program if version 无论是本地还是远程的 `update` 都不会刷新 Claude 技能。从所需发布版本的检出目录重新运行 `install.sh` 可同时刷新独立程序和技能。本地的 `session-peer update --check` 和 `session-peer update` 也会填充自动通知所使用的相同缓存。 +### Orca 管理的 Claude 配置档案 + +在 macOS 上，未指定 `CLAUDE_CONFIG_DIR` 和 `ANTHROPIC_CONFIG_DIR` 时，搜索包含 +`~/.claude/sessions` 和同一用户拥有的已知 Orca 目录 +`~/Library/Application Support/orca/claude-profiles/*/home/sessions`。 +显式指定目录时保留原有优先级和范围，不扩展到其他档案。只检查一个档案可用 +`CLAUDE_CONFIG_DIR="/path/to/profile/home" session-peer list --agent claude --json`。 +安装路径不变。 + +所有平台的默认及显式注册目录搜索合计最多 4,096 项，每个注册文件最多 1 MiB。 +只有额外的 Orca 档案枚举为 macOS 专用，最多 128 个直接子项。自动搜索的档案目录 +拒绝符号链接、重解析点及其他用户所有权。搜索和 doctor 不读取凭据、收件箱密钥或对话记录。 +同一实际注册目录的别名只搜索一次。不同记录声称同一存活 PID 或收件箱时， +即使公开元数据相同也视为冲突。 + +部分错误时保留正常列表，但**任何搜索错误（包括存活会话冲突）都会拒绝所有目标选择， +即使目标是无关的正常 PID**。此规则适用于 dry-run、发送和 handoff。 +显式目录可隔离其他档案的错误，但不能修复所选目录内部的冲突或搜索限制。 +保留原有单目录对无效记录的处理方式。核对所选记录和端点时不会重新搜索其他档案。 +保留普通 Unix 套接字别名支持及其既有可选代际检查限制；代际令牌不是发送权限或 ACK。 +这些检查不验证模型身份，不消除所有文件系统竞争，也不证明收件。 +不会自动搜索其他 Orca 布局或任意配置目录。 + ### 环境变量 | 变量 | 作用 | diff --git a/session_peer.py b/session_peer.py index 89a4ba6..d045f89 100644 --- a/session_peer.py +++ b/session_peer.py @@ -1996,6 +1996,114 @@ def sessions_dir() -> Path: # Registry records contain identity/inbox metadata, not conversation history. # A 1 MiB ceiling leaves ample room for additional metadata without unbounded IO. MAX_CLAUDE_RECORD_BYTES = 1024 * 1024 +MAX_CLAUDE_PROFILES = 128 +MAX_CLAUDE_RECORDS = 4096 + + +class ClaudeSessions(list): + """List-compatible internal discovery errors; never serialized as rows.""" + def __init__(self): + super().__init__() + self.errors = [] + self.enumerated = 0 + self.invalid_records = 0 + self.permission_failures = 0 + self.scanned = set() + + +class ClaudeSession(dict): + """Native row with non-JSON provenance for the original registry selection.""" + pass + + +def claude_registry_directories(): + """Bounded known Orca inventory, never an arbitrary recursive home scan.""" + selected = sessions_dir() + directories, errors = [selected], [] + if any(os.environ.get(key) for key in ("CLAUDE_CONFIG_DIR", "ANTHROPIC_CONFIG_DIR")): + return directories, errors + # Orca's macOS account-profile location is not a cross-platform convention. + if sys.platform != "darwin" or selected != Path.home() / ".claude/sessions": + return directories, errors + base = Path.home() / "Library/Application Support/orca/claude-profiles" + try: + _claude_safe_directory(base) + with os.scandir(base) as iterator: + entries = [] + for entry in iterator: + if len(entries) >= MAX_CLAUDE_PROFILES: + errors.append("profile_inventory_limit") + break + entries.append(Path(entry.path)) + for profile in sorted(entries): + try: + node = profile.lstat() + if (not stat.S_ISDIR(node.st_mode) and not stat.S_ISLNK(node.st_mode) + and not getattr(node, "st_file_attributes", 0) & 0x400): + continue # Finder/metadata files are not profile homes. + for directory in (profile, profile / "home", profile / "home/sessions"): + _claude_safe_directory(directory) + candidate = profile / "home/sessions" + if candidate not in directories: + directories.append(candidate) + except FileNotFoundError: + continue # A profile need not have started Claude yet. + except (OSError, ValueError): + errors.append("profile_directory_unavailable") + except FileNotFoundError: + pass + except (OSError, ValueError): + errors.append("profile_inventory_unavailable") + return directories, errors + + +def _claude_safe_directory(path, *, same_user=True): + node = path.lstat() + if (not stat.S_ISDIR(node.st_mode) or getattr(node, "st_file_attributes", 0) & 0x400 + or (same_user and hasattr(os, "getuid") and node.st_uid != os.getuid())): + raise ValueError("Unsafe Claude profile directory") + return (node.st_dev, node.st_ino) + + +def claude_selection_directory(session): + return getattr(session, "registry_directory", sessions_dir()) + + +def claude_selection_record(session): + return getattr(session, "registry_record", claude_selection_directory(session) / f"{session['pid']}.json") + + +def claude_endpoint_identity(endpoint): + """Preserve ordinary socket aliases, binding both alias and socket target.""" + path = Path(endpoint) + canonical = path.resolve() + alias, target = path.lstat(), canonical.lstat() + if not stat.S_ISSOCK(target.st_mode): + raise ValueError("Inbox is not a socket") + return (str(canonical), alias.st_dev, alias.st_ino, target.st_dev, target.st_ino) + + +def revalidate_claude_selection(session): + """Do not rediscover or retarget a selected profile after replacement.""" + if not isinstance(session, ClaudeSession): + return # Compatibility for adapter/test callers supplying legacy rows. + try: + directory = session.registry_directory + if (directory.resolve() != session.registry_canonical + or tuple(_claude_safe_directory(path, same_user=session.registry_same_user) + for path in session.registry_chain) != session.registry_identity): + raise ValueError("Registry directory changed") + record = read_claude_record(claude_selection_record(session)) + if (record["pid"] != session["pid"] or record.get("messagingSocketPath") != session["socket"] + or record.get("startedAt") != session.registry_started + or not claude_process_state(record, session["pid"])[0]): + raise ValueError("Registry selection changed") + if not IS_WINDOWS: + if claude_endpoint_identity(session["socket"]) != session.registry_endpoint: + raise ValueError("Inbox changed") + except (OSError, ValueError, KeyError, RuntimeError): + raise generation_refused("target_generation_unavailable", + "Selected Claude profile or inbox changed; discover again; nothing sent") from None def parse_claude_pid(value) -> int | None: @@ -2148,21 +2256,97 @@ def discover(include_unreachable: bool = False) -> list[dict]: under /tmp: a session may bind under $XDG_RUNTIME_DIR, or under a private per-user directory when Claude Code rejects the one it would have used. """ - found: list[dict] = [] - directory = sessions_dir() + found = ClaudeSessions() + directories, found.errors = claude_registry_directories() + found.multiple = len(directories) > 1 or bool(found.errors) + for directory in directories: + _discover_claude_directory(directory, found) + return _finish_claude_discovery(found, include_unreachable) + + +def _finish_claude_discovery(found, include_unreachable): + # Neither PID nor inbox ownership may depend on inventory order. Exact + # duplicates are aliases; divergent claims remain visible but unreachable. + unique, by_pid, by_socket, by_endpoint, conflicts = [], {}, {}, {}, set() + for row in found: + if not row["alive"]: + unique.append(row) + continue # A proven dead/stale process no longer claims an inbox. + endpoint = row.registry_endpoint[-2:] if not IS_WINDOWS and row.registry_endpoint else None + candidates = list({id(item): item for item in ( + by_pid.get(row["pid"], []) + by_socket.get(row["socket"], []) + + (by_endpoint.get(endpoint, []) if endpoint is not None else []))}.values()) + if any(dict(item) == dict(row) and item.registry_canonical == row.registry_canonical + and item.registry_record.name == row.registry_record.name + and item.registry_started == row.registry_started for item in candidates): + continue + for other in candidates: + conflicts.update((id(row), id(other))) + unique.append(row) + by_pid.setdefault(row["pid"], []).append(row) + if row["socket"]: + by_socket.setdefault(row["socket"], []).append(row) + if endpoint is not None: + by_endpoint.setdefault(endpoint, []).append(row) + found.clear() + if conflicts: + found.errors.append("conflicting_session_records") + for row in unique: + if id(row) in conflicts: + row.update(reachable=False, staleReason="conflicting_session_records") + if row["reachable"] or include_unreachable: + found.append(row) + return found + + +def _discover_claude_directory(directory, found): try: - entries = sorted(path for path in directory.iterdir() if path.suffix == ".json") + selected = directory.resolve() + base = Path.home() / "Library/Application Support/orca/claude-profiles" + automatic_profile = directory != sessions_dir() and directory.parent.parent.parent == base + chain = ([base, directory.parent.parent, directory.parent, directory] + if automatic_profile else [selected]) + identity = tuple(_claude_safe_directory(path, same_user=automatic_profile) for path in chain) + scanned = (str(selected), identity[-1]) + if scanned in found.scanned: + return + found.scanned.add(scanned) + entries = [] + with os.scandir(directory) as iterator: + for entry in iterator: + if found.enumerated >= MAX_CLAUDE_RECORDS: + found.errors.append("session_inventory_limit") + break + # Count all entries, not only records, to bound enumeration. + entries.append(Path(entry.path)) + found.enumerated += 1 + entries.sort() except FileNotFoundError: - return found - except OSError as exc: - raise CcPeerError(f"Cannot read Claude sessions at {directory}: {exc}") from exc + if directory != sessions_dir(): + found.errors.append("sessions_directory_unavailable") + return + except (OSError, ValueError, RuntimeError) as exc: + if not found.multiple and directory == sessions_dir(): + raise CcPeerError("Cannot read Claude sessions: selected directory is unavailable") from exc + found.errors.append("sessions_directory_unavailable") + return for record_file in entries: + if record_file.suffix != ".json": + continue record_pid = parse_claude_pid(record_file.stem) if record_pid is None: + if record_file.stem.isdigit(): + found.invalid_records += 1 continue try: record = read_claude_record(record_file) - except (OSError, ValueError): + except (OSError, ValueError) as exc: + if isinstance(exc, PermissionError): + found.permission_failures += 1 + else: + found.invalid_records += 1 + if found.multiple or directory != sessions_dir(): + found.errors.append("session_record_unavailable") continue pid = record.get("pid") @@ -2174,9 +2358,12 @@ def discover(include_unreachable: bool = False) -> list[dict]: if IS_WINDOWS: has_inbox = bool(sock) and sock.startswith("\\\\.\\pipe\\") else: - has_inbox = bool(sock) and Path(sock).is_socket() + try: + has_inbox = bool(sock) and Path(sock).is_socket() + except OSError: + has_inbox = False - entry = { + entry = ClaudeSession({ "agent": "claude", "pid": pid, "name": record.get("name"), @@ -2188,17 +2375,43 @@ def discover(include_unreachable: bool = False) -> list[dict]: "socket": sock, "alive": alive, "reachable": alive and has_inbox, - } + }) + try: + if (directory.resolve() != selected + or tuple(_claude_safe_directory(path, same_user=automatic_profile) + for path in chain) != identity): + raise ValueError("Registry changed") + entry.registry_directory = directory + entry.registry_canonical = selected + entry.registry_chain = chain + entry.registry_same_user = automatic_profile + entry.registry_identity = identity + entry.registry_started = record.get("startedAt") + entry.registry_record = record_file + if not IS_WINDOWS and has_inbox: + try: + entry.registry_endpoint = claude_endpoint_identity(sock) + except (OSError, ValueError, RuntimeError): + # Presence is only a snapshot: a disappearing inbox makes + # this row unreachable, not every other profile ambiguous. + entry.registry_endpoint = None + entry["reachable"] = False + else: + entry.registry_endpoint = None + except (OSError, ValueError, RuntimeError): + found.errors.append("session_record_unavailable") + continue if stale_reason: entry["staleReason"] = stale_reason - if entry["reachable"] or include_unreachable: - found.append(entry) - - return found + found.append(entry) def resolve_target(sessions: list[dict], target: str) -> dict: """Find one session by name, or by pid when the target is all digits.""" + if getattr(sessions, "errors", []): + raise CcPeerError("Claude profile discovery is incomplete; cannot select an unambiguous target; " + "inspect conflicts or limits within the selected directory, or set " + "CLAUDE_CONFIG_DIR to isolate errors in other profiles") reachable = [s for s in sessions if s["reachable"]] if target.isdecimal(): @@ -2245,9 +2458,9 @@ def check_message(text: str, remote: bool) -> None: ) -def _read_win_auth(pid: int) -> str | None: +def _read_win_auth(pid: int, directory=None) -> str | None: """Read the Windows auth key for a session and return the auth JSON line.""" - directory = sessions_dir() + directory = sessions_dir() if directory is None else directory for key_file in directory.glob(f"{pid}.*.key"): try: data = json.loads(key_file.read_text(encoding="utf-8")) @@ -2270,7 +2483,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=Non """Write one message to a Windows named pipe inbox.""" check_message(text, remote=False) - auth_line = _read_win_auth(pid) + auth_line = (_read_win_auth(pid, claude_selection_directory(generation_session)) + if isinstance(generation_session, ClaudeSession) else _read_win_auth(pid)) if auth_line is None: raise CcPeerError(f"no auth key found for pid {pid} — cannot post to Windows pipe") @@ -2282,6 +2496,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=Non import time try: with open(pipe_path, "wb") as pipe: + if generation_session is not None: + revalidate_claude_selection(generation_session) if expected_generation is not None: verify_connected_generation(generation_session, expected_generation, connected_pipe_pid(pipe)) pipe.write((auth_line + "\n").encode("utf-8")) @@ -2303,6 +2519,8 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_sess a complete line within 30 seconds, so the message is built before the socket is opened. """ + if generation_session is not None: + revalidate_claude_selection(generation_session) bounded = effect_deadline is not None or total_deadline is not None if bounded: import math @@ -2342,6 +2560,8 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_sess raise CcPeerError(f"cannot reach inbox at {socket_path}: {exc}") from exc try: + if generation_session is not None: + revalidate_claude_selection(generation_session) if expected_generation is not None: verify_connected_generation(generation_session, expected_generation, connected_inbox_pid(conn)) if bounded: @@ -2426,8 +2646,9 @@ def claude_generation(session: dict) -> str | None: birth = process_generation(pid) if not birth: return None - directory = sessions_dir().resolve() - record = read_claude_record(directory / f"{pid}.json") + revalidate_claude_selection(session) + directory = claude_selection_directory(session).resolve() + record = read_claude_record(claude_selection_record(session)) endpoint = record.get("messagingSocketPath") if record["pid"] != pid or endpoint != session["socket"]: return None @@ -2443,7 +2664,7 @@ def claude_generation(session: dict) -> str | None: identity += [record.get("startedAt")] return "tg1:" + hashlib.sha256(json.dumps(identity, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest() - except (OSError, ValueError, KeyError, RuntimeError): + except (OSError, ValueError, KeyError, RuntimeError, CcPeerError): return None @@ -4204,119 +4425,68 @@ def _diagnostic(status: str, code: str, message: str, **detail) -> dict: def diagnose_claude() -> dict: - directory = sessions_dir() + directories, errors = claude_registry_directories() result = { - "sessionsDir": str(directory), "records": 0, "invalidRecords": 0, + "sessionsDir": str(sessions_dir()), "records": 0, "invalidRecords": 0, "aliveSessions": 0, "availableInboxes": 0, "staleRecords": 0, "checks": [], } - try: - mode = directory.stat().st_mode - except FileNotFoundError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_missing", - "Claude sessions directory does not exist; check CLAUDE_CONFIG_DIR", - )) - return {"status": "missing_home", **result} - except PermissionError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_permission_denied", - "Claude sessions directory is not readable by this user", - )) - return {"status": "permission_denied", **result} - except OSError as exc: - result["checks"].append(_diagnostic( - "unknown", "sessions_dir_unreadable", f"Cannot inspect Claude sessions: {exc}", - )) - return {"status": "unknown", **result} - if not stat.S_ISDIR(mode): - result["checks"].append(_diagnostic( - "error", "sessions_path_not_directory", - "Configured Claude sessions path is not a directory", - )) - return {"status": "wrong_home", **result} - try: - entries = sorted( - entry for entry in directory.iterdir() - if entry.name.endswith(".json") - ) - except PermissionError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_permission_denied", - "Claude sessions directory cannot be listed by this user", - )) - return {"status": "permission_denied", **result} - except OSError as exc: - result["checks"].append(_diagnostic( - "unknown", "sessions_dir_unreadable", f"Cannot list Claude sessions: {exc}", - )) - return {"status": "unknown", **result} - - permission_failures = 0 - for record_file in entries: - record_pid = parse_claude_pid(record_file.stem) - if record_pid is None: - if record_file.stem.isdigit(): - result["invalidRecords"] += 1 - continue + # Preserve legacy single-root diagnostics without a second/unbounded reader. + if len(directories) == 1 and not errors: try: - record = read_claude_record(record_file) + mode = directories[0].stat().st_mode + except FileNotFoundError: + result["checks"].append(_diagnostic("error", "sessions_dir_missing", + "Claude sessions directory does not exist; check CLAUDE_CONFIG_DIR")) + return {"status": "missing_home", **result} except PermissionError: - permission_failures += 1 - continue - except (OSError, ValueError): - result["invalidRecords"] += 1 - continue - pid = record.get("pid") - if not isinstance(pid, int): - result["invalidRecords"] += 1 - continue - result["records"] += 1 - alive, stale_reason = claude_process_state(record, record_pid) - if stale_reason: - result["staleRecords"] += 1 - if alive: - result["aliveSessions"] += 1 - inbox = str(record.get("messagingSocketPath") or "") - try: - present = ( - inbox.startswith("\\\\.\\pipe\\") if IS_WINDOWS - else bool(inbox) and Path(inbox).is_socket() - ) + result["checks"].append(_diagnostic("error", "sessions_dir_permission_denied", + "Claude sessions directory is not readable by this user")) + return {"status": "permission_denied", **result} except OSError: - present = False - if alive and present: - result["availableInboxes"] += 1 - - if permission_failures: - result["checks"].append(_diagnostic( - "error", "session_record_permission_denied", - "One or more Claude session records are not readable", - count=permission_failures, - )) + result["checks"].append(_diagnostic("unknown", "sessions_dir_unreadable", + "Cannot inspect Claude sessions")) + return {"status": "unknown", **result} + if not stat.S_ISDIR(mode): + result["checks"].append(_diagnostic("error", "sessions_path_not_directory", + "Configured Claude sessions path is not a directory")) + return {"status": "wrong_home", **result} + rows = ClaudeSessions() + rows.errors = list(errors) + rows.multiple = len(directories) > 1 or bool(errors) + try: + for directory in directories: + _discover_claude_directory(directory, rows) + except CcPeerError as exc: + denied = isinstance(exc.__cause__, PermissionError) + result["checks"].append(_diagnostic("error", "sessions_dir_permission_denied" if denied else "sessions_dir_unreadable", + "Cannot inspect the selected Claude registry")) + return {"status": "permission_denied" if denied else "unknown", **result} + rows = _finish_claude_discovery(rows, include_unreachable=True) + result.update(records=len(rows), invalidRecords=rows.invalid_records, + aliveSessions=sum(row["alive"] for row in rows), + availableInboxes=sum(row["reachable"] for row in rows), + staleRecords=sum(bool(row.get("staleReason")) for row in rows)) + if rows.permission_failures: + result["checks"].append(_diagnostic("error", "session_record_permission_denied", + "One or more Claude session records are not readable", count=rows.permission_failures)) return {"status": "permission_denied", **result} + if rows.errors: + result["checks"].append(_diagnostic("error", "claude_discovery_incomplete", + "Some Claude profiles or records could not be inspected", reasons=sorted(set(rows.errors)))) + return {"status": "unknown", **result} if result["staleRecords"]: - result["checks"].append(_diagnostic( - "warning", "stale_session_records", - "Claude session records do not match live process identities", - count=result["staleRecords"], - )) + result["checks"].append(_diagnostic("warning", "stale_session_records", + "Claude session records do not match live process identities", count=result["staleRecords"])) if result["availableInboxes"]: - result["checks"].append(_diagnostic( - "ok", "inbox_present", - "At least one live Claude session advertises an inbox", - verification="filesystem_only", - )) + result["checks"].append(_diagnostic("ok", "inbox_present", + "At least one live Claude session advertises an inbox", verification="filesystem_only")) return {"status": "available", **result} if result["aliveSessions"]: - result["checks"].append(_diagnostic( - "warning", "inbox_unavailable", - "Live Claude sessions exist but none advertises an available inbox", - )) + result["checks"].append(_diagnostic("warning", "inbox_unavailable", + "Live Claude sessions exist but none advertises an available inbox")) return {"status": "inbox_unavailable", **result} - result["checks"].append(_diagnostic( - "warning", "no_live_sessions", - "No live Claude session with an inbox was found", - )) + result["checks"].append(_diagnostic("warning", "no_live_sessions", + "No live Claude session with an inbox was found")) return {"status": "unavailable", **result} @@ -4689,9 +4859,12 @@ def identity(self, target: str, context: ExecutionContext) -> SessionIdentity: return SessionIdentity(self.name, context.host, target) def list(self, context: ExecutionContext) -> DiscoveryResult: - return {"sessions": [{**row, "agent": self.name} for row in - discover(include_unreachable=context.options.all)], - "discovery": {"status": "ok"}} + rows = discover(include_unreachable=context.options.all) + errors = getattr(rows, "errors", []) + return {"sessions": [row if isinstance(row, ClaudeSession) else {**row, "agent": self.name} + for row in rows], + "discovery": ({"status": "error", "error": "Claude profile discovery incomplete: " + + ", ".join(sorted(set(errors)))} if errors else {"status": "ok"})} def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: args = context.options @@ -4705,12 +4878,16 @@ def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: raise if expected is not None: require_claude_generation(session, expected) + revalidate_claude_selection(session) if not args.dry_run: if expected is not None: post_to_socket(session["socket"], text, pid=session["pid"], generation_session=session, expected_generation=expected) else: - post_to_socket(session["socket"], text, pid=session["pid"]) + if isinstance(session, ClaudeSession): + post_to_socket(session["socket"], text, pid=session["pid"], generation_session=session) + else: + post_to_socket(session["socket"], text, pid=session["pid"]) return {"ok": True, "target": {"pid": session["pid"], "name": session["name"]}, "chars": len(text), "dryRun": args.dry_run, **({"targetGeneration": expected} if expected is not None else {})} diff --git a/session_peer_core/adapters.py b/session_peer_core/adapters.py index 45cdf39..b60509b 100644 --- a/session_peer_core/adapters.py +++ b/session_peer_core/adapters.py @@ -126,9 +126,12 @@ def identity(self, target: str, context: ExecutionContext) -> SessionIdentity: return SessionIdentity(self.name, context.host, target) def list(self, context: ExecutionContext) -> DiscoveryResult: - return {"sessions": [{**row, "agent": self.name} for row in - discover(include_unreachable=context.options.all)], - "discovery": {"status": "ok"}} + rows = discover(include_unreachable=context.options.all) + errors = getattr(rows, "errors", []) + return {"sessions": [row if isinstance(row, ClaudeSession) else {**row, "agent": self.name} + for row in rows], + "discovery": ({"status": "error", "error": "Claude profile discovery incomplete: " + + ", ".join(sorted(set(errors)))} if errors else {"status": "ok"})} def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: args = context.options @@ -142,12 +145,16 @@ def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: raise if expected is not None: require_claude_generation(session, expected) + revalidate_claude_selection(session) if not args.dry_run: if expected is not None: post_to_socket(session["socket"], text, pid=session["pid"], generation_session=session, expected_generation=expected) else: - post_to_socket(session["socket"], text, pid=session["pid"]) + if isinstance(session, ClaudeSession): + post_to_socket(session["socket"], text, pid=session["pid"], generation_session=session) + else: + post_to_socket(session["socket"], text, pid=session["pid"]) return {"ok": True, "target": {"pid": session["pid"], "name": session["name"]}, "chars": len(text), "dryRun": args.dry_run, **({"targetGeneration": expected} if expected is not None else {})} diff --git a/session_peer_core/claude.py b/session_peer_core/claude.py index 115b495..496e64d 100644 --- a/session_peer_core/claude.py +++ b/session_peer_core/claude.py @@ -21,6 +21,114 @@ def sessions_dir() -> Path: # Registry records contain identity/inbox metadata, not conversation history. # A 1 MiB ceiling leaves ample room for additional metadata without unbounded IO. MAX_CLAUDE_RECORD_BYTES = 1024 * 1024 +MAX_CLAUDE_PROFILES = 128 +MAX_CLAUDE_RECORDS = 4096 + + +class ClaudeSessions(list): + """List-compatible internal discovery errors; never serialized as rows.""" + def __init__(self): + super().__init__() + self.errors = [] + self.enumerated = 0 + self.invalid_records = 0 + self.permission_failures = 0 + self.scanned = set() + + +class ClaudeSession(dict): + """Native row with non-JSON provenance for the original registry selection.""" + pass + + +def claude_registry_directories(): + """Bounded known Orca inventory, never an arbitrary recursive home scan.""" + selected = sessions_dir() + directories, errors = [selected], [] + if any(os.environ.get(key) for key in ("CLAUDE_CONFIG_DIR", "ANTHROPIC_CONFIG_DIR")): + return directories, errors + # Orca's macOS account-profile location is not a cross-platform convention. + if sys.platform != "darwin" or selected != Path.home() / ".claude/sessions": + return directories, errors + base = Path.home() / "Library/Application Support/orca/claude-profiles" + try: + _claude_safe_directory(base) + with os.scandir(base) as iterator: + entries = [] + for entry in iterator: + if len(entries) >= MAX_CLAUDE_PROFILES: + errors.append("profile_inventory_limit") + break + entries.append(Path(entry.path)) + for profile in sorted(entries): + try: + node = profile.lstat() + if (not stat.S_ISDIR(node.st_mode) and not stat.S_ISLNK(node.st_mode) + and not getattr(node, "st_file_attributes", 0) & 0x400): + continue # Finder/metadata files are not profile homes. + for directory in (profile, profile / "home", profile / "home/sessions"): + _claude_safe_directory(directory) + candidate = profile / "home/sessions" + if candidate not in directories: + directories.append(candidate) + except FileNotFoundError: + continue # A profile need not have started Claude yet. + except (OSError, ValueError): + errors.append("profile_directory_unavailable") + except FileNotFoundError: + pass + except (OSError, ValueError): + errors.append("profile_inventory_unavailable") + return directories, errors + + +def _claude_safe_directory(path, *, same_user=True): + node = path.lstat() + if (not stat.S_ISDIR(node.st_mode) or getattr(node, "st_file_attributes", 0) & 0x400 + or (same_user and hasattr(os, "getuid") and node.st_uid != os.getuid())): + raise ValueError("Unsafe Claude profile directory") + return (node.st_dev, node.st_ino) + + +def claude_selection_directory(session): + return getattr(session, "registry_directory", sessions_dir()) + + +def claude_selection_record(session): + return getattr(session, "registry_record", claude_selection_directory(session) / f"{session['pid']}.json") + + +def claude_endpoint_identity(endpoint): + """Preserve ordinary socket aliases, binding both alias and socket target.""" + path = Path(endpoint) + canonical = path.resolve() + alias, target = path.lstat(), canonical.lstat() + if not stat.S_ISSOCK(target.st_mode): + raise ValueError("Inbox is not a socket") + return (str(canonical), alias.st_dev, alias.st_ino, target.st_dev, target.st_ino) + + +def revalidate_claude_selection(session): + """Do not rediscover or retarget a selected profile after replacement.""" + if not isinstance(session, ClaudeSession): + return # Compatibility for adapter/test callers supplying legacy rows. + try: + directory = session.registry_directory + if (directory.resolve() != session.registry_canonical + or tuple(_claude_safe_directory(path, same_user=session.registry_same_user) + for path in session.registry_chain) != session.registry_identity): + raise ValueError("Registry directory changed") + record = read_claude_record(claude_selection_record(session)) + if (record["pid"] != session["pid"] or record.get("messagingSocketPath") != session["socket"] + or record.get("startedAt") != session.registry_started + or not claude_process_state(record, session["pid"])[0]): + raise ValueError("Registry selection changed") + if not IS_WINDOWS: + if claude_endpoint_identity(session["socket"]) != session.registry_endpoint: + raise ValueError("Inbox changed") + except (OSError, ValueError, KeyError, RuntimeError): + raise generation_refused("target_generation_unavailable", + "Selected Claude profile or inbox changed; discover again; nothing sent") from None def parse_claude_pid(value) -> int | None: @@ -173,21 +281,97 @@ def discover(include_unreachable: bool = False) -> list[dict]: under /tmp: a session may bind under $XDG_RUNTIME_DIR, or under a private per-user directory when Claude Code rejects the one it would have used. """ - found: list[dict] = [] - directory = sessions_dir() + found = ClaudeSessions() + directories, found.errors = claude_registry_directories() + found.multiple = len(directories) > 1 or bool(found.errors) + for directory in directories: + _discover_claude_directory(directory, found) + return _finish_claude_discovery(found, include_unreachable) + + +def _finish_claude_discovery(found, include_unreachable): + # Neither PID nor inbox ownership may depend on inventory order. Exact + # duplicates are aliases; divergent claims remain visible but unreachable. + unique, by_pid, by_socket, by_endpoint, conflicts = [], {}, {}, {}, set() + for row in found: + if not row["alive"]: + unique.append(row) + continue # A proven dead/stale process no longer claims an inbox. + endpoint = row.registry_endpoint[-2:] if not IS_WINDOWS and row.registry_endpoint else None + candidates = list({id(item): item for item in ( + by_pid.get(row["pid"], []) + by_socket.get(row["socket"], []) + + (by_endpoint.get(endpoint, []) if endpoint is not None else []))}.values()) + if any(dict(item) == dict(row) and item.registry_canonical == row.registry_canonical + and item.registry_record.name == row.registry_record.name + and item.registry_started == row.registry_started for item in candidates): + continue + for other in candidates: + conflicts.update((id(row), id(other))) + unique.append(row) + by_pid.setdefault(row["pid"], []).append(row) + if row["socket"]: + by_socket.setdefault(row["socket"], []).append(row) + if endpoint is not None: + by_endpoint.setdefault(endpoint, []).append(row) + found.clear() + if conflicts: + found.errors.append("conflicting_session_records") + for row in unique: + if id(row) in conflicts: + row.update(reachable=False, staleReason="conflicting_session_records") + if row["reachable"] or include_unreachable: + found.append(row) + return found + + +def _discover_claude_directory(directory, found): try: - entries = sorted(path for path in directory.iterdir() if path.suffix == ".json") + selected = directory.resolve() + base = Path.home() / "Library/Application Support/orca/claude-profiles" + automatic_profile = directory != sessions_dir() and directory.parent.parent.parent == base + chain = ([base, directory.parent.parent, directory.parent, directory] + if automatic_profile else [selected]) + identity = tuple(_claude_safe_directory(path, same_user=automatic_profile) for path in chain) + scanned = (str(selected), identity[-1]) + if scanned in found.scanned: + return + found.scanned.add(scanned) + entries = [] + with os.scandir(directory) as iterator: + for entry in iterator: + if found.enumerated >= MAX_CLAUDE_RECORDS: + found.errors.append("session_inventory_limit") + break + # Count all entries, not only records, to bound enumeration. + entries.append(Path(entry.path)) + found.enumerated += 1 + entries.sort() except FileNotFoundError: - return found - except OSError as exc: - raise CcPeerError(f"Cannot read Claude sessions at {directory}: {exc}") from exc + if directory != sessions_dir(): + found.errors.append("sessions_directory_unavailable") + return + except (OSError, ValueError, RuntimeError) as exc: + if not found.multiple and directory == sessions_dir(): + raise CcPeerError("Cannot read Claude sessions: selected directory is unavailable") from exc + found.errors.append("sessions_directory_unavailable") + return for record_file in entries: + if record_file.suffix != ".json": + continue record_pid = parse_claude_pid(record_file.stem) if record_pid is None: + if record_file.stem.isdigit(): + found.invalid_records += 1 continue try: record = read_claude_record(record_file) - except (OSError, ValueError): + except (OSError, ValueError) as exc: + if isinstance(exc, PermissionError): + found.permission_failures += 1 + else: + found.invalid_records += 1 + if found.multiple or directory != sessions_dir(): + found.errors.append("session_record_unavailable") continue pid = record.get("pid") @@ -199,9 +383,12 @@ def discover(include_unreachable: bool = False) -> list[dict]: if IS_WINDOWS: has_inbox = bool(sock) and sock.startswith("\\\\.\\pipe\\") else: - has_inbox = bool(sock) and Path(sock).is_socket() + try: + has_inbox = bool(sock) and Path(sock).is_socket() + except OSError: + has_inbox = False - entry = { + entry = ClaudeSession({ "agent": "claude", "pid": pid, "name": record.get("name"), @@ -213,17 +400,43 @@ def discover(include_unreachable: bool = False) -> list[dict]: "socket": sock, "alive": alive, "reachable": alive and has_inbox, - } + }) + try: + if (directory.resolve() != selected + or tuple(_claude_safe_directory(path, same_user=automatic_profile) + for path in chain) != identity): + raise ValueError("Registry changed") + entry.registry_directory = directory + entry.registry_canonical = selected + entry.registry_chain = chain + entry.registry_same_user = automatic_profile + entry.registry_identity = identity + entry.registry_started = record.get("startedAt") + entry.registry_record = record_file + if not IS_WINDOWS and has_inbox: + try: + entry.registry_endpoint = claude_endpoint_identity(sock) + except (OSError, ValueError, RuntimeError): + # Presence is only a snapshot: a disappearing inbox makes + # this row unreachable, not every other profile ambiguous. + entry.registry_endpoint = None + entry["reachable"] = False + else: + entry.registry_endpoint = None + except (OSError, ValueError, RuntimeError): + found.errors.append("session_record_unavailable") + continue if stale_reason: entry["staleReason"] = stale_reason - if entry["reachable"] or include_unreachable: - found.append(entry) - - return found + found.append(entry) def resolve_target(sessions: list[dict], target: str) -> dict: """Find one session by name, or by pid when the target is all digits.""" + if getattr(sessions, "errors", []): + raise CcPeerError("Claude profile discovery is incomplete; cannot select an unambiguous target; " + "inspect conflicts or limits within the selected directory, or set " + "CLAUDE_CONFIG_DIR to isolate errors in other profiles") reachable = [s for s in sessions if s["reachable"]] if target.isdecimal(): @@ -270,9 +483,9 @@ def check_message(text: str, remote: bool) -> None: ) -def _read_win_auth(pid: int) -> str | None: +def _read_win_auth(pid: int, directory=None) -> str | None: """Read the Windows auth key for a session and return the auth JSON line.""" - directory = sessions_dir() + directory = sessions_dir() if directory is None else directory for key_file in directory.glob(f"{pid}.*.key"): try: data = json.loads(key_file.read_text(encoding="utf-8")) @@ -295,7 +508,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=Non """Write one message to a Windows named pipe inbox.""" check_message(text, remote=False) - auth_line = _read_win_auth(pid) + auth_line = (_read_win_auth(pid, claude_selection_directory(generation_session)) + if isinstance(generation_session, ClaudeSession) else _read_win_auth(pid)) if auth_line is None: raise CcPeerError(f"no auth key found for pid {pid} — cannot post to Windows pipe") @@ -307,6 +521,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=Non import time try: with open(pipe_path, "wb") as pipe: + if generation_session is not None: + revalidate_claude_selection(generation_session) if expected_generation is not None: verify_connected_generation(generation_session, expected_generation, connected_pipe_pid(pipe)) pipe.write((auth_line + "\n").encode("utf-8")) @@ -328,6 +544,8 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_sess a complete line within 30 seconds, so the message is built before the socket is opened. """ + if generation_session is not None: + revalidate_claude_selection(generation_session) bounded = effect_deadline is not None or total_deadline is not None if bounded: import math @@ -367,6 +585,8 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_sess raise CcPeerError(f"cannot reach inbox at {socket_path}: {exc}") from exc try: + if generation_session is not None: + revalidate_claude_selection(generation_session) if expected_generation is not None: verify_connected_generation(generation_session, expected_generation, connected_inbox_pid(conn)) if bounded: diff --git a/session_peer_core/diagnostics.py b/session_peer_core/diagnostics.py index 2ebfe6e..2c3f386 100644 --- a/session_peer_core/diagnostics.py +++ b/session_peer_core/diagnostics.py @@ -9,119 +9,68 @@ def _diagnostic(status: str, code: str, message: str, **detail) -> dict: def diagnose_claude() -> dict: - directory = sessions_dir() + directories, errors = claude_registry_directories() result = { - "sessionsDir": str(directory), "records": 0, "invalidRecords": 0, + "sessionsDir": str(sessions_dir()), "records": 0, "invalidRecords": 0, "aliveSessions": 0, "availableInboxes": 0, "staleRecords": 0, "checks": [], } - try: - mode = directory.stat().st_mode - except FileNotFoundError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_missing", - "Claude sessions directory does not exist; check CLAUDE_CONFIG_DIR", - )) - return {"status": "missing_home", **result} - except PermissionError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_permission_denied", - "Claude sessions directory is not readable by this user", - )) - return {"status": "permission_denied", **result} - except OSError as exc: - result["checks"].append(_diagnostic( - "unknown", "sessions_dir_unreadable", f"Cannot inspect Claude sessions: {exc}", - )) - return {"status": "unknown", **result} - if not stat.S_ISDIR(mode): - result["checks"].append(_diagnostic( - "error", "sessions_path_not_directory", - "Configured Claude sessions path is not a directory", - )) - return {"status": "wrong_home", **result} - try: - entries = sorted( - entry for entry in directory.iterdir() - if entry.name.endswith(".json") - ) - except PermissionError: - result["checks"].append(_diagnostic( - "error", "sessions_dir_permission_denied", - "Claude sessions directory cannot be listed by this user", - )) - return {"status": "permission_denied", **result} - except OSError as exc: - result["checks"].append(_diagnostic( - "unknown", "sessions_dir_unreadable", f"Cannot list Claude sessions: {exc}", - )) - return {"status": "unknown", **result} - - permission_failures = 0 - for record_file in entries: - record_pid = parse_claude_pid(record_file.stem) - if record_pid is None: - if record_file.stem.isdigit(): - result["invalidRecords"] += 1 - continue + # Preserve legacy single-root diagnostics without a second/unbounded reader. + if len(directories) == 1 and not errors: try: - record = read_claude_record(record_file) + mode = directories[0].stat().st_mode + except FileNotFoundError: + result["checks"].append(_diagnostic("error", "sessions_dir_missing", + "Claude sessions directory does not exist; check CLAUDE_CONFIG_DIR")) + return {"status": "missing_home", **result} except PermissionError: - permission_failures += 1 - continue - except (OSError, ValueError): - result["invalidRecords"] += 1 - continue - pid = record.get("pid") - if not isinstance(pid, int): - result["invalidRecords"] += 1 - continue - result["records"] += 1 - alive, stale_reason = claude_process_state(record, record_pid) - if stale_reason: - result["staleRecords"] += 1 - if alive: - result["aliveSessions"] += 1 - inbox = str(record.get("messagingSocketPath") or "") - try: - present = ( - inbox.startswith("\\\\.\\pipe\\") if IS_WINDOWS - else bool(inbox) and Path(inbox).is_socket() - ) + result["checks"].append(_diagnostic("error", "sessions_dir_permission_denied", + "Claude sessions directory is not readable by this user")) + return {"status": "permission_denied", **result} except OSError: - present = False - if alive and present: - result["availableInboxes"] += 1 - - if permission_failures: - result["checks"].append(_diagnostic( - "error", "session_record_permission_denied", - "One or more Claude session records are not readable", - count=permission_failures, - )) + result["checks"].append(_diagnostic("unknown", "sessions_dir_unreadable", + "Cannot inspect Claude sessions")) + return {"status": "unknown", **result} + if not stat.S_ISDIR(mode): + result["checks"].append(_diagnostic("error", "sessions_path_not_directory", + "Configured Claude sessions path is not a directory")) + return {"status": "wrong_home", **result} + rows = ClaudeSessions() + rows.errors = list(errors) + rows.multiple = len(directories) > 1 or bool(errors) + try: + for directory in directories: + _discover_claude_directory(directory, rows) + except CcPeerError as exc: + denied = isinstance(exc.__cause__, PermissionError) + result["checks"].append(_diagnostic("error", "sessions_dir_permission_denied" if denied else "sessions_dir_unreadable", + "Cannot inspect the selected Claude registry")) + return {"status": "permission_denied" if denied else "unknown", **result} + rows = _finish_claude_discovery(rows, include_unreachable=True) + result.update(records=len(rows), invalidRecords=rows.invalid_records, + aliveSessions=sum(row["alive"] for row in rows), + availableInboxes=sum(row["reachable"] for row in rows), + staleRecords=sum(bool(row.get("staleReason")) for row in rows)) + if rows.permission_failures: + result["checks"].append(_diagnostic("error", "session_record_permission_denied", + "One or more Claude session records are not readable", count=rows.permission_failures)) return {"status": "permission_denied", **result} + if rows.errors: + result["checks"].append(_diagnostic("error", "claude_discovery_incomplete", + "Some Claude profiles or records could not be inspected", reasons=sorted(set(rows.errors)))) + return {"status": "unknown", **result} if result["staleRecords"]: - result["checks"].append(_diagnostic( - "warning", "stale_session_records", - "Claude session records do not match live process identities", - count=result["staleRecords"], - )) + result["checks"].append(_diagnostic("warning", "stale_session_records", + "Claude session records do not match live process identities", count=result["staleRecords"])) if result["availableInboxes"]: - result["checks"].append(_diagnostic( - "ok", "inbox_present", - "At least one live Claude session advertises an inbox", - verification="filesystem_only", - )) + result["checks"].append(_diagnostic("ok", "inbox_present", + "At least one live Claude session advertises an inbox", verification="filesystem_only")) return {"status": "available", **result} if result["aliveSessions"]: - result["checks"].append(_diagnostic( - "warning", "inbox_unavailable", - "Live Claude sessions exist but none advertises an available inbox", - )) + result["checks"].append(_diagnostic("warning", "inbox_unavailable", + "Live Claude sessions exist but none advertises an available inbox")) return {"status": "inbox_unavailable", **result} - result["checks"].append(_diagnostic( - "warning", "no_live_sessions", - "No live Claude session with an inbox was found", - )) + result["checks"].append(_diagnostic("warning", "no_live_sessions", + "No live Claude session with an inbox was found")) return {"status": "unavailable", **result} diff --git a/session_peer_core/target_generation.py b/session_peer_core/target_generation.py index ba8e0d6..1aed5f8 100644 --- a/session_peer_core/target_generation.py +++ b/session_peer_core/target_generation.py @@ -55,8 +55,9 @@ def claude_generation(session: dict) -> str | None: birth = process_generation(pid) if not birth: return None - directory = sessions_dir().resolve() - record = read_claude_record(directory / f"{pid}.json") + revalidate_claude_selection(session) + directory = claude_selection_directory(session).resolve() + record = read_claude_record(claude_selection_record(session)) endpoint = record.get("messagingSocketPath") if record["pid"] != pid or endpoint != session["socket"]: return None @@ -72,7 +73,7 @@ def claude_generation(session: dict) -> str | None: identity += [record.get("startedAt")] return "tg1:" + hashlib.sha256(json.dumps(identity, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest() - except (OSError, ValueError, KeyError, RuntimeError): + except (OSError, ValueError, KeyError, RuntimeError, CcPeerError): return None diff --git a/tests/core/test_claude_profiles.py b/tests/core/test_claude_profiles.py new file mode 100644 index 0000000..4ef4783 --- /dev/null +++ b/tests/core/test_claude_profiles.py @@ -0,0 +1,615 @@ +"""Offline known-profile inventory; no native inbox connections or credentials.""" +import argparse +import base64 +import contextlib +import io +import json +import os +from pathlib import Path +import socket +import tempfile +import unittest +from unittest import mock + +import session_peer as peer + + +class ClaudeProfiles(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory(prefix="sp317-") + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.default = self.root / ".claude/sessions" + self.default.mkdir(parents=True) + self.base = self.root / "Library/Application Support/orca/claude-profiles" + self.base.mkdir(parents=True) + self.counter = 0 + for patcher in ( + mock.patch.object(Path, "home", return_value=self.root), + mock.patch.object(peer.sys, "platform", "darwin"), + mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": "", "ANTHROPIC_CONFIG_DIR": ""}), + mock.patch.object(peer, "pid_alive", return_value=True), + mock.patch.object(peer, "windows_process_start_ms", return_value=1000), + mock.patch.object(peer, "process_generation", return_value="fixture-birth"), + ): + patcher.start() + self.addCleanup(patcher.stop) + + def profile(self, name): + directory = self.base / name / "home/sessions" + directory.mkdir(parents=True) + return directory + + def record(self, directory, pid, name="worker", inbox=None): + if inbox is None: + self.counter += 1 + if peer.IS_WINDOWS: + inbox = r"\\.\pipe\fixture-" + str(self.counter) + else: + inbox = str(self.root / (str(self.counter) + ".sock")) + listener = socket.socket(socket.AF_UNIX) + listener.bind(inbox) + self.addCleanup(listener.close) + record = {"pid": pid, "name": name, "startedAt": 1000, + "cwd": "/fixture", "status": "idle", "messagingSocketPath": inbox} + (directory / (str(pid) + ".json")).write_text(json.dumps(record), encoding="utf-8") + return record + + def test_default_and_multiple_orca_profiles_without_private_json_fields(self): + self.record(self.default, 100, "default") + self.record(self.profile("one"), 101, "one") + self.record(self.profile("two"), 102, "two") + with mock.patch.object(peer, "_read_win_auth") as auth: + rows = peer.discover() + diagnostic = peer.diagnose_claude() + self.assertEqual({row["pid"] for row in rows}, {100, 101, 102}) + self.assertEqual(rows.errors, []) + self.assertEqual(diagnostic["availableInboxes"], 3) + self.assertEqual(diagnostic["status"], "available") + self.assertNotIn("registry", json.dumps(rows)) + self.assertNotIn("claude-profiles", json.dumps(rows)) + auth.assert_not_called() + + def test_explicit_environment_scope_and_precedence(self): + first, second = self.profile("first"), self.profile("second") + self.record(self.default, 100) + self.record(first, 101) + self.record(second, 102) + for primary in (True, False): + with self.subTest(primary=primary), mock.patch.dict(os.environ, { + "CLAUDE_CONFIG_DIR": str(first.parent) if primary else "", + "ANTHROPIC_CONFIG_DIR": str(second.parent), + }), mock.patch.object(peer.os, "scandir", wraps=os.scandir) as scan: + rows = peer.discover() + self.assertEqual([row["pid"] for row in rows], [101 if primary else 102]) + self.assertNotIn(mock.call(self.base), scan.call_args_list) + + def test_non_macos_does_not_guess_orca_locations(self): + self.record(self.profile("one"), 101) + with mock.patch.object(peer.sys, "platform", "linux"): + self.assertEqual(peer.discover(), []) + + def test_missing_default_and_unstarted_profiles_are_not_errors(self): + self.default.rmdir() + (self.base / "unstarted").mkdir() + self.record(self.profile("ready"), 101) + rows = peer.discover() + self.assertEqual([row["pid"] for row in rows], [101]) + self.assertEqual(rows.errors, []) + + def test_profile_metadata_files_are_not_opened_or_treated_as_homes(self): + self.record(self.profile("ready"), 101) + for name in (".DS_Store", "metadata.txt", "account.key"): + (self.base / name).write_text("not a profile", encoding="utf-8") + original = os.open + def read(path, *args, **kwargs): + self.assertEqual(Path(path).suffix, ".json", "Only registry JSON may be opened") + return original(path, *args, **kwargs) + with mock.patch.object(peer.os, "open", side_effect=read), \ + mock.patch.object(peer, "_read_win_auth") as auth: + rows = peer.discover() + self.assertEqual(peer.diagnose_claude()["status"], "available") + self.assertEqual([row["pid"] for row in rows], [101]) + self.assertEqual(rows.errors, []) + auth.assert_not_called() + + def test_malformed_oversized_and_unreadable_records_keep_healthy_rows(self): + good, bad = self.profile("good"), self.profile("bad") + self.record(good, 101) + (bad / "102.json").write_text("{", encoding="utf-8") + (bad / "103.json").write_bytes(b"x" * (peer.MAX_CLAUDE_RECORD_BYTES + 1)) + self.record(bad, 104) + original = peer.read_claude_record + def read(path): + if path.name == "104.json": + raise PermissionError("fixture") + return original(path) + with mock.patch.object(peer, "read_claude_record", side_effect=read): + result = peer.ClaudeAdapter().list(peer.ExecutionContext("local", argparse.Namespace(all=False))) + diagnostic = peer.diagnose_claude() + self.assertEqual([row["pid"] for row in result["sessions"]], [101]) + self.assertEqual(result["discovery"]["status"], "error") + self.assertEqual(diagnostic["status"], "permission_denied") + self.assertEqual(diagnostic["availableInboxes"], 1) + self.assertNotIn(str(bad), json.dumps(result["discovery"])) + + def test_incomplete_inventory_refuses_dry_run_send_and_handoff(self): + self.record(self.profile("good"), 101) + (self.profile("bad") / "102.json").write_text("{", encoding="utf-8") + with mock.patch.object(peer, "post_to_socket") as post: + for dry in (False, True): + args = argparse.Namespace(to="101", dry_run=dry) + with self.subTest(dry=dry), self.assertRaises(peer.CcPeerError): + peer.ClaudeAdapter().submit(peer.ExecutionContext("local", args), "fixture") + with self.assertRaises(peer.CcPeerError): + peer.handoff_binding(argparse.Namespace(to="101"), "fixture") + post.assert_not_called() + + def test_doctor_uses_same_bound_and_conflict_rules(self): + self.record(self.default, 100) + with mock.patch.object(peer, "MAX_CLAUDE_RECORDS", 0): + self.assertEqual(peer.diagnose_claude()["status"], "unknown") + one, two = self.profile("one"), self.profile("two") + record = self.record(one, 101) + (two / "101.json").write_text(json.dumps(record), encoding="utf-8") + diagnostic = peer.diagnose_claude() + self.assertEqual(diagnostic["status"], "unknown") + self.assertEqual(diagnostic["availableInboxes"], 1) + + def test_unreadable_profile_preserves_other_profile(self): + good, bad = self.profile("good"), self.profile("bad") + self.record(good, 101) + original = os.scandir + def scan(path): + if Path(path) == bad: + raise PermissionError("fixture") + return original(path) + with mock.patch.object(peer.os, "scandir", side_effect=scan): + rows = peer.discover() + self.assertEqual([row["pid"] for row in rows], [101]) + self.assertIn("sessions_directory_unavailable", rows.errors) + + def test_inventory_limits_are_reported(self): + self.record(self.profile("one"), 101) + self.record(self.profile("two"), 102) + with mock.patch.object(peer, "MAX_CLAUDE_PROFILES", 1): + self.assertIn("profile_inventory_limit", peer.discover().errors) + with mock.patch.object(peer, "MAX_CLAUDE_RECORDS", 1): + rows = peer.discover() + self.assertLessEqual(len(rows), 1) + self.assertIn("session_inventory_limit", rows.errors) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_profile_symlink_and_home_symlink_are_not_followed(self): + target = self.root / "outside/sessions" + target.mkdir(parents=True) + self.record(target, 101) + (self.base / "alias").symlink_to(target.parent, target_is_directory=True) + profile = self.base / "home-alias" + profile.mkdir() + (profile / "home").symlink_to(target.parent, target_is_directory=True) + rows = peer.discover() + self.assertEqual(rows, []) + self.assertIn("profile_directory_unavailable", rows.errors) + + def test_reparse_and_foreign_uid_directories_are_rejected(self): + node = self.base.stat() + fields = dict(st_mode=node.st_mode, st_uid=getattr(node, "st_uid", 0), + st_dev=node.st_dev, st_ino=node.st_ino, st_file_attributes=0x400) + with mock.patch.object(Path, "lstat", return_value=argparse.Namespace(**fields)): + with self.assertRaises(ValueError): + peer._claude_safe_directory(self.base) + if hasattr(os, "getuid"): + fields.update(st_file_attributes=0, st_uid=os.getuid() + 1) + with mock.patch.object(Path, "lstat", return_value=argparse.Namespace(**fields)): + with self.assertRaises(ValueError): + peer._claude_safe_directory(self.base) + + @unittest.skipUnless(hasattr(os, "getuid"), "POSIX directory ownership") + def test_explicit_directory_preserves_existing_readable_owner_scope(self): + self.record(self.default, 100) + actual_uid = os.getuid() + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}), \ + mock.patch.object(peer.os, "getuid", return_value=actual_uid + 1): + selected = peer.discover()[0] + peer.revalidate_claude_selection(selected) + self.assertEqual(selected["pid"], 100) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_default_symlink_loop_preserves_healthy_profile_as_partial(self): + self.default.rmdir() + self.default.symlink_to(self.default, target_is_directory=True) + self.record(self.profile("one"), 101) + rows = peer.discover() + self.assertEqual([row["pid"] for row in rows], [101]) + self.assertIn("sessions_directory_unavailable", rows.errors) + + def test_profile_disappearance_after_inventory_is_incomplete(self): + directory = self.profile("one") + self.record(directory, 101) + inventory = peer.claude_registry_directories() + directory.rename(directory.parent / "old-sessions") + with mock.patch.object(peer, "claude_registry_directories", return_value=inventory): + rows = peer.discover() + self.assertIn("sessions_directory_unavailable", rows.errors) + + def test_stale_pid_or_inbox_is_not_a_reachable_profile(self): + directory = self.profile("one") + self.record(directory, 101) + with mock.patch.object(peer, "claude_process_state", return_value=(False, "process_not_live")): + self.assertEqual(peer.discover(), []) + self.assertFalse(peer.discover(True)[0]["reachable"]) + value = self.record(directory, 102) + value["messagingSocketPath"] = "" + (directory / "102.json").write_text(json.dumps(value), encoding="utf-8") + self.assertEqual([row["pid"] for row in peer.discover()], [101]) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_canonical_alias_dedup_and_name_ambiguity(self): + one, two = self.profile("one"), self.profile("two") + self.record(one, 101) + self.default.rmdir() + self.default.symlink_to(one, target_is_directory=True) + self.assertEqual(len(peer.discover()), 1) + self.record(two, 102) + rows = peer.discover() + with self.assertRaises(peer.CcPeerError): + peer.resolve_target(rows, "worker") + self.assertEqual(peer.resolve_target(rows, "102")["pid"], 102) + + def test_identical_records_in_distinct_profile_roots_are_conflicting(self): + one, two = self.profile("one"), self.profile("two") + record = self.record(one, 101) + (two / "101.json").write_text(json.dumps(record), encoding="utf-8") + self.assertEqual(peer.discover(), []) + self.assertEqual(len(peer.discover(True)), 2) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_explicit_default_alias_cannot_retarget_after_selection(self): + one, two = self.profile("one"), self.profile("two") + record = self.record(one, 101) + (two / "101.json").write_text(json.dumps(record), encoding="utf-8") + self.default.rmdir() + self.default.symlink_to(one, target_is_directory=True) + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}): + selected = peer.discover()[0] + peer.revalidate_claude_selection(selected) + self.default.unlink() + self.default.symlink_to(two, target_is_directory=True) + with self.assertRaises(peer.CcPeerError): + peer.revalidate_claude_selection(selected) + + def test_conflicting_pid_or_inbox_never_selects_by_inventory_order(self): + one, two = self.profile("one"), self.profile("two") + original = self.record(one, 101) + for pid, inbox in ((101, None), (102, original["messagingSocketPath"])): + with self.subTest(pid=pid): + for path in two.glob("*.json"): + path.unlink() + self.record(two, pid, "conflict", inbox) + self.assertEqual(peer.discover(), []) + rows = peer.discover(True) + self.assertEqual(len(rows), 2) + self.assertTrue(all(not row["reachable"] for row in rows)) + with self.assertRaises(peer.CcPeerError): + peer.resolve_target(rows, "101") + + def test_proven_dead_record_does_not_claim_successors_reused_socket(self): + one, two = self.profile("one"), self.profile("two") + record = self.record(one, 101) + self.record(two, 102, inbox=record["messagingSocketPath"]) + with mock.patch.object(peer, "claude_process_state", side_effect=lambda row, pid: + (pid == 102, None if pid == 102 else "process_not_live")): + rows = peer.discover() + self.assertEqual([row["pid"] for row in rows], [102]) + self.assertEqual(rows.errors, []) + + def test_generation_uses_selected_profile_and_survives_adapter_listing(self): + directory = self.profile("one") + self.record(directory, 101) + result = peer.ClaudeAdapter().list(peer.ExecutionContext("local", argparse.Namespace(all=False))) + selected = result["sessions"][0] + generation = peer.claude_generation(selected) + self.assertRegex(generation, r"^tg1:[0-9a-f]{64}$") + self.assertEqual(selected.registry_directory, directory) + self.assertFalse((self.default / "101.json").exists()) + with mock.patch.object(peer, "discover", return_value=[selected]): + binding, pinned, handoff_session = peer.handoff_binding( + argparse.Namespace(to="101", target_generation=generation), "fixture") + self.assertIs(handoff_session, selected) + self.assertEqual(pinned, generation) + self.assertEqual(binding["target"], "101") + args = argparse.Namespace(all=False, with_target_generation=True) + listing = peer.LocalTransport().execute("list", peer.ClaudeAdapter(), args) + self.assertIsInstance(listing["sessions"][0], peer.ClaudeSession) + self.assertEqual(listing["sessions"][0]["targetGeneration"], generation) + self.assertNotIn("registry", json.dumps(listing)) + + def test_replaced_profile_or_record_refuses_before_dry_run_or_send(self): + directory = self.profile("one") + self.record(directory, 101) + selected = peer.discover()[0] + replacement = directory.parent / "replaced" + directory.rename(replacement) + directory.mkdir() + self.record(directory, 101) + with mock.patch.object(peer, "discover", return_value=[selected]), \ + mock.patch.object(peer, "post_to_socket") as post: + for dry in (True, False): + with self.subTest(dry=dry), self.assertRaises(peer.CcPeerError): + peer.ClaudeAdapter().submit(peer.ExecutionContext("local", + argparse.Namespace(to="101", dry_run=dry)), "fixture") + post.assert_not_called() + self.assertIsNone(peer.claude_generation(selected)) + + def test_endpoint_replacement_and_symlink_retarget_are_refused(self): + directory = self.profile("one") + record = self.record(directory, 101) + selected = peer.discover()[0] + record["messagingSocketPath"] = "replaced" + (directory / "101.json").write_text(json.dumps(record), encoding="utf-8") + with self.assertRaises(peer.CcPeerError): + peer.revalidate_claude_selection(selected) + + @unittest.skipIf(os.name == "nt", "POSIX socket boundary") + def test_profile_replacement_during_connect_prevents_any_write(self): + directory = self.profile("one") + self.record(directory, 101) + selected = peer.discover()[0] + def replace(_): + directory.rename(directory.parent / "old-sessions") + directory.mkdir() + connection = mock.Mock() + connection.connect.side_effect = replace + with mock.patch.object(peer.socket, "socket", return_value=connection): + with self.assertRaises(peer.CcPeerError): + peer.post_to_socket(selected["socket"], "fixture", pid=101, + generation_session=selected) + connection.sendall.assert_not_called() + connection.close.assert_called_once() + + def test_windows_auth_reader_uses_selected_root_only_when_sending(self): + directory = self.profile("one") + self.record(directory, 101) + selected = peer.discover()[0] + with mock.patch.object(peer, "IS_WINDOWS", True), \ + mock.patch.object(peer, "_read_win_auth", return_value=None) as auth: + with self.assertRaises(peer.CcPeerError): + peer._post_to_pipe(r"\\.\pipe\fixture", 101, "fixture", generation_session=selected) + auth.assert_called_once_with(101, directory) + + def test_selected_decimal_filename_aliases_preserve_ordinary_dry_run(self): + for stem in ("00101", "١٠١"): + with self.subTest(stem=stem): + for old_record in self.default.glob("*.json"): + old_record.unlink() + self.record(self.default, 101) + canonical_generation = peer.claude_generation(peer.discover()[0]) + (self.default / "101.json").rename(self.default / (stem + ".json")) + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}): + selected = peer.discover()[0] + peer.revalidate_claude_selection(selected) + result = peer.ClaudeAdapter().submit(peer.ExecutionContext("local", + argparse.Namespace(to="101", dry_run=True)), "fixture") + self.assertTrue(result["dryRun"]) + self.assertEqual(peer.claude_generation(selected), canonical_generation) + self.assertNotIn("registry", json.dumps(selected)) + value = json.loads((self.default / (stem + ".json")).read_text(encoding="utf-8")) + value["startedAt"] += 1 + (self.default / (stem + ".json")).write_text(json.dumps(value), encoding="utf-8") + with self.assertRaises(peer.CcPeerError): + peer.revalidate_claude_selection(selected) + (self.default / (stem + ".json")).unlink() + + @unittest.skipIf(os.name == "nt", "POSIX socket aliases") + def test_ordinary_socket_alias_remains_usable_and_retarget_is_refused(self): + record = self.record(self.default, 101) + alias = self.root / "alias.sock" + alias.symlink_to(record["messagingSocketPath"]) + record["messagingSocketPath"] = str(alias) + (self.default / "101.json").write_text(json.dumps(record), encoding="utf-8") + selected = peer.discover()[0] + peer.revalidate_claude_selection(selected) + self.assertIsNone(peer.claude_generation(selected), "Existing alias-generation limit remains") + connection = mock.Mock() + with mock.patch.object(peer.socket, "socket", return_value=connection): + peer.post_to_socket(selected["socket"], "fixture", pid=101, generation_session=selected) + connection.connect.assert_called_once_with(str(alias)) + connection.sendall.assert_called_once() + alias.unlink() + other = self.record(self.default, 102) + alias.symlink_to(other["messagingSocketPath"]) + with self.assertRaises(peer.CcPeerError) as caught: + peer.revalidate_claude_selection(selected) + self.assertEqual(caught.exception.details.get("status"), "refused") + + def test_doctor_directory_and_record_permission_classifications_are_preserved(self): + self.record(self.default, 101) + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}), \ + mock.patch.object(peer.os, "scandir", side_effect=PermissionError("fixture")): + result = peer.diagnose_claude() + self.assertEqual(result["status"], "permission_denied") + self.assertEqual(result["checks"][0]["code"], "sessions_dir_permission_denied") + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}), \ + mock.patch.object(peer, "read_claude_record", side_effect=PermissionError("fixture")): + result = peer.diagnose_claude() + self.assertEqual(result["status"], "permission_denied") + self.assertEqual(result["invalidRecords"], 0) + self.assertEqual(result["checks"][0]["count"], 1) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_canonical_registry_is_read_once_despite_status_changes(self): + directory = self.profile("one") + self.record(directory, 101) + self.default.rmdir() + self.default.symlink_to(directory, target_is_directory=True) + original = peer.read_claude_record + reads = [] + def read(path): + record = original(path) + reads.append(path) + record["status"] = "idle" if len(reads) == 1 else "busy" + return record + with mock.patch.object(peer, "read_claude_record", side_effect=read): + rows = peer.discover() + self.assertEqual(len(reads), 1) + self.assertEqual(len(rows), 1) + self.assertEqual(rows.errors, []) + + @unittest.skipIf(os.name == "nt", "POSIX socket boundary") + def test_after_connect_selection_refusal_is_typed_no_effect(self): + directory = self.profile("one") + self.record(directory, 101) + selected = peer.discover()[0] + connection = mock.Mock() + def replace(_): + directory.rename(directory.parent / "old-sessions") + directory.mkdir() + connection.connect.side_effect = replace + with mock.patch.object(peer.socket, "socket", return_value=connection): + with self.assertRaises(peer.CcPeerError) as caught: + peer.post_to_socket(selected["socket"], "fixture", pid=101, generation_session=selected) + self.assertEqual(caught.exception.details.get("status"), "refused") + self.assertFalse(caught.exception.details.get("submitted", True)) + self.assertFalse(caught.exception.details.get("retryAllowed", True)) + self.assertNotIn("lastSeenTarget", caught.exception.details) + connection.sendall.assert_not_called() + + @unittest.skipIf(os.name == "nt", "POSIX handoff boundary") + def test_local_handoff_records_before_and_after_connect_changes_as_refused(self): + for boundary in ("before", "after"): + with self.subTest(boundary=boundary): + directory = self.profile(boundary) + self.record(directory, 101) + ledger_root = self.root / ("ledger-" + boundary) + peer.HandoffLedger(ledger_root).initialize() + output = io.StringIO() + connection = mock.Mock() + def replace(): + directory.rename(directory.parent / "old-sessions") + directory.mkdir() + real_post = peer.post_to_socket + def post(*args, **kwargs): + if boundary == "before": + replace() + return real_post(*args, **kwargs) + if boundary == "after": + connection.connect.side_effect = lambda _: replace() + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(directory.parent)}), \ + mock.patch.object(peer, "handoff_root", return_value=ledger_root), \ + mock.patch.object(peer, "handoff_now_ns", side_effect=peer.time.monotonic_ns), \ + mock.patch.object(peer, "handoff_boot_clock", return_value="a" * 64), \ + mock.patch.object(peer, "post_to_socket", side_effect=post), \ + mock.patch.object(peer.socket, "socket", return_value=connection), \ + contextlib.redirect_stdout(output): + code = peer.main(["send", "--to=101", "--message=fixture", "--observe-delivery", + "--no-from", "--no-reply-to", "--no-update-notice", "--json"]) + result = json.loads(output.getvalue()) + self.assertEqual(code, 1) + self.assertEqual(result["handoff"]["submission"]["status"], "refused") + self.assertEqual(result["reason"], "native_submission_refused") + self.assertFalse(result["retryAllowed"]) + connection.sendall.assert_not_called() + self.assertEqual(connection.connect.call_count, int(boundary == "after")) + + @unittest.skipIf(os.name == "nt", "POSIX private source streaming") + def test_remote_handoff_after_connect_change_is_refused_without_writes(self): + directory = self.profile("remote") + self.record(directory, 101) + context = {"schemaVersion": 1, "phase": "probe", "requestId": str(peer.uuid.uuid4()), + "agent": "claude", "target": "101", "home": None, "nativeContext": None, + "generation": None, "anchor": None, "remainingCutoffMs": None, "remainingTotalMs": None} + def invoke(request): + argv = ["send", "--_handoff-native-context=" + base64.b64encode(peer.HandoffLedger.encode(request)).decode(), + "--to=101", "--b64=" + base64.b64encode(b"fixture").decode(), + "--no-from", "--no-reply-to", "--no-update-notice", "--json"] + if request["phase"] == "probe": + argv += ["--_handoff-native-cutoff-ms=2000", "--_handoff-native-total-ms=7000"] + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = peer.main(argv) + return code, json.loads(output.getvalue()) + connection = mock.Mock() + def replace(_): + directory.rename(directory.parent / "old-sessions") + directory.mkdir() + connection.connect.side_effect = replace + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(directory.parent)}), \ + mock.patch.object(peer, "handoff_now_ns", side_effect=peer.time.monotonic_ns), \ + mock.patch.object(peer, "handoff_boot_clock", return_value="a" * 64): + code, response = invoke(context) + self.assertEqual(code, 0) + peer.handoff_validate_remote_response(response, context) + proof = response["handoffNative"] + context.update(phase="effect", generation=proof["generation"], nativeContext=proof["nativeContext"], + anchor=proof["anchor"], remainingCutoffMs=2000, remainingTotalMs=7000) + with mock.patch.object(peer.socket, "socket", return_value=connection): + code, response = invoke(context) + native = peer.handoff_validate_remote_response(response, context) + self.assertEqual(code, 1) + self.assertEqual(native["reason"], "native_submission_refused") + self.assertFalse(native["retryAllowed"]) + connection.connect.assert_called_once() + connection.sendall.assert_not_called() + + def test_selected_refusal_is_a_valid_pinned_ssh_response(self): + refusal = peer.generation_refused("target_generation_unavailable", "fixture") + response = {"schemaVersion": 1, "command": "send", "ok": False, "error": str(refusal), **refusal.details} + token = "tg1:" + "a" * 64 + result = peer.parse_ssh_response(json.dumps(response), + ["send", "--to=101", "--target-generation=" + token]) + self.assertTrue(result[3]) + self.assertNotIn("lastSeenTarget", response) + + @unittest.skipIf(os.name == "nt", "POSIX socket snapshot") + def test_endpoint_disappearance_does_not_invalidate_other_inventory_rows(self): + value = self.record(self.default, 101) + original = Path.is_socket + def disappear(path): + if str(path) == value["messagingSocketPath"]: + path.unlink() + return True # It existed at the preceding presence checkpoint. + return original(path) + with mock.patch.object(Path, "is_socket", disappear): + rows = peer.discover(True) + self.assertEqual(len(rows), 1) + self.assertFalse(rows[0]["reachable"]) + self.assertEqual(rows.errors, []) + + def test_distinct_same_registry_records_are_conflicting_even_if_identical(self): + value = self.record(self.default, 101) + (self.default / "00101.json").write_text(json.dumps(value), encoding="utf-8") + rows = peer.discover(True) + self.assertEqual(len(rows), 2) + self.assertIn("conflicting_session_records", rows.errors) + self.assertTrue(all(not row["reachable"] for row in rows)) + + def test_unrelated_healthy_target_is_not_selected_amid_live_conflicts(self): + self.record(self.default, 100, "healthy") + value = self.record(self.profile("one"), 101) + two = self.profile("two") + (two / "101.json").write_text(json.dumps(value), encoding="utf-8") + rows = peer.discover(True) + self.assertTrue(next(row for row in rows if row["pid"] == 100)["reachable"]) + with self.assertRaises(peer.CcPeerError): + peer.resolve_target(rows, "100") + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.default.parent)}): + self.assertEqual(peer.resolve_target(peer.discover(), "100")["pid"], 100) + + @unittest.skipIf(os.name == "nt", "POSIX endpoint inode identity") + def test_live_socket_alias_claims_conflict_in_both_inventory_orders(self): + one, two = self.profile("one"), self.profile("two") + original = self.record(one, 101, "physical") + alias = self.root / "same-inbox-alias.sock" + alias.symlink_to(original["messagingSocketPath"]) + self.record(two, 102, "alias", str(alias)) + for directories in ([one, two], [two, one]): + with self.subTest(reversed=directories[0] == two), \ + mock.patch.object(peer, "claude_registry_directories", return_value=(directories, [])): + rows = peer.discover(True) + self.assertEqual(len(rows), 2) + self.assertIn("conflicting_session_records", rows.errors) + self.assertTrue(all(not row["reachable"] for row in rows)) + for target in ("101", "102", "physical", "alias"): + with self.assertRaises(peer.CcPeerError): + peer.resolve_target(rows, target) diff --git a/tests/core/test_cli_runtime.py b/tests/core/test_cli_runtime.py index 2ba38bb..8da7738 100644 --- a/tests/core/test_cli_runtime.py +++ b/tests/core/test_cli_runtime.py @@ -135,7 +135,9 @@ def test_main_catches_unexpected_exception_human(self): self.assertEqual(code, session_peer.EXIT_ERROR) def test_discover_reports_permission_error(self): - with mock.patch.object(Path, "iterdir", side_effect=PermissionError("denied")): + with tempfile.TemporaryDirectory() as directory, \ + mock.patch.object(session_peer, "sessions_dir", return_value=Path(directory)), \ + mock.patch.object(session_peer.os, "scandir", side_effect=PermissionError("denied")): with self.assertRaisesRegex(session_peer.CcPeerError, "Cannot read Claude sessions"): session_peer.discover() diff --git a/tests/core/test_unified_list.py b/tests/core/test_unified_list.py index 88c149f..583249a 100644 --- a/tests/core/test_unified_list.py +++ b/tests/core/test_unified_list.py @@ -79,7 +79,9 @@ def test_missing_home_and_missing_claude_directory(self): self.assertEqual(result["discovery"]["codex"]["status"], "error") def test_claude_directory_permission_is_not_empty_success(self): - with mock.patch.object(Path, "iterdir", side_effect=PermissionError("denied")): + with tempfile.TemporaryDirectory() as directory, \ + mock.patch.object(peer, "sessions_dir", return_value=Path(directory)), \ + mock.patch.object(peer.os, "scandir", side_effect=PermissionError("denied")): with self.assertRaises(peer.CcPeerError): peer.discover()