diff --git a/docs/cli-reference.md b/docs/cli-reference.md index d0a20d29..4b68ee91 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -80,6 +80,47 @@ installation path. See the [v0.9 release notes](releases/v0.9.0.md). ### Message input and result output +#### Optional inbox generation precondition (development) + +`list --with-target-generation --json` adds `targetGeneration` and +`generationStatus` to each row. For Claude inboxes with native process-creation +evidence, copy the opaque `tg1:` token into `send --target-generation TOKEN` +(including `--dry-run` or SSH). Keep the token with the original destination; +it is a precondition, not a secret, permission, ACK, or retry/deduplication key. +Unpinned commands keep their existing behavior and output. + +Pinned Claude sends recheck the registry, process creation and inbox endpoint, +then verify the connected socket/pipe server PID before writing any message or +Windows authentication line. Reused names/PIDs or changed endpoints are refused +with `stale_target`; unprovable native identity is refused rather than guessed. +The token does not authenticate the model or defend against a malicious process +with the same OS account. OS process creation and connected-peer APIs are needed; +unsupported platforms cannot use this mode. Same-generation process state/name +updates do not by themselves invalidate the token. +Native creation values have OS-specific granularity (Linux uses boot ID and +start-time clock ticks); this is not a globally unique or clone-proof identity. +The opaque token is also not cryptographic SSH host-key continuity. + +A `stale_target` refusal includes `lastSeenTarget` with the original requested +Claude generation, not a fresh lookup of the successor. This bounded metadata +does not include process creation values, paths, user names or host names: + +`{"status":"refused","reason":"stale_target","submitted":false,"retryAllowed":false,"lastSeenTarget":{"agent":"claude","targetGeneration":"tg1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}` + +`target_generation_unavailable` means native identity could not be verified; +`unsupported_target_generation` means the target does not support this +precondition. Neither result includes `lastSeenTarget` or authorizes a retry. +The source-streamed restart regression uses an owned protocol inbox and a local +SSH stand-in; it is not evidence of a physical remote host or native agent test. + +Codex queue targets, Antigravity bridges and paired devices do **not** acquire +this inbox-incarnation guarantee: they report an unsupported generation, and a +generic pinned send to them is refused before queue/wake. In particular, a +Codex writer recheck cannot atomically constrain which later writer consumes a +persisted queue row. Their existing separate identity contracts remain intact. +The broader cross-agent work in #180 remains open; this development feature is +not shipped in v1.0.4 and adds no automatic fallback or resubmission. + `--message TEXT` (short form `-m`) names the text sent to the destination. `--output-format text|json` selects the command result format, not the message format. It is available on `list`, `send`, `doctor`, and `update`; the default is diff --git a/docs/i18n-manifest.json b/docs/i18n-manifest.json index be30337e..e9cd47b7 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": "56c0a808c9b8393c5de1ea476111153c7a84ef61376b1da3987d0ff4b71fcc66", + "docs/cli-reference.md": "3a366585e7ef093f79fc7e83c843ce9b291731fa4c0055348b92bcc2a2765cd7", "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 9582a45b..2cb5dda6 100644 --- a/docs/ja/cli-reference.md +++ b/docs/ja/cli-reference.md @@ -2,6 +2,39 @@ [概要に戻る](../../README.ja.md) +## 受信セッションの世代を固定する条件（開発中） + +`list --with-target-generation --json` は各行に `targetGeneration` と +`generationStatus` を追加します。OS のプロセス生成情報を確認できる Claude の受信先では、 +`tg1:` トークンを `send --target-generation TOKEN` に指定できます。`--dry-run` と SSH でも +元の宛先のトークンを使います。これは送信の前提条件であり、秘密情報、権限、ACK、再送や +重複排除のキーではありません。指定しない場合、既存の出力と動作は変わりません。 + +送信直前に登録情報、プロセス生成情報と受信先を再確認し、接続したソケット・パイプの +サーバー PID を確認してから、メッセージや Windows の認証行を書き込みます。 +名前・PID の再利用や受信先の変更は `stale_target` で拒否し、確認できない身元は推測しません。 +同一プロセスの状態・名前の更新だけでは世代は変わりません。モデルの身元を認証したり、 +同じ OS アカウントの悪意あるプロセスを防いだりする保証はなく、必要な OS API がなければ利用できません。 +生成情報の粒度は OS に依存します（Linux は起動 ID と開始時刻の clock tick を使用）。 +世界的に一意・複製不能な身元や、暗号学的な SSH ホスト鍵の連続性は保証しません。 + +`stale_target` の拒否には、元の要求で指定した Claude 世代の `lastSeenTarget` を含めます。 +後継プロセスを再検索した結果ではありません。この限定的なメタデータには、プロセス生成値、 +パス、ユーザー名やホスト名を含めません。 + +`{"status":"refused","reason":"stale_target","submitted":false,"retryAllowed":false,"lastSeenTarget":{"agent":"claude","targetGeneration":"tg1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}` + +`target_generation_unavailable` は OS の身元を検証できないこと、 +`unsupported_target_generation` は対象がこの前提条件に対応していないことを示します。 +どちらも `lastSeenTarget` を含めず、再送を許可しません。 +ソース転送・再起動の回帰テストは専用のプロトコル受信先とローカルの SSH 代替を使います。 +実際の遠隔ホストやネイティブエージェントの検証を示すものではありません。 + +Codex キュー、Antigravity とペアリング済み機器には、この受信プロセス世代の保証はありません。 +一般の世代固定要求はキュー投入・wake 前に拒否し、それぞれの既存契約を維持します。 +Codex の書き込みプロセスを再確認しても、後からキューを消費するプロセスを原子的に制限できません。 +#180 全体は未完了です。v1.0.4 の公開機能ではなく、自動フォールバックや再送も追加しません。 + コマンドの詳細な動作、インストールオプション、トランスポートの制限、および過去の検証結果について以下に説明します。 機械consumerは[v1互換性契約](compatibility-v1.md)にも従ってください。 diff --git a/docs/ko/cli-reference.md b/docs/ko/cli-reference.md index 202ece8c..37699af2 100644 --- a/docs/ko/cli-reference.md +++ b/docs/ko/cli-reference.md @@ -2,6 +2,39 @@ [개요로 돌아가기](../../README.ko.md) +## 수신 세션 세대 고정 조건 (개발 중) + +`list --with-target-generation --json`은 각 행에 `targetGeneration`과 +`generationStatus`를 추가합니다. 운영체제의 프로세스 생성 정보를 확인할 수 있는 +Claude 세션은 `tg1:` 토큰을 `send --target-generation TOKEN`에 전달할 수 있습니다. +`--dry-run`과 SSH에서도 원래 대상의 토큰을 사용합니다. 토큰은 전송 조건이며 비밀정보, +권한, ACK 또는 재전송·중복 방지 키가 아닙니다. 사용하지 않으면 기존 출력과 동작을 유지합니다. + +전송 직전에 세션 기록, 프로세스 생성 정보, 수신 경로를 재확인하고 실제 연결된 소켓·파이프의 +서버 PID를 확인한 뒤 메시지와 Windows 인증 줄을 씁니다. 이름·PID 재사용이나 경로 변경은 +`stale_target`으로 거부하며, 확인 불가능한 신원은 추정하지 않습니다. 같은 프로세스의 상태나 +이름 변경만으로 토큰이 바뀌지는 않습니다. 모델의 신원을 인증하거나 같은 OS 계정의 악성 +프로세스를 막는 기능은 아니며, 필요한 운영체제 API가 없으면 사용할 수 없습니다. +생성 정보의 정밀도는 운영체제마다 다릅니다(Linux는 부팅 ID와 시작 시각의 clock tick 사용). +전역적으로 고유하거나 복제를 방지하는 신원, 암호학적 SSH 호스트 키 연속성을 보장하지 않습니다. + +`stale_target` 거부에는 원래 요청한 Claude 세대를 담은 `lastSeenTarget`이 포함됩니다. +새 수신 프로세스를 다시 조회한 결과가 아닙니다. 이 제한된 메타데이터에는 프로세스 생성값, +경로, 사용자 이름이나 호스트 이름을 넣지 않습니다. + +`{"status":"refused","reason":"stale_target","submitted":false,"retryAllowed":false,"lastSeenTarget":{"agent":"claude","targetGeneration":"tg1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}` + +`target_generation_unavailable`은 운영체제의 신원을 확인할 수 없다는 뜻이고, +`unsupported_target_generation`은 대상이 이 전송 조건을 지원하지 않는다는 뜻입니다. +두 결과에는 `lastSeenTarget`을 넣지 않으며 재전송 권한도 주지 않습니다. +소스 전송·재시작 회귀 테스트는 전용 프로토콜 수신함과 로컬 SSH 대역을 사용합니다. +실제 원격 장비나 실제 에이전트의 검증 근거는 아닙니다. + +Codex 큐, Antigravity와 페어링 기기는 이 수신 프로세스 세대 보장을 제공하지 않습니다. +일반 세대 고정 요청은 큐 삽입·wake 전에 거부하고 각각의 기존 계약을 유지합니다. +Codex의 작성자 재확인만으로 나중에 큐를 소비하는 작성자를 원자적으로 제한할 수는 없습니다. +#180의 전체 작업은 아직 열려 있습니다. v1.0.4에 발행된 기능이 아니며 자동 대체 전송·재전송도 없습니다. + 명령어의 상세 동작, 설치 옵션, 전송 한계 및 과거 검증 내역은 아래와 같습니다. 기계 소비자는 [v1 호환성 계약](compatibility-v1.md)도 따라야 합니다. diff --git a/docs/zh-CN/cli-reference.md b/docs/zh-CN/cli-reference.md index 6423a88d..55c51268 100644 --- a/docs/zh-CN/cli-reference.md +++ b/docs/zh-CN/cli-reference.md @@ -2,6 +2,35 @@ [返回概览](../../README.zh-CN.md) +## 固定接收会话代次的前置条件（开发中） + +`list --with-target-generation --json` 为每行添加 `targetGeneration` 和 +`generationStatus`。对于能核实操作系统进程创建信息的 Claude 接收端，可将 `tg1:` +标记传给 `send --target-generation TOKEN`；`--dry-run` 和 SSH 也使用原目标的标记。 +它是发送前置条件，不是秘密、权限、ACK 或重发、去重凭据。不使用时保留现有输出和行为。 + +发送前重新核实注册记录、进程创建信息和接收端，再核实已连接的套接字或管道服务器 PID， +之后才写入消息和 Windows 认证行。名称、PID 被复用或接收端变化时以 `stale_target` +拒绝；无法证明身份时不作猜测。同一进程的状态或名称更新本身不会改变代次。 +此功能不认证模型身份，也不防范同一 OS 账户的恶意进程；缺少所需 OS API 时不可使用。 +创建信息的精度取决于 OS（Linux 使用启动 ID 和开始时间的 clock tick），并不保证全局唯一、 +防克隆的身份或密码学上的 SSH 主机密钥连续性。 + +`stale_target` 拒绝结果包含 `lastSeenTarget`，记录原请求指定的 Claude 代次， +不是重新查找后继进程的结果。这一有限元数据不包含进程创建值、路径、用户名或主机名。 + +`{"status":"refused","reason":"stale_target","submitted":false,"retryAllowed":false,"lastSeenTarget":{"agent":"claude","targetGeneration":"tg1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}` + +`target_generation_unavailable` 表示无法核实操作系统身份； +`unsupported_target_generation` 表示目标不支持该前置条件。 +这两种结果均不包含 `lastSeenTarget`，也不授权重发。 +源码传输与重启回归测试使用专用协议收件箱和本地 SSH 替身， +不代表实际远程主机或原生代理的验证。 + +Codex 队列、Antigravity 和配对设备不具备此接收进程代次保证。通用代次固定请求会在入队、 +wake 前拒绝，并保留各自既有契约。重新核实 Codex 写入进程，不能原子地限制之后消费 +持久队列的进程。#180 的完整工作仍未完成；该功能尚未在 v1.0.4 发布，不会自动回退或重发。 + 以下是详细的命令行为、安装选项、传输限制和历史验证记录。 机器consumer还应遵循[v1兼容性契约](compatibility-v1.md)。 diff --git a/session_peer.py b/session_peer.py index 8eccc514..3e759d88 100644 --- a/session_peer.py +++ b/session_peer.py @@ -1650,7 +1650,8 @@ def _read_win_auth(pid: int) -> str | None: return None -def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: +def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=None, + expected_generation=None) -> None: """Write one message to a Windows named pipe inbox.""" check_message(text, remote=False) @@ -1666,6 +1667,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: import time try: with open(pipe_path, "wb") as pipe: + 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")) pipe.write((payload + "\n").encode("utf-8")) pipe.flush() @@ -1674,7 +1677,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: raise CcPeerError(f"cannot reach inbox at {pipe_path}: {exc}") from exc -def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: +def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_session=None, + expected_generation=None, effect_deadline=None, total_deadline=None) -> None: """Write one message to a session's inbox socket. On macOS and Linux the {"type":"auth",...} line the docs describe is @@ -1684,8 +1688,22 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: a complete line within 30 seconds, so the message is built before the socket is opened. """ + bounded = effect_deadline is not None or total_deadline is not None + if bounded: + import math + if (type(effect_deadline) not in (int, float) or type(total_deadline) not in (int, float) + or not math.isfinite(effect_deadline) or not math.isfinite(total_deadline) + or effect_deadline > total_deadline): + raise generation_refused("invalid_effect_deadline", "Invalid private effect budget; nothing sent") + if IS_WINDOWS: + # Python's synchronous named-pipe open/write has no proven bounded + # cancellation contract. Never advertise it as bounded handoff. + raise generation_refused("unsupported_bounded_inbox", "Bounded Windows pipe submission is unsupported") + if time.monotonic() >= effect_deadline: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") if IS_WINDOWS and socket_path.startswith("\\\\.\\pipe\\"): - _post_to_pipe(socket_path, pid, text) + _post_to_pipe(socket_path, pid, text, generation_session=generation_session, + expected_generation=expected_generation) return check_message(text, remote=False) @@ -1696,27 +1714,176 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: ) conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - conn.settimeout(CONNECT_TIMEOUT) try: + remaining = effect_deadline - time.monotonic() if bounded else CONNECT_TIMEOUT + if remaining <= 0: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") + conn.settimeout(min(CONNECT_TIMEOUT, remaining)) try: conn.connect(socket_path) except OSError as exc: + if bounded: + raise generation_refused("inbox_connection_failed", "Inbox connection failed before any write") from exc raise CcPeerError(f"cannot reach inbox at {socket_path}: {exc}") from exc try: + if expected_generation is not None: + verify_connected_generation(generation_session, expected_generation, connected_inbox_pid(conn)) + if bounded: + remaining = effect_deadline - time.monotonic() + if remaining <= 0: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") + conn.settimeout(remaining) conn.sendall((payload + "\n").encode("utf-8")) conn.shutdown(socket.SHUT_WR) - conn.settimeout(DRAIN_TIMEOUT) + remaining = effect_deadline - time.monotonic() if bounded else DRAIN_TIMEOUT + if remaining <= 0: + return # Full write retained; no further wait beyond cutoff. + conn.settimeout(min(DRAIN_TIMEOUT, remaining)) try: conn.recv(1) except OSError: pass except OSError as exc: + if bounded: + raise CcPeerError("Inbox write outcome unknown; do not automatically retry", + {"status": "unknown", "reason": "outcome_unknown", + "retryAllowed": False}) from exc raise CcPeerError(f"failed writing to {socket_path}: {exc}") from exc finally: conn.close() +# Optional discovery-to-inbox preconditions. These are not ACK or dedup keys. + +def generation_refused(reason: str, message: str, *, expected_generation=None) -> CcPeerError: + details = {"status": "refused", "reason": reason, + "submitted": False, "retryAllowed": False} + # Original discovery precondition only: never reflect a successor's name, + # PID, host, path or native creation evidence as "last seen" metadata. + if (reason == "stale_target" and isinstance(expected_generation, str) + and re.fullmatch(r"tg1:[0-9a-f]{64}", expected_generation)): + details["lastSeenTarget"] = {"agent": "claude", "targetGeneration": expected_generation} + return CcPeerError(message, details) + + +def process_generation(pid: int) -> str | None: + """Use native creation evidence, never the second-resolution ps display.""" + if type(pid) is not int or pid <= 1: + return None + try: + if sys.platform == "linux": + # comm may contain spaces and ')'; fields after the final ')' start + # at field 3, so starttime (22) is offset 19. + fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split() + boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + return f"linux:{boot}:{fields[19]}" if fields[19].isdigit() else None + if sys.platform == "darwin": + import ctypes + class BsdInfo(ctypes.Structure): + _fields_ = [("prefix", ctypes.c_uint32 * 12), + ("comm", ctypes.c_char * 16), ("name", ctypes.c_char * 32), + ("suffix", ctypes.c_uint32 * 6), + ("seconds", ctypes.c_uint64), ("micros", ctypes.c_uint64)] + lib = ctypes.CDLL("/usr/lib/libproc.dylib", use_errno=True) + lib.proc_pidinfo.argtypes = (ctypes.c_int, ctypes.c_int, ctypes.c_uint64, + ctypes.c_void_p, ctypes.c_int) + lib.proc_pidinfo.restype = ctypes.c_int + value = BsdInfo() + if lib.proc_pidinfo(pid, 3, 0, ctypes.byref(value), ctypes.sizeof(value)) != ctypes.sizeof(value): + return None + if value.prefix[3] != pid or not value.seconds: + return None + return f"darwin:{value.seconds}:{value.micros}" + if sys.platform == "win32": + rows = _windows_process_inspect("identity", str(pid)) + return "win32:" + rows[0]["startTime"] if rows and len(rows) == 1 else None + except (OSError, ValueError, IndexError, AttributeError): + pass + return None + + +def claude_generation(session: dict) -> str | None: + """Hash bounded registry/process/endpoint facts; expose no raw native key.""" + import hashlib + try: + pid = session["pid"] + birth = process_generation(pid) + if not birth: + return None + directory = sessions_dir().resolve() + record = read_claude_record(directory / f"{pid}.json") + endpoint = record.get("messagingSocketPath") + if record["pid"] != pid or endpoint != session["socket"]: + return None + identity = ["claude", sys.platform, socket.gethostname(), str(directory), pid, birth, endpoint] + if not IS_WINDOWS: + node = Path(endpoint).lstat() + if not stat.S_ISSOCK(node.st_mode): + return None + identity += [node.st_dev, node.st_ino] + else: + if not endpoint.startswith("\\\\.\\pipe\\"): + return 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): + return None + + +def validate_target_generation(value: str | None) -> None: + if value is not None and (not isinstance(value, str) or not re.fullmatch(r"tg1:[0-9a-f]{64}", value)): + raise generation_refused("invalid_target_generation", "Invalid target generation; run discovery again") + + +def require_claude_generation(session: dict, expected: str) -> None: + validate_target_generation(expected) + current = claude_generation(session) + if current is None: + raise generation_refused("target_generation_unavailable", "Cannot prove the selected inbox generation; nothing sent") + if current != expected: + raise generation_refused("stale_target", "The selected inbox generation changed; nothing sent", + expected_generation=expected) + + +def connected_inbox_pid(conn) -> int | None: + """Read credentials for the connected endpoint, not its current pathname.""" + import struct + try: + if sys.platform == "linux" and hasattr(socket, "SO_PEERCRED"): + return struct.unpack("3i", conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[0] + if sys.platform == "darwin": + # SOL_LOCAL / LOCAL_PEERPID from sys/un.h. + return struct.unpack("i", conn.getsockopt(0, 2, 4))[0] + except (OSError, ValueError, struct.error): + pass + return None + + +def connected_pipe_pid(pipe) -> int | None: + try: + import ctypes + from ctypes import wintypes + import msvcrt + function = ctypes.windll.kernel32.GetNamedPipeServerProcessId + function.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.ULONG)) + function.restype = wintypes.BOOL + pid = wintypes.ULONG() + return pid.value if function(msvcrt.get_osfhandle(pipe.fileno()), ctypes.byref(pid)) else None + except (OSError, AttributeError, ValueError): + return None + + +def verify_connected_generation(session: dict, expected: str, server_pid: int | None) -> None: + if server_pid is None: + raise generation_refused("target_generation_unavailable", "Connected inbox identity unavailable; nothing sent") + if server_pid != session["pid"]: + raise generation_refused("stale_target", "Connected inbox belongs to another process; nothing sent", + expected_generation=expected) + require_claude_generation(session, expected) + + # -------------------------------------------------------------------------- # Reply address. A cross-machine message carries no reply address of its own, # so the receiving Claude has no way to know answering is even possible. This @@ -2395,6 +2562,42 @@ def constant_value(value): and type(result.get("ok")) is bool and bool(argv) and result.get("command") == argv[0] ) + if valid and "lastSeenTarget" in result: + observed = result["lastSeenTarget"] + requested = [] + targets = [] + for index, argument in enumerate(argv): + if not isinstance(argument, str): + continue + if argument == "--target-generation" and index + 1 < len(argv): + requested.append(argv[index + 1]) + elif argument.startswith("--target-generation="): + requested.append(argument.split("=", 1)[1]) + if argument == "--to" and index + 1 < len(argv): + targets.append(argv[index + 1]) + elif argument.startswith("--to="): + targets.append(argument.split("=", 1)[1]) + last_seen_valid = ( + isinstance(observed, dict) and set(observed) == {"agent", "targetGeneration"} + and observed["agent"] == "claude" and isinstance(observed["targetGeneration"], str) + and re.fullmatch(r"tg1:[0-9a-f]{64}", observed["targetGeneration"]) is not None + and argv[0] == "send" and len(targets) == 1 + and isinstance(targets[0], str) and bool(targets[0]) + and AGENTS.for_target(targets[0]).name == "claude" + and requested == [observed["targetGeneration"]] + and result["ok"] is False and result.get("status") == "refused" + and result.get("reason") == "stale_target" and result.get("submitted") is False + and result.get("retryAllowed") is False + ) + if not last_seen_valid: + # Bad optional refusal metadata cannot prove no effect. It also + # cannot erase a complete positive response accepted by the + # existing ordinary parser. Keep its existing native semantics without reflecting + # the invalid optional metadata or granting resend permission. + if result["ok"] is True or result.get("submitted") is True: + result.pop("lastSeenTarget") + else: + valid = False return stdout, result, bool(stdout), valid @@ -3421,6 +3624,11 @@ def identity(self, target: str, context: ExecutionContext) -> SessionIdentity: return SessionIdentity(self.name, context.host, target[len(prefix):]) def validate_send(self, args: argparse.Namespace, text: str | None = None) -> None: + expected = getattr(args, "target_generation", None) + validate_target_generation(expected) + if expected is not None and self.name != "claude": + raise generation_refused("unsupported_target_generation", + "This native transport cannot atomically bind an inbox incarnation; nothing sent") if not self.capabilities.send: raise AdapterError(self.name, "unsupported_capability", "Agent does not support send") if getattr(args, "wake", False) and not self.capabilities.wake: @@ -3481,11 +3689,25 @@ def list(self, context: ExecutionContext) -> DiscoveryResult: def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: args = context.options - session = resolve_target(discover(include_unreachable=True), args.to) + expected = getattr(args, "target_generation", None) + try: + session = resolve_target(discover(include_unreachable=True), args.to) + except CcPeerError as exc: + if expected is not None: + raise generation_refused("stale_target", "Pinned target disappeared or became ambiguous; nothing sent", + expected_generation=expected) from exc + raise + if expected is not None: + require_claude_generation(session, expected) if not args.dry_run: - post_to_socket(session["socket"], text, pid=session["pid"]) + 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"]) return {"ok": True, "target": {"pid": session["pid"], "name": session["name"]}, - "chars": len(text), "dryRun": args.dry_run} + "chars": len(text), "dryRun": args.dry_run, + **({"targetGeneration": expected} if expected is not None else {})} def diagnose(self, context: ExecutionContext) -> dict: return diagnose_claude() @@ -3512,6 +3734,7 @@ def submission_text(self, result: dict, where: str) -> str: def remote_submission(self, result: dict, args: argparse.Namespace, text: str) -> dict: return {"target": result.get("target", {}), **ssh_metadata_from(result), + **({"targetGeneration": result["targetGeneration"]} if "targetGeneration" in result else {}), "chars": result.get("chars", len(text)), "dryRun": args.dry_run} @@ -4039,6 +4262,10 @@ def execute(self, operation: str, adapter: AgentAdapter, # destroy successful results from other adapters, even in JSON mode. for row in result["sessions"]: adapter.display_row(row) + if getattr(args, "with_target_generation", False): + value = claude_generation(row) if adapter.name == "claude" else None + row["targetGeneration"] = value + row["generationStatus"] = "available" if value else "unsupported" return result if operation == "send": check_message(text, remote=False) @@ -4143,6 +4370,8 @@ def render_listing(payload: dict, where: str, selected: str | None) -> str: def cmd_list(args: argparse.Namespace) -> int: + if getattr(args, "device", None) and getattr(args, "with_target_generation", False): + raise generation_refused("unsupported_target_generation", "Paired devices use a separate generation contract") if getattr(args, "device", None): return optional_relay().invoke_core(args) selected = getattr(args, "agent", None) @@ -4160,6 +4389,8 @@ def cmd_list(args: argparse.Namespace) -> int: transport = SshTransport(requested_host, args, tailnet_status) host, ssh_opts = transport.host, transport.ssh_opts argv = ["list", "--no-update-notice"] + (["--all"] if args.all else []) + if getattr(args, "with_target_generation", False): + argv.append("--with-target-generation") if selected: argv += ["--agent", selected] argv += agent_remote_options(args, selected) @@ -5117,6 +5348,10 @@ def cmd_update(args: argparse.Namespace) -> int: def cmd_send(args: argparse.Namespace) -> int: + expected_generation = getattr(args, "target_generation", None) + validate_target_generation(expected_generation) + if getattr(args, "device", None) and expected_generation is not None: + raise generation_refused("unsupported_target_generation", "Paired devices use a separate generation contract") if getattr(args, "device", None): return optional_relay().invoke_core(args) resolved_address = apply_reply_target(args) @@ -5209,9 +5444,22 @@ def cmd_send(args: argparse.Namespace) -> int: encoded = base64.b64encode(text.encode("utf-8")).decode("ascii") remote_argv = ["send", "--no-update-notice", "--to", args.to, "--b64", encoded] remote_argv += adapter.remote_options(args) + if expected_generation is not None: + remote_argv += ["--target-generation", expected_generation] if args.dry_run: remote_argv.append("--dry-run") result = transport.execute(remote_argv) + if expected_generation is not None and result.get("targetGeneration") != expected_generation: + # Failed generation evidence does not erase a complete native + # response. Retain its reported facts without inventing Claude's + # normally absent submission fields or authorizing another send. + native_facts = {key: result[key] for key in ( + "target", "chars", "dryRun", "submitted", "consumptionConfirmed", + "queueId", + ) if key in result} + raise CcPeerError("SSH response did not preserve the requested generation; do not retry automatically", + {**native_facts, **ssh_metadata_from(result), + "status": "unknown", "reason": "outcome_unknown", "retryAllowed": False}) payload = adapter.remote_submission(result, args, text) payload.update(host_metadata(requested_host, host)) payload.update(routing_metadata) @@ -5338,6 +5586,8 @@ def add_common(sub: argparse.ArgumentParser) -> None: "set explicitly for Orca/multiple homes") listing.add_argument("--codex-home", help="list only this destination home (default: known default, CODEX_HOME, Orca and configured homes)") listing.add_argument("--codex-bin", help="Codex executable on the destination (used by send)") + listing.add_argument("--with-target-generation", action="store_true", + help="report optional inbox generation preconditions (not ACK or consumption)") listing.add_argument( "--all", action="store_true", help="include stale records and sessions with no inbox" ) @@ -5367,6 +5617,8 @@ def add_common(sub: argparse.ArgumentParser) -> None: help="session name, PID, codex:UUID, antigravity:UUID, or session-peer://v1/reply address", ) sending.add_argument("--codex-home", help=home_help) + sending.add_argument("--target-generation", metavar="TOKEN", + help="require a previously discovered inbox generation; fail closed if unsupported") sending.add_argument( "--allow-inactive-codex-home", action="store_true", help="with --codex-home, intentionally queue an inactive thread for a future resume", diff --git a/session_peer_core/adapters.py b/session_peer_core/adapters.py index 4761d058..d00fea10 100644 --- a/session_peer_core/adapters.py +++ b/session_peer_core/adapters.py @@ -67,6 +67,11 @@ def identity(self, target: str, context: ExecutionContext) -> SessionIdentity: return SessionIdentity(self.name, context.host, target[len(prefix):]) def validate_send(self, args: argparse.Namespace, text: str | None = None) -> None: + expected = getattr(args, "target_generation", None) + validate_target_generation(expected) + if expected is not None and self.name != "claude": + raise generation_refused("unsupported_target_generation", + "This native transport cannot atomically bind an inbox incarnation; nothing sent") if not self.capabilities.send: raise AdapterError(self.name, "unsupported_capability", "Agent does not support send") if getattr(args, "wake", False) and not self.capabilities.wake: @@ -127,11 +132,25 @@ def list(self, context: ExecutionContext) -> DiscoveryResult: def submit(self, context: ExecutionContext, text: str) -> SubmissionResult: args = context.options - session = resolve_target(discover(include_unreachable=True), args.to) + expected = getattr(args, "target_generation", None) + try: + session = resolve_target(discover(include_unreachable=True), args.to) + except CcPeerError as exc: + if expected is not None: + raise generation_refused("stale_target", "Pinned target disappeared or became ambiguous; nothing sent", + expected_generation=expected) from exc + raise + if expected is not None: + require_claude_generation(session, expected) if not args.dry_run: - post_to_socket(session["socket"], text, pid=session["pid"]) + 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"]) return {"ok": True, "target": {"pid": session["pid"], "name": session["name"]}, - "chars": len(text), "dryRun": args.dry_run} + "chars": len(text), "dryRun": args.dry_run, + **({"targetGeneration": expected} if expected is not None else {})} def diagnose(self, context: ExecutionContext) -> dict: return diagnose_claude() @@ -158,6 +177,7 @@ def submission_text(self, result: dict, where: str) -> str: def remote_submission(self, result: dict, args: argparse.Namespace, text: str) -> dict: return {"target": result.get("target", {}), **ssh_metadata_from(result), + **({"targetGeneration": result["targetGeneration"]} if "targetGeneration" in result else {}), "chars": result.get("chars", len(text)), "dryRun": args.dry_run} diff --git a/session_peer_core/claude.py b/session_peer_core/claude.py index fb6a3d51..6cb64e9d 100644 --- a/session_peer_core/claude.py +++ b/session_peer_core/claude.py @@ -270,7 +270,8 @@ def _read_win_auth(pid: int) -> str | None: return None -def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: +def _post_to_pipe(pipe_path: str, pid: int, text: str, *, generation_session=None, + expected_generation=None) -> None: """Write one message to a Windows named pipe inbox.""" check_message(text, remote=False) @@ -286,6 +287,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: import time try: with open(pipe_path, "wb") as pipe: + 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")) pipe.write((payload + "\n").encode("utf-8")) pipe.flush() @@ -294,7 +297,8 @@ def _post_to_pipe(pipe_path: str, pid: int, text: str) -> None: raise CcPeerError(f"cannot reach inbox at {pipe_path}: {exc}") from exc -def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: +def post_to_socket(socket_path: str, text: str, pid: int = 0, *, generation_session=None, + expected_generation=None, effect_deadline=None, total_deadline=None) -> None: """Write one message to a session's inbox socket. On macOS and Linux the {"type":"auth",...} line the docs describe is @@ -304,8 +308,22 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: a complete line within 30 seconds, so the message is built before the socket is opened. """ + bounded = effect_deadline is not None or total_deadline is not None + if bounded: + import math + if (type(effect_deadline) not in (int, float) or type(total_deadline) not in (int, float) + or not math.isfinite(effect_deadline) or not math.isfinite(total_deadline) + or effect_deadline > total_deadline): + raise generation_refused("invalid_effect_deadline", "Invalid private effect budget; nothing sent") + if IS_WINDOWS: + # Python's synchronous named-pipe open/write has no proven bounded + # cancellation contract. Never advertise it as bounded handoff. + raise generation_refused("unsupported_bounded_inbox", "Bounded Windows pipe submission is unsupported") + if time.monotonic() >= effect_deadline: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") if IS_WINDOWS and socket_path.startswith("\\\\.\\pipe\\"): - _post_to_pipe(socket_path, pid, text) + _post_to_pipe(socket_path, pid, text, generation_session=generation_session, + expected_generation=expected_generation) return check_message(text, remote=False) @@ -316,22 +334,41 @@ def post_to_socket(socket_path: str, text: str, pid: int = 0) -> None: ) conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - conn.settimeout(CONNECT_TIMEOUT) try: + remaining = effect_deadline - time.monotonic() if bounded else CONNECT_TIMEOUT + if remaining <= 0: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") + conn.settimeout(min(CONNECT_TIMEOUT, remaining)) try: conn.connect(socket_path) except OSError as exc: + if bounded: + raise generation_refused("inbox_connection_failed", "Inbox connection failed before any write") from exc raise CcPeerError(f"cannot reach inbox at {socket_path}: {exc}") from exc try: + if expected_generation is not None: + verify_connected_generation(generation_session, expected_generation, connected_inbox_pid(conn)) + if bounded: + remaining = effect_deadline - time.monotonic() + if remaining <= 0: + raise generation_refused("effect_deadline_exhausted", "Effect budget exhausted; nothing sent") + conn.settimeout(remaining) conn.sendall((payload + "\n").encode("utf-8")) conn.shutdown(socket.SHUT_WR) - conn.settimeout(DRAIN_TIMEOUT) + remaining = effect_deadline - time.monotonic() if bounded else DRAIN_TIMEOUT + if remaining <= 0: + return # Full write retained; no further wait beyond cutoff. + conn.settimeout(min(DRAIN_TIMEOUT, remaining)) try: conn.recv(1) except OSError: pass except OSError as exc: + if bounded: + raise CcPeerError("Inbox write outcome unknown; do not automatically retry", + {"status": "unknown", "reason": "outcome_unknown", + "retryAllowed": False}) from exc raise CcPeerError(f"failed writing to {socket_path}: {exc}") from exc finally: conn.close() diff --git a/session_peer_core/cli.py b/session_peer_core/cli.py index 73abbc11..7f443ee7 100644 --- a/session_peer_core/cli.py +++ b/session_peer_core/cli.py @@ -54,6 +54,8 @@ def add_common(sub: argparse.ArgumentParser) -> None: "set explicitly for Orca/multiple homes") listing.add_argument("--codex-home", help="list only this destination home (default: known default, CODEX_HOME, Orca and configured homes)") listing.add_argument("--codex-bin", help="Codex executable on the destination (used by send)") + listing.add_argument("--with-target-generation", action="store_true", + help="report optional inbox generation preconditions (not ACK or consumption)") listing.add_argument( "--all", action="store_true", help="include stale records and sessions with no inbox" ) @@ -83,6 +85,8 @@ def add_common(sub: argparse.ArgumentParser) -> None: help="session name, PID, codex:UUID, antigravity:UUID, or session-peer://v1/reply address", ) sending.add_argument("--codex-home", help=home_help) + sending.add_argument("--target-generation", metavar="TOKEN", + help="require a previously discovered inbox generation; fail closed if unsupported") sending.add_argument( "--allow-inactive-codex-home", action="store_true", help="with --codex-home, intentionally queue an inactive thread for a future resume", diff --git a/session_peer_core/commands.py b/session_peer_core/commands.py index 48d49a62..ccf33173 100644 --- a/session_peer_core/commands.py +++ b/session_peer_core/commands.py @@ -1,4 +1,6 @@ def cmd_list(args: argparse.Namespace) -> int: + if getattr(args, "device", None) and getattr(args, "with_target_generation", False): + raise generation_refused("unsupported_target_generation", "Paired devices use a separate generation contract") if getattr(args, "device", None): return optional_relay().invoke_core(args) selected = getattr(args, "agent", None) @@ -16,6 +18,8 @@ def cmd_list(args: argparse.Namespace) -> int: transport = SshTransport(requested_host, args, tailnet_status) host, ssh_opts = transport.host, transport.ssh_opts argv = ["list", "--no-update-notice"] + (["--all"] if args.all else []) + if getattr(args, "with_target_generation", False): + argv.append("--with-target-generation") if selected: argv += ["--agent", selected] argv += agent_remote_options(args, selected) diff --git a/session_peer_core/registry.py b/session_peer_core/registry.py index 038d9a5a..7e739fac 100644 --- a/session_peer_core/registry.py +++ b/session_peer_core/registry.py @@ -65,6 +65,10 @@ def execute(self, operation: str, adapter: AgentAdapter, # destroy successful results from other adapters, even in JSON mode. for row in result["sessions"]: adapter.display_row(row) + if getattr(args, "with_target_generation", False): + value = claude_generation(row) if adapter.name == "claude" else None + row["targetGeneration"] = value + row["generationStatus"] = "available" if value else "unsupported" return result if operation == "send": check_message(text, remote=False) diff --git a/session_peer_core/sending.py b/session_peer_core/sending.py index 0b2e7a17..8f0b5e0f 100644 --- a/session_peer_core/sending.py +++ b/session_peer_core/sending.py @@ -1,4 +1,8 @@ def cmd_send(args: argparse.Namespace) -> int: + expected_generation = getattr(args, "target_generation", None) + validate_target_generation(expected_generation) + if getattr(args, "device", None) and expected_generation is not None: + raise generation_refused("unsupported_target_generation", "Paired devices use a separate generation contract") if getattr(args, "device", None): return optional_relay().invoke_core(args) resolved_address = apply_reply_target(args) @@ -91,9 +95,22 @@ def cmd_send(args: argparse.Namespace) -> int: encoded = base64.b64encode(text.encode("utf-8")).decode("ascii") remote_argv = ["send", "--no-update-notice", "--to", args.to, "--b64", encoded] remote_argv += adapter.remote_options(args) + if expected_generation is not None: + remote_argv += ["--target-generation", expected_generation] if args.dry_run: remote_argv.append("--dry-run") result = transport.execute(remote_argv) + if expected_generation is not None and result.get("targetGeneration") != expected_generation: + # Failed generation evidence does not erase a complete native + # response. Retain its reported facts without inventing Claude's + # normally absent submission fields or authorizing another send. + native_facts = {key: result[key] for key in ( + "target", "chars", "dryRun", "submitted", "consumptionConfirmed", + "queueId", + ) if key in result} + raise CcPeerError("SSH response did not preserve the requested generation; do not retry automatically", + {**native_facts, **ssh_metadata_from(result), + "status": "unknown", "reason": "outcome_unknown", "retryAllowed": False}) payload = adapter.remote_submission(result, args, text) payload.update(host_metadata(requested_host, host)) payload.update(routing_metadata) diff --git a/session_peer_core/ssh.py b/session_peer_core/ssh.py index 5561c026..f7aca075 100644 --- a/session_peer_core/ssh.py +++ b/session_peer_core/ssh.py @@ -192,6 +192,42 @@ def constant_value(value): and type(result.get("ok")) is bool and bool(argv) and result.get("command") == argv[0] ) + if valid and "lastSeenTarget" in result: + observed = result["lastSeenTarget"] + requested = [] + targets = [] + for index, argument in enumerate(argv): + if not isinstance(argument, str): + continue + if argument == "--target-generation" and index + 1 < len(argv): + requested.append(argv[index + 1]) + elif argument.startswith("--target-generation="): + requested.append(argument.split("=", 1)[1]) + if argument == "--to" and index + 1 < len(argv): + targets.append(argv[index + 1]) + elif argument.startswith("--to="): + targets.append(argument.split("=", 1)[1]) + last_seen_valid = ( + isinstance(observed, dict) and set(observed) == {"agent", "targetGeneration"} + and observed["agent"] == "claude" and isinstance(observed["targetGeneration"], str) + and re.fullmatch(r"tg1:[0-9a-f]{64}", observed["targetGeneration"]) is not None + and argv[0] == "send" and len(targets) == 1 + and isinstance(targets[0], str) and bool(targets[0]) + and AGENTS.for_target(targets[0]).name == "claude" + and requested == [observed["targetGeneration"]] + and result["ok"] is False and result.get("status") == "refused" + and result.get("reason") == "stale_target" and result.get("submitted") is False + and result.get("retryAllowed") is False + ) + if not last_seen_valid: + # Bad optional refusal metadata cannot prove no effect. It also + # cannot erase a complete positive response accepted by the + # existing ordinary parser. Keep its existing native semantics without reflecting + # the invalid optional metadata or granting resend permission. + if result["ok"] is True or result.get("submitted") is True: + result.pop("lastSeenTarget") + else: + valid = False return stdout, result, bool(stdout), valid diff --git a/session_peer_core/target_generation.py b/session_peer_core/target_generation.py new file mode 100644 index 00000000..ba8e0d6d --- /dev/null +++ b/session_peer_core/target_generation.py @@ -0,0 +1,128 @@ +# Optional discovery-to-inbox preconditions. These are not ACK or dedup keys. + +def generation_refused(reason: str, message: str, *, expected_generation=None) -> CcPeerError: + details = {"status": "refused", "reason": reason, + "submitted": False, "retryAllowed": False} + # Original discovery precondition only: never reflect a successor's name, + # PID, host, path or native creation evidence as "last seen" metadata. + if (reason == "stale_target" and isinstance(expected_generation, str) + and re.fullmatch(r"tg1:[0-9a-f]{64}", expected_generation)): + details["lastSeenTarget"] = {"agent": "claude", "targetGeneration": expected_generation} + return CcPeerError(message, details) + + +def process_generation(pid: int) -> str | None: + """Use native creation evidence, never the second-resolution ps display.""" + if type(pid) is not int or pid <= 1: + return None + try: + if sys.platform == "linux": + # comm may contain spaces and ')'; fields after the final ')' start + # at field 3, so starttime (22) is offset 19. + fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split() + boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + return f"linux:{boot}:{fields[19]}" if fields[19].isdigit() else None + if sys.platform == "darwin": + import ctypes + class BsdInfo(ctypes.Structure): + _fields_ = [("prefix", ctypes.c_uint32 * 12), + ("comm", ctypes.c_char * 16), ("name", ctypes.c_char * 32), + ("suffix", ctypes.c_uint32 * 6), + ("seconds", ctypes.c_uint64), ("micros", ctypes.c_uint64)] + lib = ctypes.CDLL("/usr/lib/libproc.dylib", use_errno=True) + lib.proc_pidinfo.argtypes = (ctypes.c_int, ctypes.c_int, ctypes.c_uint64, + ctypes.c_void_p, ctypes.c_int) + lib.proc_pidinfo.restype = ctypes.c_int + value = BsdInfo() + if lib.proc_pidinfo(pid, 3, 0, ctypes.byref(value), ctypes.sizeof(value)) != ctypes.sizeof(value): + return None + if value.prefix[3] != pid or not value.seconds: + return None + return f"darwin:{value.seconds}:{value.micros}" + if sys.platform == "win32": + rows = _windows_process_inspect("identity", str(pid)) + return "win32:" + rows[0]["startTime"] if rows and len(rows) == 1 else None + except (OSError, ValueError, IndexError, AttributeError): + pass + return None + + +def claude_generation(session: dict) -> str | None: + """Hash bounded registry/process/endpoint facts; expose no raw native key.""" + import hashlib + try: + pid = session["pid"] + birth = process_generation(pid) + if not birth: + return None + directory = sessions_dir().resolve() + record = read_claude_record(directory / f"{pid}.json") + endpoint = record.get("messagingSocketPath") + if record["pid"] != pid or endpoint != session["socket"]: + return None + identity = ["claude", sys.platform, socket.gethostname(), str(directory), pid, birth, endpoint] + if not IS_WINDOWS: + node = Path(endpoint).lstat() + if not stat.S_ISSOCK(node.st_mode): + return None + identity += [node.st_dev, node.st_ino] + else: + if not endpoint.startswith("\\\\.\\pipe\\"): + return 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): + return None + + +def validate_target_generation(value: str | None) -> None: + if value is not None and (not isinstance(value, str) or not re.fullmatch(r"tg1:[0-9a-f]{64}", value)): + raise generation_refused("invalid_target_generation", "Invalid target generation; run discovery again") + + +def require_claude_generation(session: dict, expected: str) -> None: + validate_target_generation(expected) + current = claude_generation(session) + if current is None: + raise generation_refused("target_generation_unavailable", "Cannot prove the selected inbox generation; nothing sent") + if current != expected: + raise generation_refused("stale_target", "The selected inbox generation changed; nothing sent", + expected_generation=expected) + + +def connected_inbox_pid(conn) -> int | None: + """Read credentials for the connected endpoint, not its current pathname.""" + import struct + try: + if sys.platform == "linux" and hasattr(socket, "SO_PEERCRED"): + return struct.unpack("3i", conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[0] + if sys.platform == "darwin": + # SOL_LOCAL / LOCAL_PEERPID from sys/un.h. + return struct.unpack("i", conn.getsockopt(0, 2, 4))[0] + except (OSError, ValueError, struct.error): + pass + return None + + +def connected_pipe_pid(pipe) -> int | None: + try: + import ctypes + from ctypes import wintypes + import msvcrt + function = ctypes.windll.kernel32.GetNamedPipeServerProcessId + function.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.ULONG)) + function.restype = wintypes.BOOL + pid = wintypes.ULONG() + return pid.value if function(msvcrt.get_osfhandle(pipe.fileno()), ctypes.byref(pid)) else None + except (OSError, AttributeError, ValueError): + return None + + +def verify_connected_generation(session: dict, expected: str, server_pid: int | None) -> None: + if server_pid is None: + raise generation_refused("target_generation_unavailable", "Connected inbox identity unavailable; nothing sent") + if server_pid != session["pid"]: + raise generation_refused("stale_target", "Connected inbox belongs to another process; nothing sent", + expected_generation=expected) + require_claude_generation(session, expected) diff --git a/tests/core/test_peer_framing.py b/tests/core/test_peer_framing.py index 08efc6b8..03a99112 100644 --- a/tests/core/test_peer_framing.py +++ b/tests/core/test_peer_framing.py @@ -114,3 +114,10 @@ def test_remote_claude_chars_preserve_actual_receiving_frame_length(self): self.assertNotEqual(result['chars'], len('hello')) self.assertNotIn('submitted', result) self.assertNotIn('consumptionConfirmed', result) + self.assertNotIn('targetGeneration', result) + native['targetGeneration'] = 'tg1:' + 'a' * 64 + pinned = peer.AGENTS.get('claude').remote_submission(native, args, 'hello') + self.assertEqual(pinned['targetGeneration'], native['targetGeneration']) + self.assertEqual(pinned['chars'], native['chars']) + self.assertNotIn('submitted', pinned) + self.assertNotIn('consumptionConfirmed', pinned) diff --git a/tests/core/test_target_generation.py b/tests/core/test_target_generation.py new file mode 100644 index 00000000..ae2d587a --- /dev/null +++ b/tests/core/test_target_generation.py @@ -0,0 +1,375 @@ +"""Offline generation pinning: owned endpoints, synthetic reuse and fake SSH.""" +import argparse +import contextlib +import io +import json +import os +from pathlib import Path +import selectors +import socket +import subprocess +import tempfile +import threading +import unittest +from unittest import mock + +import session_peer as peer + + +class TargetGeneration(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory(prefix="codex-generation-") + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.sessions = self.root / "sessions" + self.sessions.mkdir() + self.pid = os.getpid() + self.sock = str(self.root / "inbox.sock") + self.record = {"pid": self.pid, "name": "worker", "messagingSocketPath": self.sock, + "startedAt": 1000} + self.path = self.sessions / f"{self.pid}.json" + self.write_record() + self.row = {"pid": self.pid, "name": "worker", "socket": self.sock, + "reachable": True, "alive": True, "status": "idle", "cwd": "/fixture", "agent": "claude"} + patcher = mock.patch.object(peer, "sessions_dir", return_value=self.sessions) + patcher.start() + self.addCleanup(patcher.stop) + if not peer.IS_WINDOWS: + self.listener = socket.socket(socket.AF_UNIX) + self.listener.bind(self.sock) + self.listener.listen(1) + self.addCleanup(self.listener.close) + else: + self.sock = self.record["messagingSocketPath"] = r"\\.\pipe\generation-test" + self.row["socket"] = self.sock + self.write_record() + + def write_record(self): + self.path.write_text(json.dumps(self.record), encoding="utf-8") + + def token(self): + with mock.patch.object(peer, "process_generation", return_value="precise-birth-1"): + return peer.claude_generation(self.row) + + def invoke(self, *argv): + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = peer.main([*argv, "--json", "--no-update-notice"]) + return code, json.loads(out.getvalue()) + + def test_legacy_output_unchanged_without_opt_in(self): + args = argparse.Namespace(to="worker", dry_run=True) + with mock.patch.object(peer, "discover", return_value=[self.row]): + result = peer.ClaudeAdapter().submit(peer.ExecutionContext("local", args), "hello") + self.assertNotIn("targetGeneration", result) + self.assertNotIn("submitted", result) + + def test_same_generation_and_changed_birth(self): + token = self.token() + self.assertRegex(token, r"^tg1:[0-9a-f]{64}$") + with mock.patch.object(peer, "process_generation", return_value="precise-birth-1"): + peer.require_claude_generation(self.row, token) + with mock.patch.object(peer, "process_generation", return_value="precise-birth-2"): + with self.assertRaises(peer.CcPeerError) as caught: + peer.require_claude_generation(self.row, token) + self.assertEqual(caught.exception.details["reason"], "stale_target") + self.assertFalse(caught.exception.details["submitted"]) + + def test_status_and_name_change_do_not_change_incarnation(self): + token = self.token() + self.record.update(status="busy", name="renamed") + self.write_record() + self.assertEqual(token, self.token()) + + def test_same_pid_other_home_or_socket_changes_token(self): + token = self.token() + other = self.root / "other" + other.mkdir() + (other / self.path.name).write_text(json.dumps(self.record)) + with mock.patch.object(peer, "sessions_dir", return_value=other): + self.assertNotEqual(token, self.token()) + self.record["messagingSocketPath"] += "-successor" + self.write_record() + self.assertIsNone(self.token()) + + def test_unavailable_creation_evidence_is_not_a_token(self): + with mock.patch.object(peer, "process_generation", return_value=None): + self.assertIsNone(peer.claude_generation(self.row)) + with self.assertRaises(peer.CcPeerError) as caught: + peer.require_claude_generation(self.row, self.token() or "tg1:" + "a" * 64) + self.assertEqual(caught.exception.details["reason"], "target_generation_unavailable") + + def test_stale_name_or_pid_refuses_before_post(self): + token = self.token() + for rows in ([], [{**self.row, "pid": self.pid + 1}], [self.row, self.row]): + with mock.patch.object(peer, "discover", return_value=rows), \ + mock.patch.object(peer, "post_to_socket") as post, \ + mock.patch.object(peer, "process_generation", return_value="precise-birth-1"): + code, result = self.invoke("send", "--to", "worker", "--message=x", + "--no-from", "--no-reply-to", "--target-generation", token) + self.assertEqual(code, 1) + self.assertIn(result["reason"], ("stale_target", "target_generation_unavailable")) + post.assert_not_called() + + def test_native_profiles_without_atomic_generation_are_refused(self): + with mock.patch.object(peer, "subprocess") as native: + code, result = self.invoke("send", "--to", "codex:11111111-1111-1111-1111-111111111111", + "--target-generation", "tg1:" + "a" * 64, + "--no-from", "--no-reply-to", "--message=x") + self.assertEqual(code, 1) + self.assertEqual(result["reason"], "unsupported_target_generation") + native.run.assert_not_called() + + def test_codex_restart_cannot_upgrade_unsupported_generation_or_start_effect(self): + # This tests the unsupported boundary, not atomic Codex queue pinning. + # Discovery before/after a synthetic writer restart must never mint a + # Claude inbox token for the saved Codex UUID. Neither dry-run, wake nor + # SSH may turn that unsupported request into a native submission. + thread_id = "11111111-1111-1111-1111-111111111111" + for owner in (41, 42): + row = {"agent": "codex", "id": thread_id, "name": "worker", + "status": "active", "archived": False, "cwd": "/fixture", "codexHome": "/fixture/codex", + "ownerPid": owner} + with mock.patch.object(peer.CodexAdapter, "list", return_value={ + "sessions": [row], "discovery": {"status": "ok"}}): + code, result = self.invoke("list", "--agent", "codex", "--with-target-generation") + self.assertEqual(code, 0) + self.assertIsNone(result["sessions"][0]["targetGeneration"]) + self.assertEqual(result["sessions"][0]["generationStatus"], "unsupported") + + for options in ([], ["--dry-run"], ["--wake"], ["--wake", "--dry-run"], + ["--host", "test-host"], ["--host", "test-host", "--dry-run"], + ["--host", "test-host", "--wake"]): + with self.subTest(options=options), \ + mock.patch.object(peer, "queue_codex") as queue, \ + mock.patch.object(peer, "run_codex_wake") as wake, \ + mock.patch.object(peer.SshTransport, "execute") as remote, \ + mock.patch.object(peer, "tailscale_status") as tailnet, \ + mock.patch.object(peer.subprocess, "run") as child: + code, result = self.invoke("send", "--to", "codex:" + thread_id, + "--target-generation", self.token(), "--message=x", + "--no-from", "--no-reply-to", *options) + self.assertEqual(code, 1) + self.assertEqual(result["reason"], "unsupported_target_generation") + self.assertEqual(result["status"], "refused") + self.assertFalse(result["submitted"]) + self.assertFalse(result["retryAllowed"]) + for effect in (queue, wake, remote, tailnet, child): + effect.assert_not_called() + + def test_malformed_and_paired_generation_rejected_before_effect(self): + for token in ("bad", "tg1:" + "a" * 63, "tg1:" + "A" * 64): + with mock.patch.object(peer, "post_to_socket") as post: + code, result = self.invoke("send", "--to", "worker", "--message=x", + "--target-generation", token) + self.assertEqual(code, 1) + self.assertEqual(result["reason"], "invalid_target_generation") + post.assert_not_called() + with mock.patch.object(peer, "optional_relay") as relay: + code, result = self.invoke("send", "--to", "worker", "--device", "device", + "--target-generation", "tg1:" + "a" * 64, "--message=x") + self.assertEqual(code, 1) + relay.assert_not_called() + + def test_opt_in_discovery_and_remote_arg_forwarding(self): + with mock.patch.object(peer, "discover", return_value=[self.row]), \ + mock.patch.object(peer, "process_generation", return_value="precise-birth-1"): + code, result = self.invoke("list", "--agent", "claude", "--with-target-generation") + self.assertEqual(code, 0) + self.assertEqual(result["sessions"][0]["targetGeneration"], self.token()) + with mock.patch.object(peer, "tailscale_status", return_value={}), \ + mock.patch.object(peer, "resolve_ssh_destination", return_value="test-host"), \ + mock.patch.object(peer, "tailscale_ssh_options", return_value=[]), \ + mock.patch.object(peer.SshTransport, "execute", return_value={"ok": True, "target": {"pid": self.pid}, + "targetGeneration": self.token()}) as remote: + code, result = self.invoke("send", "--host", "test-host", "--to", "worker", + "--target-generation", self.token(), "--message=x", + "--no-from", "--no-reply-to") + self.assertEqual(code, 0) + self.assertIn("--target-generation", remote.call_args[0][0]) + self.assertEqual(result["targetGeneration"], self.token()) + + def test_remote_missing_or_wrong_generation_retains_native_facts_without_resend(self): + token = self.token() + native = {"schemaVersion": 1, "command": "send", "ok": True, + "target": {"pid": self.pid, "name": "worker"}, "chars": 1, + "dryRun": False, "sshUser": "fixture", "sshUserSource": "explicit"} + for observed in (None, "tg1:" + "f" * 64): + reply = dict(native) + if observed is not None: + reply["targetGeneration"] = observed + with mock.patch.object(peer, "tailscale_status", return_value={}), \ + mock.patch.object(peer, "resolve_ssh_destination", return_value="test-host"), \ + mock.patch.object(peer, "tailscale_ssh_options", return_value=[]), \ + mock.patch.object(peer.SshTransport, "execute", return_value=reply) as remote: + code, result = self.invoke("send", "--host", "test-host", "--to", "worker", + "--target-generation", token, "--message=x", + "--no-from", "--no-reply-to") + self.assertEqual(code, 1) + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "unknown") + self.assertEqual(result["target"], native["target"]) + self.assertEqual(result["chars"], 1) + self.assertFalse(result["dryRun"]) + self.assertEqual(result["sshUser"], "fixture") + self.assertFalse(result["retryAllowed"]) + self.assertNotIn("submitted", result) + self.assertNotIn("consumptionConfirmed", result) + self.assertNotIn("targetGeneration", result) + remote.assert_called_once() + + def test_remote_completed_generation_refusal_is_not_downgraded(self): + token = self.token() + failure = peer.generation_refused("stale_target", "Pinned target changed") + with mock.patch.object(peer, "tailscale_status", return_value={}), \ + mock.patch.object(peer, "resolve_ssh_destination", return_value="test-host"), \ + mock.patch.object(peer, "tailscale_ssh_options", return_value=[]), \ + mock.patch.object(peer.SshTransport, "execute", side_effect=failure) as remote: + code, result = self.invoke("send", "--host", "test-host", "--to", "worker", + "--target-generation", token, "--message=x", + "--no-from", "--no-reply-to") + self.assertEqual(code, 1) + self.assertEqual(result["status"], "refused") + self.assertEqual(result["reason"], "stale_target") + self.assertFalse(result["submitted"]) + self.assertFalse(result["retryAllowed"]) + remote.assert_called_once() + + @unittest.skipUnless(peer.sys.platform in ("darwin", "linux"), "owned Unix inbox") + def test_owned_socket_peer_identity_and_zero_write_on_wrong_peer(self): + token = peer.claude_generation(self.row) + self.assertIsNotNone(token) + received = [] + def receive(): + self.listener.settimeout(5) + conn, _ = self.listener.accept() + with conn: + conn.settimeout(5) + received.append(conn.recv(65536)) + thread = threading.Thread(target=receive) + thread.start() + peer.post_to_socket(self.sock, "owned fixture", self.pid, + generation_session=self.row, expected_generation=token) + thread.join(6) + self.assertFalse(thread.is_alive()) + self.assertIn(b"owned fixture", received[0]) + thread = threading.Thread(target=receive) + thread.start() + with mock.patch.object(peer, "connected_inbox_pid", return_value=self.pid + 1): + with self.assertRaises(peer.CcPeerError) as caught: + peer.post_to_socket(self.sock, "must not arrive", self.pid, + generation_session=self.row, expected_generation=token) + thread.join(6) + self.assertEqual(caught.exception.details["reason"], "stale_target") + self.assertEqual(received[1], b"") + + def test_final_birth_change_and_missing_peer_refuse(self): + token = self.token() + for pid, birth, reason in ((None, "precise-birth-1", "target_generation_unavailable"), + (self.pid, "precise-birth-2", "stale_target")): + with mock.patch.object(peer, "process_generation", return_value=birth): + with self.assertRaises(peer.CcPeerError) as caught: + peer.verify_connected_generation(self.row, token, pid) + self.assertEqual(caught.exception.details["reason"], reason) + + @unittest.skipUnless(peer.sys.platform in ("darwin", "linux"), "owned Unix process incarnation") + def test_actual_process_restart_reusing_name_never_receives_stale_pin(self): + fixture = Path(__file__).parents[1] / "fixtures" / "claude_generation_inbox.py" + endpoint = self.root / "restart.sock" + counter = self.root / "writes.txt" + name = "owned-restarted-inbox" + + def spawn(): + process = subprocess.Popen([peer.sys.executable, str(fixture), str(self.sessions), + str(endpoint), str(counter), name], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + start_new_session=True) + self.addCleanup(stop, process) + with selectors.DefaultSelector() as selector: + selector.register(process.stdout, selectors.EVENT_READ) + self.assertTrue(selector.select(5), "owned inbox did not become ready") + raw = os.read(process.stdout.fileno(), 512) + self.assertEqual(json.loads(raw)["pid"], process.pid) + return process + + def stop(process): + if process.poll() is None: + process.terminate() + try: + process.communicate(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.communicate(timeout=5) + + def observed(): + code, result = self.invoke("list", "--agent", "claude", "--with-target-generation") + self.assertEqual(code, 0) + rows = [row for row in result["sessions"] if row["name"] == name] + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["generationStatus"], "available") + return rows[0]["targetGeneration"] + + first = spawn() + original = observed() + self.assertEqual(counter.read_text(encoding="ascii"), "0") + stop(first) + endpoint.unlink() # exact owned fixture socket, not a user inbox + second = spawn() + successor = observed() + self.assertNotEqual(successor, original) + for mode in ([], ["--dry-run"]): + code, result = self.invoke("send", "--to", name, "--message=must not arrive", + "--no-from", "--no-reply-to", "--target-generation", original, *mode) + self.assertEqual(code, 1) + self.assertEqual(result["reason"], "stale_target") + self.assertFalse(result["submitted"]) + self.assertEqual(counter.read_text(encoding="ascii"), "0") + code, result = self.invoke("send", "--to", name, "--message=owned positive control", + "--no-from", "--no-reply-to", "--target-generation", successor) + self.assertEqual(code, 0) + self.assertEqual(result["target"]["pid"], second.pid) + self.assertEqual(result["targetGeneration"], successor) + self.assertEqual(counter.read_text(encoding="ascii"), "1") + + def test_private_deadline_refuses_before_effect_and_bounds_write_drain(self): + conn = mock.Mock() + with mock.patch.object(peer, "IS_WINDOWS", False), \ + mock.patch.object(peer.socket, "AF_UNIX", 1, create=True), \ + mock.patch.object(peer.socket, "socket", return_value=conn), \ + mock.patch.object(peer.time, "monotonic", side_effect=[0, 0, 10]): + with self.assertRaises(peer.CcPeerError) as caught: + peer.post_to_socket("fixture", "hello", effect_deadline=10, total_deadline=15) + self.assertEqual(caught.exception.details["reason"], "effect_deadline_exhausted") + conn.sendall.assert_not_called() + conn.close.assert_called_once() + conn = mock.Mock() + with mock.patch.object(peer, "IS_WINDOWS", False), \ + mock.patch.object(peer.socket, "AF_UNIX", 1, create=True), \ + mock.patch.object(peer.socket, "socket", return_value=conn), \ + mock.patch.object(peer.time, "monotonic", side_effect=[0, 1, 2, 9]): + peer.post_to_socket("fixture", "hello", effect_deadline=10, total_deadline=15) + self.assertEqual([call.args[0] for call in conn.settimeout.call_args_list], [9, 8, 1]) + conn.sendall.assert_called_once() + + def test_private_deadline_post_write_error_unknown_and_windows_unsupported(self): + conn = mock.Mock() + conn.sendall.side_effect = OSError("synthetic partial write") + with mock.patch.object(peer, "IS_WINDOWS", False), \ + mock.patch.object(peer.socket, "AF_UNIX", 1, create=True), \ + mock.patch.object(peer.socket, "socket", return_value=conn), \ + mock.patch.object(peer.time, "monotonic", return_value=0): + with self.assertRaises(peer.CcPeerError) as caught: + peer.post_to_socket("fixture", "hello", effect_deadline=10, total_deadline=15) + self.assertEqual(caught.exception.details["status"], "unknown") + self.assertNotIn("submitted", caught.exception.details) + self.assertFalse(caught.exception.details["retryAllowed"]) + with mock.patch.object(peer, "IS_WINDOWS", True), \ + mock.patch.object(peer, "_post_to_pipe") as pipe: + with self.assertRaises(peer.CcPeerError) as caught: + peer.post_to_socket(r"\\.\pipe\fixture", "hello", effect_deadline=10, total_deadline=15) + self.assertEqual(caught.exception.details["reason"], "unsupported_bounded_inbox") + pipe.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/core/test_target_generation_ssh.py b/tests/core/test_target_generation_ssh.py new file mode 100644 index 00000000..9d1b0554 --- /dev/null +++ b/tests/core/test_target_generation_ssh.py @@ -0,0 +1,160 @@ +"""Actual streamed Python source against an owned restarted protocol inbox.""" +import contextlib +import io +import json +import os +from pathlib import Path +import selectors +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +import session_peer as peer + + +@unittest.skipUnless(sys.platform in ("darwin", "linux"), "owned Unix peer PID") +class SourceStreamedGeneration(unittest.TestCase): + def test_restart_between_source_streamed_list_and_send_never_retargets(self): + with tempfile.TemporaryDirectory(prefix="codex-generation-wire-") as directory: + root = Path(directory) + root.chmod(0o700) + sessions = root / "sessions" + sessions.mkdir(mode=0o700) + endpoint, counter = root / "owned.sock", root / "writes.txt" + attempts = root / "attempts.txt" + attempts.write_text("0", encoding="ascii") + name = "owned-streamed-inbox" + fixtures = Path(__file__).parents[1] / "fixtures" + processes = [] + + def stop(process): + if process.poll() is None: + process.terminate() + try: + process.communicate(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.communicate(timeout=5) + + def spawn(): + process = subprocess.Popen([sys.executable, str(fixtures / "claude_generation_inbox.py"), + str(sessions), str(endpoint), str(counter), name], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True) + processes.append(process) + with selectors.DefaultSelector() as selector: + selector.register(process.stdout, selectors.EVENT_READ) + self.assertTrue(selector.select(5)) + ready = os.read(process.stdout.fileno(), 512) + self.assertEqual(json.loads(ready)["pid"], process.pid) + return process + + env = {"PATH": "/usr/bin:/bin", "HOME": str(root), "LANG": "C.UTF-8", + "CLAUDE_CONFIG_DIR": str(root), "SESSION_PEER_TEST_SSH_COUNTER": str(attempts)} + native_run = subprocess.run + def fake_ssh(command, *args, **kwargs): + if command[0] == "ssh": + return native_run([sys.executable, str(fixtures / "source_streamed_fake_ssh.py"), + *command[1:]], *args, **{**kwargs, "env": env, "cwd": root}) + return native_run(command, *args, **kwargs) + + def invoke(*arguments): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = peer.main([*arguments, "--host", "owned-fixture", "--json", "--no-update-notice"]) + return code, json.loads(output.getvalue()) + + try: + with mock.patch.object(peer.subprocess, "run", side_effect=fake_ssh), \ + mock.patch.object(peer, "tailscale_status", return_value={}), \ + mock.patch.object(peer, "resolve_ssh_destination", return_value="owned-fixture"), \ + mock.patch.object(peer, "tailscale_ssh_options", return_value=[]): + first = spawn() + code, listing = invoke("list", "--agent", "claude", "--with-target-generation") + self.assertEqual(code, 0, listing) + original = listing["sessions"][0]["targetGeneration"] + self.assertEqual(counter.read_text(encoding="ascii"), "0") + stop(first) + endpoint.unlink() + successor = spawn() + for mode in ([], ["--dry-run"]): + before = int(attempts.read_text(encoding="ascii")) + code, result = invoke("send", "--to", name, "--message=must not arrive", + "--no-from", "--no-reply-to", "--target-generation", original, *mode) + self.assertEqual(code, 1, result) + self.assertEqual(result["reason"], "stale_target") + self.assertIs(result["submitted"], False) + self.assertFalse(result["retryAllowed"]) + self.assertEqual(result["lastSeenTarget"], {"agent": "claude", "targetGeneration": original}) + self.assertEqual(counter.read_text(encoding="ascii"), "0") + self.assertEqual(int(attempts.read_text(encoding="ascii")), before + 1) + _, listing = invoke("list", "--agent", "claude", "--with-target-generation") + observed = listing["sessions"][0]["targetGeneration"] + self.assertNotEqual(original, observed) + before = int(attempts.read_text(encoding="ascii")) + code, result = invoke("send", "--to", name, "--message=owned positive control", + "--no-from", "--no-reply-to", "--target-generation", observed) + self.assertEqual(code, 0, result) + self.assertEqual(result["target"]["pid"], successor.pid) + self.assertEqual(result["targetGeneration"], observed) + self.assertNotIn("lastSeenTarget", result) + self.assertNotIn("submitted", result) + self.assertEqual(counter.read_text(encoding="ascii"), "1") + self.assertEqual(int(attempts.read_text(encoding="ascii")), before + 1) + finally: + for process in processes: + stop(process) + + +class LastSeenWire(unittest.TestCase): + def test_only_valid_original_stale_generation_adds_metadata(self): + token = "tg1:" + "a" * 64 + expected = {"agent": "claude", "targetGeneration": token} + self.assertEqual(peer.generation_refused("stale_target", "fixture", + expected_generation=token).details["lastSeenTarget"], expected) + for reason, original in (("target_generation_unavailable", token), + ("unsupported_target_generation", token), + ("invalid_target_generation", token), + ("stale_target", None), ("stale_target", "not a token")): + with self.subTest(reason=reason, original=original): + self.assertNotIn("lastSeenTarget", peer.generation_refused(reason, "fixture", + expected_generation=original).details) + + def test_closed_original_metadata_only_proves_requested_stale_refusal(self): + token = "tg1:" + "a" * 64 + argv = ["send", "--to", "owned", "--target-generation", token] + response = {"schemaVersion": 1, "command": "send", "ok": False, + "status": "refused", "reason": "stale_target", "submitted": False, "retryAllowed": False, + "lastSeenTarget": {"agent": "claude", "targetGeneration": token}} + self.assertTrue(peer.parse_ssh_response(json.dumps(response), argv)[3]) + self.assertTrue(peer.parse_ssh_response(json.dumps(response), + ["send", "--to=owned", "--target-generation=" + token])[3]) + for mismatched in (["list", "--to", "owned", "--target-generation", token], + ["send", "--to", "codex:00000000-0000-4000-8000-000000000000", "--target-generation", token], + ["send", "--to", "antigravity:owned", "--target-generation", token], + ["send", "--target-generation", token], + ["send", "--to", "owned", "--to", "other", "--target-generation", token]): + with self.subTest(argv=mismatched): + candidate = {**response, "command": mismatched[0]} + self.assertFalse(peer.parse_ssh_response(json.dumps(candidate), mismatched)[3]) + for metadata in ({"agent": "codex", "targetGeneration": token}, + {"agent": "claude", "targetGeneration": "tg1:" + "b" * 64}, + {"agent": "claude", "targetGeneration": token, "path": "CORRUPTION_SENTINEL"}, + {"agent": "claude", "targetGeneration": True}): + with self.subTest(metadata=metadata): + altered = {**response, "lastSeenTarget": metadata} + self.assertFalse(peer.parse_ssh_response(json.dumps(altered), argv)[3]) + self.assertFalse(peer.parse_ssh_response(json.dumps(altered), ["send"])[3]) + for edits in ({"reason": "unsupported_target_generation"}, {"retryAllowed": True}, {"submitted": None}): + self.assertFalse(peer.parse_ssh_response(json.dumps({**response, **edits}), argv)[3]) + + def test_invalid_optional_metadata_preserves_existing_positive_parser_semantics(self): + native = {"schemaVersion": 1, "command": "send", "ok": True, + "target": {"pid": 4242, "name": "owned"}, "chars": 1, "dryRun": False, + "lastSeenTarget": {"secret": "CORRUPTION_SENTINEL"}} + _, result, _, valid = peer.parse_ssh_response(json.dumps(native), ["send"]) + self.assertTrue(valid) + self.assertEqual(result["target"], native["target"]) + self.assertNotIn("lastSeenTarget", result) + self.assertNotIn("submitted", result) diff --git a/tests/fixtures/claude_generation_inbox.py b/tests/fixtures/claude_generation_inbox.py new file mode 100644 index 00000000..49f61b88 --- /dev/null +++ b/tests/fixtures/claude_generation_inbox.py @@ -0,0 +1,33 @@ +"""Owned protocol inbox only: record metadata and a body-free write counter.""" +import json +import os +from pathlib import Path +import socket +import sys + + +directory, endpoint, counter, name = map(Path, sys.argv[1:]) +pid = os.getpid() +with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server: + server.bind(str(endpoint)) + server.listen(4) + counter.write_text("0", encoding="ascii") + (directory / (str(pid) + ".json")).write_text(json.dumps({ + "pid": pid, "name": str(name), "status": "idle", "cwd": str(directory), + "messagingSocketPath": str(endpoint), "startedAt": 1000}), encoding="utf-8") + print(json.dumps({"pid": pid}), flush=True) + writes = 0 + while True: + connection, _ = server.accept() + with connection: + connection.settimeout(2) + raw = bytearray() + while len(raw) < 65536: + chunk = connection.recv(min(4096, 65536 - len(raw))) + if not chunk: + break + raw.extend(chunk) + if raw: + writes += 1 + counter.write_text(str(writes), encoding="ascii") + connection.sendall(b"1") diff --git a/tests/fixtures/source_streamed_fake_ssh.py b/tests/fixtures/source_streamed_fake_ssh.py new file mode 100644 index 00000000..62bcb17b --- /dev/null +++ b/tests/fixtures/source_streamed_fake_ssh.py @@ -0,0 +1,31 @@ +"""Owned fake SSH: execute the actual streamed source, never open a network. + +This tests Python/source/argv/wire integration, not SSH authentication or a +physical remote host. Its inherited HOME/config are private fixture paths. +""" +import os +from pathlib import Path +import shlex +import subprocess +import sys + + +if sys.argv[1:2] == ["-G"]: + print("user fixture-user\nhostname owned-fixture\nport 22") + raise SystemExit(0) + +if len(sys.argv) != 3 or sys.argv[1] != "owned-fixture": + raise SystemExit(2) +command = shlex.split(sys.argv[2]) +if command[:2] != ["python3", "-"]: + raise SystemExit(2) +source = sys.stdin.buffer.read(4 * 1024 * 1024 + 1) +if not source or len(source) > 4 * 1024 * 1024: + raise SystemExit(2) +counter = Path(os.environ["SESSION_PEER_TEST_SSH_COUNTER"]) +counter.write_text(str(int(counter.read_text(encoding="ascii")) + 1), encoding="ascii") +completed = subprocess.run([sys.executable, "-I", "-", *command[2:]], input=source, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15) +sys.stdout.buffer.write(completed.stdout) +sys.stderr.buffer.write(completed.stderr) +raise SystemExit(completed.returncode) diff --git a/tools/generate_session_peer.py b/tools/generate_session_peer.py index d2ee612f..8dcadc0a 100644 --- a/tools/generate_session_peer.py +++ b/tools/generate_session_peer.py @@ -17,6 +17,7 @@ "common.py", "codex.py", "claude.py", + "target_generation.py", "replies.py", "ssh.py", "output.py",