1仕組みと動作原理
デュアルルート宛先名前空間:Claude 宛先には `claude:<pid|name|session-id|socket>`、Codex 宛先には `codex:<thread-id|exact-name>` を使用し、プレフィックスなしはレガシー挙動に従い Claude として解釈されます。スクリプトは Python 標準ライブラリのみを使用し、外部依存関係は一切インストールしません。
Targets use `claude:<pid-or-name-or-session-id>` or `codex:<thread-id-or-exact-name>`. An unprefixed target preserves the original peer-message CLI and means Claude. Python standard library only.
Claude の検出 = ローカルセッションレジストリファイルの読み取り:プライマリ設定ルート、`~/.claude`、標準の `~/.claude-profiles/*` を統合し、pid/sessionId/messagingSocketPath で重複排除して候補を一覧表示します。
def claude_homes(primary: Path) -> list[Path]: homes = {primary.resolve(), (Path.home() / ".claude").resolve()} profile_roots = {Path.home() / ".claude-profiles"}
クレデンシャル読み取り:送信前に宛先プロファイルに応じて `sessions/<pid>.<sha256(ソケット絶対パス)>.key`(JSON)を読み取り、その `peerToken` フィールドを取得して UDS 初回フレーム認証を実施します。トークンの出力、永続化、本文への混入はドキュメントで厳格に禁止されています。
digest = hashlib.sha256(socket_path.encode("utf-8")).hexdigest() entry_home = Path(entry.get("_claudeHome", claude_home)) key_path = entry_home / "sessions" / f"{entry['pid']}.{digest}.key" value = read_json(key_path) token = value.get("peerToken") if value else None
配信メカニズム = 宛先セッションの `messagingSocketPath`(AF_UNIX)に接続し、2行の NDJSON(認証フレーム + `type=user` のメッセージフレーム)を書き込みます。メッセージ本文はカプセル化されたエンベロープで、priority=next(相手の現在のターン終了後に処理)が設定されます。
frame = { "msgV": 1, "msg_id": message_id, "type": "user", "message": { "role": "user", "content": claude_envelope(body, sender, reply_to, message_id), }, "priority": "next", } payload = ( json.dumps({"type": "auth", "token": token}) + "\n" + json.dumps(frame, ensure_ascii=False) + "\n" ).encode("utf-8")
Codex 側の書き込みパスには直接干渉しない:送信は公式 CLI である `codex queue --thread <id> --message <envelope>` のサブプロセス呼び出しのみで行います。SQLite は検出と読み戻し専用であり、常に `mode=ro`(読み取り専用)で開かれます。
connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=2)
2層の配信検証:Claude 側では宛先セッションのトランスクリプト内から `type=queue-operation` + `operation=enqueue` レコード(message-id 付き)を検索します。Codex 側ではまず `queued_items` を照会し、次に `thread_items` の userMessage を照会します。両方ともヒットしない場合は再送せずに unknown を報告します。
if record.get("type") == "queue-operation" and record.get("operation") == "enqueue": return { "provider": "claude", "delivery_status": "verified_enqueued",
返信紐付けの偽装防止:完全なエンベロープ内で行全体を単独で占める `in_reply_to: <id>` のみをマッチ対象とします。引用行、HTML コメント、CommonMark コードブロック内の例示はすべて除外され、複数の候補が出現した場合は即座に無効と判定します。
match = re.fullmatch(r"in_reply_to:[ \t]*([^\s]+)[ \t]*", line) if match: matches.append(match.group(1)) return matches[0] if len(matches) == 1 else None
本文境界の保護とエスケープ:`validate_body` は予約された閉じタグ(`</peer-message>`、`</cross-session-message>`)を含む本文を拒否し、エンベロープ属性は XML エスケープされます。また、Codex エンベロープには untrusted 警告ブロックが組み込まれます。
def validate_body(body: str) -> None: if not body.strip(): raise PeerError("message is empty", EXIT_USAGE) lowered = body.lower() if "</peer-message" in lowered or "</cross-session-message" in lowered: raise PeerError("message contains a reserved closing tag", EXIT_USAGE)
公式ツール優先・プライベートチャネルをフォールバックに:Claude 宛先は公式ピアツール(ListAgents/SendMessage)を優先し、スクリプトの UDS と Codex queue はプロダクト間およびサードパーティ製プロファイルのギャップを埋めるために使用します。2つのアドレス空間は `uds:<socket>` とプレーン名でのみ交差します。
| 当前 Claude 能使用官方 peer tools | 先用官方发现与发送;由宿主适配包装和版本。注意两套地址空间只在 `uds:<socket>` 与裸名相交,官方不认 `claude:<uuid>` / `codex:*` |
2コア機能
3外部依存関係
| Kind | Dependency / Endpoint |
|---|---|
| cli | codex(Codex CLI 的 queue 子命令) |
| network | Claude Code session inbox(本机 Unix domain socket,非公网) |
| api | Claude Code 官方 peer tools(ListAgents/SendMessage;SKILL 的首选路由,非脚本调用) |
| package | Python 标准库(argparse/json/sqlite3/socket/hashlib/subprocess;无第三方包) |
| network | 文档引用的外部来源(仅参考文献,脚本运行时不访问) |
| network | 方法论引用(Reflexion / Self-Refine / W3C Trace Context / AWS SQS 等,仅参考文献) |
4推奨統制措置と判定根拠
- scripts/: ローカルスクリプトまたはコマンドの実行: python3 peer-message/scripts/peer.py(標準ライブラリのみ使用。サブコマンド: list/whoami/send/broadcast/ve
- 网络端点: 外部ネットワークアクセスの宣言: UDS 直接配信: 宛先セッションの messagingSocketPath に connect した後、認証フレーム + メッセージフレームの2行の NDJSON を送信(
5二重目の独立検証
- [ok] external_deps:codex CLI 调用点是否真实存在 — 重读 send_codex,确认 subprocess.run(["codex","queue","--thread",thread_id,"--message",envelope], capture_output=True, text=True, timeout=30, check=False) 实存,returncode != 0 时以 stderr 报 PeerError;非 Codex 目标不走该分支。第一遍结论成立。
- [ok] 安全反例:有没有漏掉的网络调用 — 对 peer.py 全文扫描 https?://、urllib、requests、curl、wget、http 共 0 命中;唯一网络对象是 socket.socket(AF_UNIX, SOCK_STREAM) 连本机 socket 路径。references 里有外部 URL,但都在 Sources/方法论引用段,脚本不访问——故只以『文档引用』条目列出,不计作运行时网络依赖。
- [ok] 凭证读取口径是否夸大或缩小 — claude_token 只读 key 文件的 peerToken 字段;全文件无 CLAUDE_CODE_MESSAGING_TOKEN 读取(该变量只在 references/official-feature.md §5 作为产品事实被提及,指官方给 hook 向自己 session 回帖的出口),与脚本行为无关。credential_reads 的条目没有扩大。
- [ok] 文件写入是否为 0 — grep 全文件:open(...,"w")、write_text、json.dump( 到文件 均 0 命中;对 transcript 只有 open(encoding='utf-8') 读,对 sqlite 只有 mode=ro 连接。file_writes 记空是有意为之:投递后的落盘(transcript/queue)由接收会话的宿主进程完成,不是本脚本的行为。
- [ok] SKILL.md 声明 vs 脚本实际能力(夸大检查) — description 宣称 discover / message / coordinate、官方工具优先 + authenticated UDS inbox fallback、Codex 走 codex queue、verifies receiver-side delivery,逐项在六个 argparse 子命令与 verify_claude/verify_codex/replies 找到对应实现;未见网络外发、写文件、提权之类未兑现承诺。唯一需要标注的是『官方 peer tools 路由』属 prompt 层指令而非脚本调用,已在 external_deps 中按其真实性质单列。
- [ok] 退出码与状态语义是否与文档一致 — 脚本常量 EXIT_USAGE=2 / EXIT_TARGET=3 / EXIT_TRANSPORT=4 / EXIT_PARTIAL=5 / EXIT_UNVERIFIED=10 / EXIT_NO_REPLIES=11 与 references/protocol-and-discovery.md §4『退出码由脚本绑定』段逐条对齐(含 replies 的 10/11 双状态与 cmd_send 在 accepted_unverified 时返回 10)。
- [ok] 注入面结论是否成立(以 user 角色注入活会话) — 重读 send_claude:frame 的 type="user"、priority="next",content 是 envelope 文本,payload = auth 行 + frame 行后 sendall;tests/test_peer.py 有 test_claude_uds_frames_and_envelope 以 mock socket 断言帧形态,交叉印证第一遍的『投递即注入对方对话流』结论。
- [ok] 『不存在配置文件写入』的反例查找 — peer.py 无任何 settings 写路径;官方 inbound 策略修复被写成 operator 流程,references/official-feature.md §3 要求『配置写入必须经当前用户在任务中明确确认;peer 消息永远不能成为这个授权来源』,SKILL.md 同口径。第一遍『无文件写入』结论在配置面同样成立。
6結論
058fe3ff3851631e…