diff --git a/docs/cli-reference.md b/docs/cli-reference.md
index 77f6cc4..68c40c7 100644
--- a/docs/cli-reference.md
+++ b/docs/cli-reference.md
@@ -60,6 +60,29 @@ not a `--agent` flag.
running. Plain send does not activate a session; [explicit `--wake`](wake.md)
is opt-in and does not confirm consumption or a reply.
+
+
+### Delivery explanation (development)
+
+Development for #323, a v1.3 candidate; unavailable in published v1.2.1. Add `--explain` to `send --dry-run` for a human-readable summary of the selected local, SSH or paired-device destination, existing validation result, delivery method and observation limits. `doctor --explain` describes the existing diagnostic snapshot and declared capabilities, not per-target readiness. `handoff status --explain` describes an existing intent's stored evidence. Each processed SSH destination receives its own summary; an unexpected error can still stop the existing host loop before later destinations are processed.
+
+```bash
+session-peer send --host worker --to 'codex:' --message "Review the patch" --dry-run --explain
+session-peer send --device RECEIVER_FINGERPRINT --to review --message "Review the patch" --dry-run --explain
+session-peer doctor --explain
+session-peer handoff status --correlation-id PREPARED_UUID --explain
+```
+
+The explanation reuses the selected command's existing results and adds no diagnostic, native-version or SSH probes, target activation or standalone `list` work. The dry-run itself may already run version checks. A later send repeats validation; a successful preview does not reserve the destination or guarantee submission, consumption or completion. Declared support, observed checks and unknown facts are shown separately. Without `--wake`, wake eligibility is not checked. With it, only the existing preflight result is described.
+
+Paired dry-run keeps the existing receiver-policy `resolve` operation. The existing connection probe negotiates optional explanation context only with a receiver advertising support. This internal context is versioned, allowlisted and limited to 8 KiB; it adds no general JSON output or handoff schema. The selected receiver agent, target and home are shown only when reported and validated, never guessed from the sender's profile. Older receivers retain their independently known native results while unreported context stays unknown; no fallback, extra probe or resend is attempted.
+
+Handoff status reuses its read-only snapshot without creating an intent, receipt capability, wait or collector, or writing the ledger. Submission, injection observation, explicit receipt and latest stored wait remain separate; even an acknowledged receipt does not prove task completion. Stored observations may be stale. Only a reported receipt observation time is described as such; a description's generation time or diagnostic snapshot is not a delivery observation time. Ordinary sends do not gain receipt ACK support.
+
+This scope is text-only: combining `--explain` with `--json` or `--output-format json` is a usage error. Only `send --explain` requires `--dry-run`; `doctor --explain` and `handoff status --explain` do not. Send still refuses experimental handoff `--correlation-id`, `--request-ack`, `--observe-delivery` and `--wait-for` with `--explain`, before message input or destination/ledger work. Existing commands without the new option keep their output, JSON and exit contracts; explanation does not turn a successful status query into proof of ACK or change the underlying command's exit meaning.
+
+For send, the normal message requirement and limits still apply: supply `--message`, `--message-file` or stdin as usual; no placeholder body is generated. Explanation fields do not include the body or receipt secrets. Normal explanations use stdout; failed results or unexpected submission evidence use stderr, without moving existing summary/error output. Unknown evidence stays unknown, and an uncertain actual send must never trigger automatic resend.
+
### Optional paired devices and Antigravity (v0.9)
These features are available starting with PyPI v0.9.0. Antigravity remains
diff --git a/docs/i18n-manifest.json b/docs/i18n-manifest.json
index 55eb227..64cd6d8 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": "d2021243cef2dd29a48ee5d9378ded8b6e3a19e3042c3fa13899a463f700d83c",
+ "docs/cli-reference.md": "90eeebe74625a9b10ed56ca543dd43ed7bffb0f5f4bc4c383b880162de3638f3",
"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 9e4e21a..016f6ee 100644
--- a/docs/ja/cli-reference.md
+++ b/docs/ja/cli-reference.md
@@ -93,6 +93,29 @@ session-peer send --host worker --to 'codex:' --dry-run -m "me
限りません。通常の送信ではセッションはアクティブ化されません。[明示的な `--wake`](wake.md)
はオプトインであり、消費または返信を確認するものではありません。
+
+
+### 配信説明（開発中）
+
+#323 の開発機能で、v1.3 の候補です。公開済み v1.2.1 には含まれません。`send --dry-run` に `--explain` を追加すると、選択したローカル・SSH・ペアリング済みデバイスの宛先、既存の検証結果、配信方法、観測上の制限を読みやすく表示します。`doctor --explain` は既存の診断スナップショットと宣言された対応機能を説明するもので、個々の宛先の送信準備状態を確認するものではありません。`handoff status --explain` は既存の intent に保存された証拠を説明します。処理した SSH の宛先ごとに説明を表示しますが、予期しないエラーが発生すると既存のホスト処理が中断され、後続の宛先が処理されない場合があります。
+
+```bash
+session-peer send --host worker --to 'codex:' --message "Review the patch" --dry-run --explain
+session-peer send --device RECEIVER_FINGERPRINT --to review --message "Review the patch" --dry-run --explain
+session-peer doctor --explain
+session-peer handoff status --correlation-id PREPARED_UUID --explain
+```
+
+説明は選択したコマンドの既存の結果を再利用し、診断、native バージョン、SSH の追加照会、宛先の起動、別途の `list` 処理は行いません。既存の dry-run 自体にはバージョン確認が含まれる場合があります。実際の送信時には再検証するため、プレビューの成功は宛先の予約や投入・消費・完了を保証しません。宣言された対応機能、実際の検証結果、不明な事実を区別して表示します。`--wake` を指定しない場合、起動の適格性は確認しません。指定した場合も既存の事前検証結果だけを説明します。
+
+ペアリングの dry-run は、受信側ポリシーに従う既存の `resolve` 処理を維持します。既存の接続確認で対応を通知した受信側とのみ、任意の説明 context をネゴシエートします。この内部 context にはバージョンが付けられ、許可されたフィールドのみが認められます。サイズの上限は 8 KiB です。一般の JSON 出力や handoff スキーマは追加しません。選択された受信側のエージェント・宛先・home は、報告され検証された場合だけ表示し、送信側のプロファイルから推測しません。旧バージョンの受信側でも、独立に確認された native の結果は維持し、報告されない context は不明のまま扱います。代替経路への切り替え、追加照会、再送は行いません。
+
+Handoff status は読み取り専用スナップショットを再利用し、intent・receipt capability・wait・collector の作成や台帳への書き込みは行いません。投入、入力注入の観測、明示的な receipt、最後に保存された wait は別々に扱います。Receipt が acknowledged でも、タスクの完了を証明するものではありません。保存された観測は古い場合があります。報告された receipt の観測時刻だけを観測時刻として説明し、説明の生成時刻や診断スナップショットの時刻を配信の観測時刻として扱いません。通常送信に receipt ACK の対応が追加されるわけではありません。
+
+この範囲はテキスト出力のみで、`--explain` と `--json` または `--output-format json` の併用は使用法エラーになります。`--dry-run` が必要なのは `send --explain` だけで、`doctor --explain` と `handoff status --explain` には不要です。Send では引き続き、実験的 handoff の `--correlation-id`、`--request-ack`、`--observe-delivery`、`--wait-for` と `--explain` の組み合わせを、メッセージ入力や宛先・台帳処理の前に拒否します。新しいオプションを使わない既存コマンドの出力・JSON・終了契約は維持されます。説明によって、状態照会の成功が ACK の証拠に変わったり、元のコマンドの終了結果の意味が変わったりすることはありません。
+
+Send の通常のメッセージ入力要件と上限は変わりません。従来どおり `--message`、`--message-file` または stdin を使い、仮の本文は自動生成しません。説明フィールドに本文や receipt の秘密値は含めません。通常の説明は stdout に、失敗した結果や予期しない投入の証拠に関する説明は stderr に出力し、既存の要約・エラーの出力先は変えません。不明な証拠は不明のまま表示し、実際の送信結果が不確実な場合は自動再送しません。
+
### オプションのペアリング済みデバイスと Antigravity (v0.9)
これらの機能は PyPI v0.9.0 以降で利用可能です。Antigravity は引き続き
diff --git a/docs/ko/cli-reference.md b/docs/ko/cli-reference.md
index 0ab1b74..07f2344 100644
--- a/docs/ko/cli-reference.md
+++ b/docs/ko/cli-reference.md
@@ -94,6 +94,29 @@ session-peer send --host worker --to 'codex:' --dry-run -m "me
실행 중인 것은 아닙니다. 일반 send는 세션을 활성화하지 않으며, [명시적 `--wake`](wake.md)는
옵트인 방식이며 소비나 응답을 확인해주지 않습니다.
+
+
+### 전달 설명 (개발 중)
+
+#323의 개발 기능으로 v1.3 후보이며, 공개 v1.2.1에는 없습니다. `send --dry-run`에 `--explain`을 추가하면 선택한 로컬·SSH·페어링 기기 대상, 기존 검증 결과, 전달 방식과 관찰 한계를 사람이 읽기 쉬운 형태로 표시합니다. `doctor --explain`은 기존 진단 스냅샷과 선언된 지원 기능을 설명하며, 특정 대상의 전송 준비 상태를 확인하지는 않습니다. `handoff status --explain`은 기존 intent에 저장된 증거를 설명합니다. 처리한 SSH 대상마다 개별 설명을 표시하지만, 예상치 못한 오류가 나면 기존 호스트 반복 처리가 중단되어 이후 대상은 처리하지 않을 수 있습니다.
+
+```bash
+session-peer send --host worker --to 'codex:' --message "Review the patch" --dry-run --explain
+session-peer send --device RECEIVER_FINGERPRINT --to review --message "Review the patch" --dry-run --explain
+session-peer doctor --explain
+session-peer handoff status --correlation-id PREPARED_UUID --explain
+```
+
+설명은 선택한 명령의 기존 결과를 재사용하며 진단·native 버전·SSH 조회, 대상 활성화나 별도의 `list` 작업을 추가하지 않습니다. 기존 dry-run 자체에는 버전 검사가 포함될 수 있습니다. 실제 전송 시에는 다시 검증하며, 미리보기 성공이 대상을 예약하거나 제출·소비·완료를 보장하지는 않습니다. 선언된 지원 기능, 실제 검사 결과와 확인할 수 없는 사실을 구분해 표시합니다. `--wake`를 지정하지 않으면 깨우기 가능 여부를 검사하지 않습니다. 지정한 경우에도 기존 사전 검사 결과만 설명합니다.
+
+페어링 dry-run은 기존 수신 기기 정책에 따른 `resolve` 작업을 유지합니다. 기존 연결 확인 과정에서 지원을 알리는 수신 기기와만 선택적 설명 context를 협상합니다. 이 내부 context에는 버전과 허용 필드 검증을 적용하며 크기를 8 KiB로 제한합니다. 일반 JSON 출력이나 handoff 스키마는 추가하지 않습니다. 선택한 수신 측 에이전트·대상·home은 보고되고 검증된 경우에만 표시하며 송신 측 프로필에서 추측하지 않습니다. 구형 수신 기기의 독립적으로 확인된 native 결과는 유지하고, 보고되지 않은 context는 알 수 없음으로 표시합니다. 대체 경로 시도, 추가 조회나 재전송은 하지 않습니다.
+
+Handoff status는 읽기 전용 스냅샷을 재사용하며 intent·receipt capability·wait·collector를 만들거나 ledger에 기록하지 않습니다. 제출, 입력 주입 관찰, 명시적 receipt와 최근 저장된 wait는 구분합니다. Receipt가 acknowledged여도 작업 완료를 증명하지는 않습니다. 저장된 관찰은 오래된 정보일 수 있습니다. 보고된 receipt 관찰 시각만 관찰 시각으로 설명하며, 설명 생성 시각이나 진단 스냅샷 시각을 전달 관찰 시각으로 해석하지 않습니다. 일반 전송에 receipt ACK 지원이 추가되지는 않습니다.
+
+이 범위는 텍스트 출력 전용이므로 `--explain`을 `--json` 또는 `--output-format json`과 함께 지정하면 사용법 오류가 발생합니다. `--dry-run`은 `send --explain`에만 필수이며 `doctor --explain`과 `handoff status --explain`에는 필요하지 않습니다. Send에서는 실험적 handoff 옵션인 `--correlation-id`, `--request-ack`, `--observe-delivery`, `--wait-for`와 `--explain`의 조합을 메시지 입력이나 대상·ledger 작업 전에 계속 거부합니다. 새 옵션을 사용하지 않은 기존 명령의 출력·JSON·종료 계약은 유지됩니다. 설명은 성공한 상태 조회를 ACK 증거로 바꾸거나 원래 명령의 종료 의미를 바꾸지 않습니다.
+
+Send의 메시지 입력 요구사항과 한도는 동일합니다. 평소처럼 `--message`, `--message-file` 또는 stdin을 사용하며 임시 본문을 자동 생성하지 않습니다. 설명 필드에는 본문이나 receipt 비밀값을 넣지 않습니다. 일반 설명은 stdout에, 실패 결과나 예상 밖 제출 증거의 설명은 stderr에 표시하며 기존 요약·오류 출력의 위치는 바꾸지 않습니다. 확인할 수 없는 증거는 그대로 알 수 없음으로 표시하고, 실제 전송 결과가 불확실하면 자동 재전송하지 않습니다.
+
### 페어링된 기기 옵션 및 Antigravity (v0.9)
이 기능들은 PyPI v0.9.0부터 제공됩니다. Antigravity는 여전히 실험적이며 페어링된 전송은
diff --git a/docs/zh-CN/cli-reference.md b/docs/zh-CN/cli-reference.md
index 52d7a6e..0c19321 100644
--- a/docs/zh-CN/cli-reference.md
+++ b/docs/zh-CN/cli-reference.md
@@ -72,6 +72,29 @@ session-peer send --host worker --to 'codex:' --dry-run -m "me
**已投递/已入队并不代表已确认。** 已保存的 Codex 线程不一定正在运行。普通的发送不会激活会话；[显式 `--wake`](wake.md) 是可选功能，且不确认消息已被消费或回复。
+
+
+### 传递说明（开发中）
+
+这是 #323 的开发功能，拟纳入 v1.3，尚未包含在已发布的 v1.2.1 中。在 `send --dry-run` 中添加 `--explain`，即可查看所选本地、SSH 或配对设备目标、现有验证结果、传递方式及观察限制。`doctor --explain` 说明现有诊断快照和声明的支持能力，不验证特定目标是否已准备好接收提交。`handoff status --explain` 说明现有 intent 中保存的证据。每个已处理的 SSH 目标都会单独显示说明；意外错误仍可能中断现有的主机循环，导致后续目标未被处理。
+
+```bash
+session-peer send --host worker --to 'codex:' --message "Review the patch" --dry-run --explain
+session-peer send --device RECEIVER_FINGERPRINT --to review --message "Review the patch" --dry-run --explain
+session-peer doctor --explain
+session-peer handoff status --correlation-id PREPARED_UUID --explain
+```
+
+说明复用所选命令的现有结果，不增加诊断、native 版本或 SSH 探测，也不激活目标或另行执行 `list`。原有 dry-run 本身可能包含版本检查。实际发送时会重新验证，因此预览成功不代表预留目标，也不保证提交、消费或完成。声明的支持能力、实际检查结果和未知事实会分别显示。未指定 `--wake` 时，不检查唤醒资格；指定后也只说明原有预检结果。
+
+配对 dry-run 保留现有的、遵循接收端策略的 `resolve` 操作。现有连接探测只会与声明支持该功能的接收端协商可选的说明 context。此内部 context 带有版本，严格校验允许的字段，大小限制为 8 KiB；不会新增通用 JSON 输出，也不引入新的 handoff schema。只有接收端实际报告且通过验证的代理、目标和 home 才会显示，绝不根据发送端配置猜测。旧接收端仍保留独立确认的 native 结果，未报告的 context 保持未知，不尝试回退、额外探测或重发。
+
+Handoff status 复用只读快照，不创建 intent、receipt capability、wait 或 collector，也不写入账本。提交、输入注入观察、显式 receipt 和最近保存的 wait 分别显示；即使 receipt 已 acknowledged，也不证明任务完成。保存的观察可能已经过时。只有已报告的 receipt 观察时间才会作为观察时间说明；说明生成时间或诊断快照时间不代表传递观察时间。普通发送不会因此获得 receipt ACK 支持。
+
+此范围仅支持文本输出：将 `--explain` 与 `--json` 或 `--output-format json` 组合使用会报告用法错误。只有 `send --explain` 必须指定 `--dry-run`，`doctor --explain` 和 `handoff status --explain` 不需要。Send 仍会在读取消息或执行目标、账本操作前，拒绝将实验性 handoff 的 `--correlation-id`、`--request-ack`、`--observe-delivery`、`--wait-for` 与 `--explain` 组合使用。不使用新选项的现有命令保留原有输出、JSON 和退出契约；说明不会将状态查询成功变成 ACK 证据，也不会改变原命令退出结果的含义。
+
+Send 的消息输入要求和限制保持不变。按原有方式使用 `--message`、`--message-file` 或 stdin，不自动生成占位正文。说明字段不包含正文或 receipt 秘密值。常规说明输出到 stdout，失败结果或意外提交证据的说明输出到 stderr，不改变现有摘要和错误的输出位置。未知证据仍显示为未知，实际发送结果不确定时禁止自动重发。
+
### 可选配对设备与 Antigravity (v0.9)
这些功能自 PyPI v0.9.0 起提供。Antigravity 仍处于实验阶段，配对传输仍处于 Beta 阶段；仍需进行操作验证。使用 `pipx install 'session-peer[relay]'` 安装额外组件。`[relay]` 额外组件（Unix、Python 3.11+）支持通过直接连接或自建 WSS 中继进行经过身份验证的配对设备传递。配对会固定设备身份；单独的操作员策略允许特定的代理目标和操作。盲中继无法解密应用消息。在公开端点之前，请遵循[配对设备设置与操作指南](paired-devices.md)。该测试版不提供 NAT 穿透或托管公共服务。
diff --git a/session_peer.py b/session_peer.py
index de9c120..8055e1c 100644
--- a/session_peer.py
+++ b/session_peer.py
@@ -4618,6 +4618,179 @@ def run_remote(host: str, argv: list[str], ssh_opts: list[str], *, handoff_budge
return _run_remote_dispatch(host, argv, ssh_opts)
+# Display-only explanation facts. No discovery, transport, ledger or clock I/O.
+
+
+def validated_delivery_context(value):
+ """Accept the complete private receiver context, never guessed fragments."""
+ if not isinstance(value, dict) or set(value) != {
+ "schemaVersion", "agent", "target", "home", "capabilities"}:
+ return None
+ if type(value["schemaVersion"]) is not int or value["schemaVersion"] != 1:
+ return None
+ if value["agent"] not in ("claude", "codex", "antigravity"):
+ return None
+ for key in ("target", "home"):
+ item = value[key]
+ if item is not None:
+ if (not isinstance(item, str) or not item or item.lower().startswith("session-peer:")
+ or any(ord(c) < 32 or 127 <= ord(c) <= 159 for c in item)):
+ return None
+ try:
+ if len(item.encode("utf-8")) > 4096:
+ return None
+ except UnicodeError:
+ return None
+ if value["target"] is not None:
+ pattern = (r"claude:[1-9][0-9]*" if value["agent"] == "claude" else
+ value["agent"] + r":[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+ if re.fullmatch(pattern, value["target"]) is None:
+ return None
+ if value["home"] is not None and not (PurePosixPath(value["home"]).is_absolute()
+ or PureWindowsPath(value["home"]).is_absolute()):
+ return None
+ caps = value["capabilities"]
+ if not isinstance(caps, dict) or set(caps) != {"send", "wake", "wait", "ack"}:
+ return None
+ if any(type(item) is not bool for item in caps.values()):
+ return None
+ if len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) > 8192:
+ return None
+ return {**value, "capabilities": dict(caps)}
+
+
+def explanation_text(value):
+ """Sanitize only opt-in explanations, preserving ordinary output behavior."""
+ text = human_text(value)
+ return "".join("\\u%04x" % ord(c) if ord(c) in (
+ 0x200b, 0x200c, 0x200d, 0x200e, 0x200f, 0x202a, 0x202b,
+ 0x202c, 0x202d, 0x202e, 0x2060, 0x2066, 0x2067, 0x2068, 0x2069, 0xfeff)
+ else c for c in text)
+
+
+def build_delivery_explanation(args, result, *, mode="readiness", failed=False):
+ """Normalize only allowlisted facts already supplied by the caller.
+
+ Declaration never proves eligibility, and a snapshot never proves delivery.
+ The returned object is internal and is not a new JSON/wire contract.
+ """
+ if mode not in ("readiness", "doctor", "existing_intent"):
+ raise ValueError("unsupported delivery explanation mode")
+ paired = bool(getattr(args, "device", None))
+ context = validated_delivery_context(getattr(args, "_delivery_explanation_context", None)) if paired else None
+ requested = str(getattr(args, "to", "") or "")
+ uri = requested.lower().startswith("session-peer:")
+ agent = (context["agent"] if context else None) if paired else (
+ None if uri or not requested else "codex" if requested.startswith("codex:") else
+ "antigravity" if requested.startswith("antigravity:") else "claude")
+ caps = context["capabilities"] if context else None
+ if not paired and agent is not None:
+ caps = {key: getattr(AGENTS.get(agent).capabilities, key)
+ for key in ("send", "wake", "wait", "ack")}
+ home = context["home"] if context else None
+ if not paired and agent in ("codex", "antigravity"):
+ value = result.get("codexHome" if agent == "codex" else "antigravityHome")
+ home = value if isinstance(value, str) else None
+ facts = {"mode": mode, "paired": paired, "agent": agent,
+ "target": context["target"] if context else None, "home": home,
+ "capabilities": caps, "agents": [], "handoff": None,
+ "selectedRoute": result.get("route") if paired and result.get("route") in ("direct", "relay") else None,
+ "available": not failed, "historyContext": None}
+ if mode == "doctor":
+ declared = result.get("capabilities", {})
+ declared = declared.get("agents", {}) if isinstance(declared, dict) else {}
+ for name in ("claude", "codex", "antigravity"):
+ diagnostic = result.get(name, {})
+ status = diagnostic.get("status") if isinstance(diagnostic, dict) else None
+ allowed_status = {"available", "unavailable", "missing_home", "permission_denied",
+ "wrong_home", "missing_tool", "unknown", "disabled", "unsupported", "inbox_unavailable"}
+ declaration = declared.get(name, {}) if isinstance(declared, dict) else {}
+ home = diagnostic.get("selectedHome") if isinstance(diagnostic, dict) else None
+ facts["agents"].append({"agent": name, "status": status if isinstance(status, str) and status in allowed_status else "unknown",
+ "home": home if isinstance(home, str) else None,
+ "capabilities": {key: declaration[key] if isinstance(declaration, dict)
+ and type(declaration.get(key)) is bool else None
+ for key in ("send", "wake", "wait", "ack")}})
+ elif mode == "existing_intent":
+ query = result.get("handoffQuery")
+ if isinstance(query, dict) and query.get("context") in (
+ "ledger_missing", "ledger_corrupt", "id_unknown", "history_unavailable"):
+ facts["historyContext"] = query["context"]
+ public = result.get("handoff")
+ if isinstance(public, dict):
+ snapshot = {}
+ allowed = {"submission": {"not_attempted", "submitted", "refused", "unknown"},
+ "observation": {"unsupported", "not_requested", "pending", "observed", "failed"},
+ "ack": {"pending", "unsupported", "not_requested", "acknowledged"},
+ "wait": {"not_requested", "pending", "satisfied", "timed_out_unknown", "unsupported", "stopped", "failed"}}
+ for key, statuses in allowed.items():
+ value = public.get(key)
+ status = value.get("status") if isinstance(value, dict) else None
+ snapshot[key] = status if isinstance(status, str) and status in statuses else "unknown"
+ ack = public.get("ack", {})
+ observed = ack.get("receivedAtUtcMs") if isinstance(ack, dict) else None
+ snapshot["receivedAtUtcMs"] = observed if type(observed) is int and 0 <= observed <= 9007199254740991 else None
+ snapshot["late"] = ack.get("late") if isinstance(ack, dict) and type(ack.get("late")) is bool else None
+ assurance = ack.get("assurance") if isinstance(ack, dict) else None
+ snapshot["assurance"] = assurance if assurance in ("token_possession", "operator_confirmed") else None
+ if snapshot["ack"] != "acknowledged":
+ snapshot.update(receivedAtUtcMs=None, late=None, assurance=None)
+ wait = public.get("wait", {})
+ goal = wait.get("for") if isinstance(wait, dict) else None
+ snapshot["waitFor"] = goal if goal in ("none", "delivered", "acknowledged") else None
+ reason = wait.get("reason") if isinstance(wait, dict) else None
+ snapshot["waitReason"] = reason if reason in ("insufficient_budget", "deadline_before_effect", "evidence_unsupported", "evidence_failed", "history_unavailable", "stopped_by_operator", "invalid_handoff") else None
+ facts["handoff"] = snapshot
+ return facts
+
+
+def render_delivery_fact_lines(facts):
+ def capability(value):
+ return "supported (declared; eligibility unverified)" if value is True else "unsupported (declared)" if value is False else "unknown (not reported)"
+ if facts["mode"] == "doctor":
+ lines = ["Delivery capabilities (existing diagnostic snapshot)" if facts["available"] else
+ "Delivery capabilities unavailable (diagnostic snapshot not received)"]
+ for item in facts["agents"]:
+ basis = "diagnostic report" if item["status"] != "unknown" else "not reported"
+ lines.append(item["agent"] + " availability: " + item["status"] + " (" + basis + "; not per-target readiness)")
+ if item["agent"] in ("codex", "antigravity"):
+ lines.append(" Selected home: " + (explanation_text(item["home"]) if item["home"] else "unknown (not reported)"))
+ for key, value in item["capabilities"].items():
+ lines.append(" " + key.capitalize() + ": " + capability(value))
+ return lines + ["Injection observation: unsupported", "Task completion: unknown",
+ "Last delivery observation: unknown (not reported)",
+ "A diagnostic snapshot does not guarantee later delivery."]
+ if facts["mode"] == "existing_intent":
+ value = facts["handoff"] or {}
+ lines = ["Delivery evidence (existing intent snapshot)" if facts["available"] else
+ "Delivery evidence unavailable (history not read)"]
+ if facts["historyContext"]:
+ lines.append("History query: " + facts["historyContext"] + "; no history initialized or repaired.")
+ for key, label in (("submission", "Submission"), ("observation", "Injection observation"),
+ ("ack", "Explicit receipt"), ("wait", "Latest stored wait")):
+ lines.append(label + ": " + value.get(key, "unknown") + (" (stored observation)" if facts["available"] else " (not available)"))
+ observed = value.get("receivedAtUtcMs")
+ lines.append("Receipt recorded at (stored UTC milliseconds): " + (str(observed) if observed is not None else "unknown (not reported)"))
+ if value.get("assurance"):
+ lines.append("Receipt assurance: " + value["assurance"] + "; does not prove task completion")
+ if value["assurance"] == "token_possession":
+ lines.append("Receipt proves token possession, not original agent/model identity.")
+ if value.get("waitFor"):
+ lines.append("Latest stored wait goal: " + value["waitFor"])
+ if value.get("waitReason"):
+ lines.append("Latest stored wait reason: " + value["waitReason"])
+ if value.get("late") is not None:
+ lines.append("Receipt late relative to first wait: " + str(value["late"]).lower())
+ return lines + ["Task completion: unknown", "Stored evidence may be stale; this view does not refresh delivery.",
+ "Do not automatically resend after an unknown outcome."]
+ caps = facts["capabilities"] or {}
+ return ["Submission capability: " + capability(caps.get("send")),
+ "Wake capability: " + capability(caps.get("wake")),
+ "Injection observation: unsupported" if not facts["paired"] else "Injection observation: unknown (not reported)",
+ "Last delivery observation: unknown (not reported)",
+ "Task completion: unknown"]
+
+
# --------------------------------------------------------------------------
# Output
# --------------------------------------------------------------------------
@@ -4663,6 +4836,119 @@ def render_sessions(sessions: list[dict], where: str) -> str:
return f"Sessions on {where}:\n" + "\n".join(lines)
+def delivery_explanation_has_submission(result: dict) -> bool:
+ """Recognize reported submission evidence without changing native results."""
+ return (result.get("submitted") is True or result.get("dryRun") is False
+ or result.get("status") in ("queued", "posted", "submitted"))
+
+
+def render_delivery_explanation(args: argparse.Namespace, result: dict, *,
+ host: str | None = None, failed: bool = False,
+ mode: str = "readiness") -> str:
+ """Describe only existing dry-run facts; never inspect or execute a target.
+
+ This display-only allowlist deliberately excludes message bodies, native
+ errors, credentials, routing URIs and receipt authority. Missing remote
+ facts stay unknown rather than being filled from the sender's environment.
+ """
+ facts = build_delivery_explanation(args, result, mode=mode, failed=failed)
+ if mode != "readiness":
+ lines = render_delivery_fact_lines(facts)
+ if host:
+ lines.insert(1, "Destination host: " + explanation_text(host))
+ return "\n".join(lines)
+ requested = str(getattr(args, "to", ""))
+ unresolved_uri = requested.lower().startswith("session-peer:")
+ if unresolved_uri:
+ requested = "Reply-To URI (resolution unavailable)"
+ agent = facts["agent"]
+ if facts["paired"]:
+ requested = facts["target"] or "paired target (not reported by receiver)"
+ where = host or (("device:" + str(args.device)) if facts["paired"] else
+ (", ".join(getattr(args, "host", None) or []) or "this machine"))
+ lines = ["Delivery preview",
+ "Requested destination: " + explanation_text(requested) + " on " + explanation_text(where)]
+ if facts["paired"]:
+ lines.append("Selected paired route: " + (facts["selectedRoute"] + " (reported)" if facts["selectedRoute"] else "unknown (not reported)"))
+ target = None if facts["paired"] else result.get("target")
+ if isinstance(target, dict):
+ if agent == "claude" and type(target.get("pid")) is int:
+ resolved = "Claude PID " + str(target["pid"])
+ if isinstance(target.get("name"), str):
+ resolved += " (" + explanation_text(target["name"]) + ")"
+ lines.append("Resolved target: " + resolved)
+ elif agent == "codex" and isinstance(target.get("id"), str):
+ lines.append("Resolved target: codex:" + explanation_text(target["id"]))
+ if agent in ("codex", "antigravity"):
+ home = facts["home"]
+ lines.append(("Codex home: " if agent == "codex" else "Antigravity home: ")
+ + (explanation_text(home) if isinstance(home, str) else "unknown (not reported)"))
+ started = getattr(args, "_delivery_explanation_started", True)
+ unknown = result.get("status") == "unknown" or result.get("reason") in (
+ "outcome_unknown", "ssh_identity_evidence_unavailable") or (
+ "submitted" in result and result["submitted"] is None)
+ unexpected = delivery_explanation_has_submission(result)
+ dry_run_reported = result.get("dryRun") is True or result.get("status") in ("dry_run", "validated")
+ if not started:
+ lines.append("Destination checks: not run (command failed before destination checks)")
+ lines.append("Validation: not confirmed")
+ else:
+ lines.append("Validation: " + ("unexpected submission evidence; dry-run not confirmed" if unexpected else
+ "result unknown" if unknown else
+ "not confirmed" if failed else
+ "passed for this dry run" if dry_run_reported else
+ "not confirmed (dry-run evidence not reported)"))
+ if unexpected:
+ for key, label in (("dryRun", "dry-run"), ("submitted", "submitted")):
+ if type(result.get(key)) is bool:
+ lines.append("Reported " + label + ": " + str(result[key]).lower())
+ if result.get("status") in ("queued", "posted", "submitted", "unknown"):
+ lines.append("Reported native status: " + result["status"])
+ method = {"claude": "Claude Code inbox", "codex": "native Codex queue",
+ "antigravity": "registered Antigravity bridge"}.get(agent)
+ lines.append("Delivery method: " + (method + " (planned)" if method else "unknown (not resolved)"))
+ if not getattr(args, "wake", False):
+ wake = "not requested; eligibility not checked"
+ else:
+ fact = result.get("wake")
+ wake = ("preflight passed for this dry run; no turn started"
+ if not unknown and not unexpected and isinstance(fact, dict) and fact.get("status") == "validated" else
+ "requested; preflight not confirmed")
+ lines.extend(["Wake: " + wake,
+ ("Receipt ACK: unknown (receiver support not reported)" if facts["paired"]
+ and facts["capabilities"] is None else "Receipt ACK: unavailable for ordinary sends"),
+ "Consumption / completion: not observed",
+ "A later send repeats validation.",
+ "Do not automatically resend after an unknown outcome."])
+ if failed or unexpected:
+ lines.append("Next step: inspect the target and existing evidence before deciding what to do."
+ if unknown or unexpected else "Next step: resolve the validation failure before attempting a send.")
+ lines.extend(render_delivery_fact_lines(facts))
+ if facts["paired"] and agent is None:
+ lines.append("Selected agent / home: unknown (receiver context not reported)")
+ generation = getattr(args, "target_generation", None)
+ if generation is not None:
+ lines.append("Target generation: requested; existing validation result applies")
+ else:
+ lines.append("Target generation: not requested" if agent == "claude" and not facts["paired"]
+ else "Target generation: unsupported for this route")
+ return "\n".join(lines)
+
+
+def emit_delivery_explanation(args: argparse.Namespace, result: dict, *,
+ host: str | None = None, failed: bool = False,
+ stream=None, mode: str = "readiness") -> None:
+ explanation = render_delivery_explanation(args, result, host=host, failed=failed, mode=mode)
+ try:
+ generated = int(time.time() * 1000)
+ if 0 <= generated <= 9007199254740991:
+ explanation += "\nExplanation generated at (UTC milliseconds): " + str(generated) + "; not a delivery observation."
+ except Exception:
+ pass
+ print(explanation,
+ file=stream or (sys.stderr if failed or delivery_explanation_has_submission(result) else sys.stdout))
+
+
JSON_RESPONSE_SCHEMA_VERSION = 1
@@ -6036,6 +6322,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
emit(
args.json, payload, render_doctor(payload, "this machine"), command="doctor",
)
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ emit_delivery_explanation(args, payload, mode="doctor")
return 0
exit_code = 0
@@ -6044,9 +6333,15 @@ def cmd_doctor(args: argparse.Namespace) -> int:
return_host = _doctor_return_host(args) if args.check_return_route else None
for requested_host in args.host:
host = requested_host
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = False
+ args._delivery_explanation_host = requested_host
try:
transport = SshTransport(requested_host, args, tailnet_status)
host, ssh_opts = transport.host, transport.ssh_opts
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ args._delivery_explanation_host = display_host(requested_host, host)
remote_argv = ["doctor", "--no-update-notice"] + agent_remote_options(args)
if return_host:
remote_argv.extend(["--_return-host", return_host])
@@ -6068,6 +6363,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
if len(all_results) > 1:
print()
print(render_doctor(payload, display_host(requested_host, host)))
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ emit_delivery_explanation(args, payload, mode="doctor", host=display_host(requested_host, host))
except CcPeerError as exc:
exit_code = EXIT_ERROR
all_results.append(json_result(
@@ -6077,6 +6375,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
))
if not args.json:
print(human_text(f"session-peer: {requested_host}: {exc}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, {}, mode="doctor", failed=True,
+ host=display_host(requested_host, host), stream=sys.stderr)
if args.json:
emit_json_results(all_results)
return exit_code
@@ -8870,7 +9171,35 @@ def handoff_receipt_stdin(timeout=5):
raise handoff_error("invalid_receipt") from exc
+def cmd_handoff_explanation(args):
+ """Project existing history without saving retention or minting authority.
+
+ Reuses the same read-only ledger/clock primitives as #296 status/watch;
+ receipt schemas, wait history and ordinary output remain unchanged.
+ """
+ ledger = HandoffLedger()
+ correlation = handoff_uuid(args.correlation_id)
+ args._delivery_explanation_started = True
+ try:
+ epoch, record = ledger.record(correlation, read_only=True)
+ except CcPeerError as exc:
+ context = ("id_unknown" if exc.details.get("reason") == "handoff_id_unknown" else
+ "ledger_corrupt" if exc.details.get("reason") == "handoff_ledger_corrupt" else
+ "ledger_missing" if isinstance(exc.__cause__, FileNotFoundError) else "history_unavailable")
+ emit_delivery_explanation(args, {"handoffQuery": {"context": context}},
+ mode="existing_intent", failed=True, stream=sys.stderr)
+ return 1
+ active_clock = None
+ if record["capability"] and not record["capability"]["revoked"] and not record["ack"]:
+ active_clock = handoff_channel_epoch(ledger, read_only=True)
+ public = handoff_public(epoch, record, active_clock)
+ emit_delivery_explanation(args, {"handoff": public}, mode="existing_intent")
+ return 0
+
+
def cmd_handoff(args):
+ if args.handoff_action == "status" and getattr(args, "explain", False):
+ return cmd_handoff_explanation(args)
ledger = HandoffLedger()
action = args.handoff_action
if action == "init":
@@ -10465,9 +10794,13 @@ def cmd_send(args: argparse.Namespace) -> int:
validate_agent_send(adapter, args, text)
if not args.host:
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
result = LocalTransport().execute("send", adapter, args, text)
result.update(routing_metadata)
emit(args.json, result, adapter.submission_text(result, "this machine"), command="send")
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, result, failed=result.get("ok") is False)
if advertised_route and advertised_route["status"] == "unverified" and not args.json:
print(
"session-peer: reverse SSH reply route was not checked; "
@@ -10480,9 +10813,14 @@ def cmd_send(args: argparse.Namespace) -> int:
for requested_host in args.host:
host = requested_host
try:
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ args._delivery_explanation_host = requested_host
transport = SshTransport(requested_host, args, tailnet_status)
host, ssh_opts = transport.host, transport.ssh_opts
shown_host = display_host(requested_host, host)
+ if getattr(args, "explain", False):
+ args._delivery_explanation_host = shown_host
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)
@@ -10510,6 +10848,11 @@ def cmd_send(args: argparse.Namespace) -> int:
exit_code = EXIT_ERROR
if not args.json:
print(adapter.submission_text(payload, shown_host))
+ if getattr(args, "explain", False):
+ # Explain the received facts, before legacy adapter shaping
+ # replaces fields such as Claude's dryRun with request data.
+ emit_delivery_explanation(args, result, host=shown_host,
+ failed=payload.get("ok") is False or result.get("ok") is False)
except CcPeerError as exc:
exit_code = EXIT_ERROR
all_results.append(json_result(
@@ -10519,6 +10862,10 @@ def cmd_send(args: argparse.Namespace) -> int:
))
if not args.json:
print(human_text(f"session-peer: {requested_host}: {exc}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, exc.details,
+ host=display_host(requested_host, host),
+ failed=True, stream=sys.stderr)
if args.json:
emit_json_results(all_results)
@@ -10784,6 +11131,8 @@ def add_common(sub: argparse.ArgumentParser) -> None:
sending.add_argument("--wake-timeout", type=int, choices=range(1, 61), default=30, metavar="SECONDS",
help="wake deadline, 1..60 seconds (default: 30)")
sending.add_argument("--dry-run", action="store_true", help="resolve the target, send nothing")
+ sending.add_argument("--explain", action="store_true", help="explain existing dry-run evidence in text")
+ doctor.add_argument("--explain", action="store_true", help="explain existing diagnostic capabilities in text")
sending.set_defaults(func=cmd_send)
handoff = subparsers.add_parser("handoff", help="manage private correlated intents without submitting native messages")
@@ -10799,6 +11148,8 @@ def add_common(sub: argparse.ArgumentParser) -> None:
metavar="SECONDS", help="finite local watch budget, 1..60 seconds (default: 30)")
sub.add_argument("--interval-ms", type=handoff_watch_interval, default=1000,
metavar="MILLISECONDS", help="sample every 1000..5000 ms (default: 1000)")
+ if action == "status":
+ sub.add_argument("--explain", action="store_true", help="explain the existing read-only intent snapshot")
if action == "wait":
sub.add_argument("--wait-for", choices=("delivered", "acknowledged"), required=True)
sub.add_argument("--wait-timeout", type=handoff_timeout, default=30)
@@ -10954,6 +11305,19 @@ def _main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ if getattr(args, "explain", False):
+ if args.json:
+ parser.error("--explain requires text output")
+ if args.command == "send":
+ if not args.dry_run:
+ parser.error("--explain requires --dry-run")
+ if handoff_requested(args) or any(getattr(args, key, None) is not None for key in (
+ "_handoff_native_context", "_handoff_native_cutoff_ms", "_handoff_native_total_ms")):
+ parser.error("--explain does not support experimental handoff sends")
+ elif args.command == "handoff" and args.handoff_action != "status":
+ parser.error("--explain requires handoff status")
+ args._delivery_explanation_started = False
+ args._delivery_explanation_host = None
if runtime_notice_preference_requested(args):
if runtime_notice_preference_routed(args):
parser.error("Runtime notice preferences are local only; do not use host or SSH options")
@@ -10989,6 +11353,10 @@ def _main(argv: list[str] | None = None) -> int:
print(json.dumps(with_client_update(payload), ensure_ascii=False))
else:
print(human_text(f"session-peer: {message}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, exc.details, failed=True, stream=sys.stderr,
+ mode="doctor" if args.command == "doctor" else "existing_intent" if args.command == "handoff" else "readiness",
+ host=getattr(args, "_delivery_explanation_host", None))
exit_code = (
EXIT_NO_TARGET
if isinstance(exc, NoTargetError) or "no reachable session" in message
@@ -11004,6 +11372,12 @@ def _main(argv: list[str] | None = None) -> int:
print(json.dumps(with_client_update(payload), ensure_ascii=False))
else:
print(human_text(f"session-peer: {message}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ explanation = ({"reason": "outcome_unknown"}
+ if getattr(args, "_delivery_explanation_started", False) else {})
+ emit_delivery_explanation(args, explanation, failed=True, stream=sys.stderr,
+ mode="doctor" if args.command == "doctor" else "existing_intent" if args.command == "handoff" else "readiness",
+ host=getattr(args, "_delivery_explanation_host", None))
exit_code = EXIT_ERROR
if show_human_notice and not args.json:
emit_human_update_notice()
diff --git a/session_peer_core/cli.py b/session_peer_core/cli.py
index ffa8c4e..48cd82e 100644
--- a/session_peer_core/cli.py
+++ b/session_peer_core/cli.py
@@ -176,6 +176,8 @@ def add_common(sub: argparse.ArgumentParser) -> None:
sending.add_argument("--wake-timeout", type=int, choices=range(1, 61), default=30, metavar="SECONDS",
help="wake deadline, 1..60 seconds (default: 30)")
sending.add_argument("--dry-run", action="store_true", help="resolve the target, send nothing")
+ sending.add_argument("--explain", action="store_true", help="explain existing dry-run evidence in text")
+ doctor.add_argument("--explain", action="store_true", help="explain existing diagnostic capabilities in text")
sending.set_defaults(func=cmd_send)
handoff = subparsers.add_parser("handoff", help="manage private correlated intents without submitting native messages")
@@ -191,6 +193,8 @@ def add_common(sub: argparse.ArgumentParser) -> None:
metavar="SECONDS", help="finite local watch budget, 1..60 seconds (default: 30)")
sub.add_argument("--interval-ms", type=handoff_watch_interval, default=1000,
metavar="MILLISECONDS", help="sample every 1000..5000 ms (default: 1000)")
+ if action == "status":
+ sub.add_argument("--explain", action="store_true", help="explain the existing read-only intent snapshot")
if action == "wait":
sub.add_argument("--wait-for", choices=("delivered", "acknowledged"), required=True)
sub.add_argument("--wait-timeout", type=handoff_timeout, default=30)
@@ -346,6 +350,19 @@ def _main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ if getattr(args, "explain", False):
+ if args.json:
+ parser.error("--explain requires text output")
+ if args.command == "send":
+ if not args.dry_run:
+ parser.error("--explain requires --dry-run")
+ if handoff_requested(args) or any(getattr(args, key, None) is not None for key in (
+ "_handoff_native_context", "_handoff_native_cutoff_ms", "_handoff_native_total_ms")):
+ parser.error("--explain does not support experimental handoff sends")
+ elif args.command == "handoff" and args.handoff_action != "status":
+ parser.error("--explain requires handoff status")
+ args._delivery_explanation_started = False
+ args._delivery_explanation_host = None
if runtime_notice_preference_requested(args):
if runtime_notice_preference_routed(args):
parser.error("Runtime notice preferences are local only; do not use host or SSH options")
@@ -381,6 +398,10 @@ def _main(argv: list[str] | None = None) -> int:
print(json.dumps(with_client_update(payload), ensure_ascii=False))
else:
print(human_text(f"session-peer: {message}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, exc.details, failed=True, stream=sys.stderr,
+ mode="doctor" if args.command == "doctor" else "existing_intent" if args.command == "handoff" else "readiness",
+ host=getattr(args, "_delivery_explanation_host", None))
exit_code = (
EXIT_NO_TARGET
if isinstance(exc, NoTargetError) or "no reachable session" in message
@@ -396,6 +417,12 @@ def _main(argv: list[str] | None = None) -> int:
print(json.dumps(with_client_update(payload), ensure_ascii=False))
else:
print(human_text(f"session-peer: {message}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ explanation = ({"reason": "outcome_unknown"}
+ if getattr(args, "_delivery_explanation_started", False) else {})
+ emit_delivery_explanation(args, explanation, failed=True, stream=sys.stderr,
+ mode="doctor" if args.command == "doctor" else "existing_intent" if args.command == "handoff" else "readiness",
+ host=getattr(args, "_delivery_explanation_host", None))
exit_code = EXIT_ERROR
if show_human_notice and not args.json:
emit_human_update_notice()
diff --git a/session_peer_core/commands.py b/session_peer_core/commands.py
index 5bdc069..a2252e5 100644
--- a/session_peer_core/commands.py
+++ b/session_peer_core/commands.py
@@ -106,6 +106,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
emit(
args.json, payload, render_doctor(payload, "this machine"), command="doctor",
)
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ emit_delivery_explanation(args, payload, mode="doctor")
return 0
exit_code = 0
@@ -114,9 +117,15 @@ def cmd_doctor(args: argparse.Namespace) -> int:
return_host = _doctor_return_host(args) if args.check_return_route else None
for requested_host in args.host:
host = requested_host
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = False
+ args._delivery_explanation_host = requested_host
try:
transport = SshTransport(requested_host, args, tailnet_status)
host, ssh_opts = transport.host, transport.ssh_opts
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ args._delivery_explanation_host = display_host(requested_host, host)
remote_argv = ["doctor", "--no-update-notice"] + agent_remote_options(args)
if return_host:
remote_argv.extend(["--_return-host", return_host])
@@ -138,6 +147,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
if len(all_results) > 1:
print()
print(render_doctor(payload, display_host(requested_host, host)))
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ emit_delivery_explanation(args, payload, mode="doctor", host=display_host(requested_host, host))
except CcPeerError as exc:
exit_code = EXIT_ERROR
all_results.append(json_result(
@@ -147,6 +159,9 @@ def cmd_doctor(args: argparse.Namespace) -> int:
))
if not args.json:
print(human_text(f"session-peer: {requested_host}: {exc}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, {}, mode="doctor", failed=True,
+ host=display_host(requested_host, host), stream=sys.stderr)
if args.json:
emit_json_results(all_results)
return exit_code
diff --git a/session_peer_core/explanation.py b/session_peer_core/explanation.py
new file mode 100644
index 0000000..01940bf
--- /dev/null
+++ b/session_peer_core/explanation.py
@@ -0,0 +1,171 @@
+# Display-only explanation facts. No discovery, transport, ledger or clock I/O.
+
+
+def validated_delivery_context(value):
+ """Accept the complete private receiver context, never guessed fragments."""
+ if not isinstance(value, dict) or set(value) != {
+ "schemaVersion", "agent", "target", "home", "capabilities"}:
+ return None
+ if type(value["schemaVersion"]) is not int or value["schemaVersion"] != 1:
+ return None
+ if value["agent"] not in ("claude", "codex", "antigravity"):
+ return None
+ for key in ("target", "home"):
+ item = value[key]
+ if item is not None:
+ if (not isinstance(item, str) or not item or item.lower().startswith("session-peer:")
+ or any(ord(c) < 32 or 127 <= ord(c) <= 159 for c in item)):
+ return None
+ try:
+ if len(item.encode("utf-8")) > 4096:
+ return None
+ except UnicodeError:
+ return None
+ if value["target"] is not None:
+ pattern = (r"claude:[1-9][0-9]*" if value["agent"] == "claude" else
+ value["agent"] + r":[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+ if re.fullmatch(pattern, value["target"]) is None:
+ return None
+ if value["home"] is not None and not (PurePosixPath(value["home"]).is_absolute()
+ or PureWindowsPath(value["home"]).is_absolute()):
+ return None
+ caps = value["capabilities"]
+ if not isinstance(caps, dict) or set(caps) != {"send", "wake", "wait", "ack"}:
+ return None
+ if any(type(item) is not bool for item in caps.values()):
+ return None
+ if len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) > 8192:
+ return None
+ return {**value, "capabilities": dict(caps)}
+
+
+def explanation_text(value):
+ """Sanitize only opt-in explanations, preserving ordinary output behavior."""
+ text = human_text(value)
+ return "".join("\\u%04x" % ord(c) if ord(c) in (
+ 0x200b, 0x200c, 0x200d, 0x200e, 0x200f, 0x202a, 0x202b,
+ 0x202c, 0x202d, 0x202e, 0x2060, 0x2066, 0x2067, 0x2068, 0x2069, 0xfeff)
+ else c for c in text)
+
+
+def build_delivery_explanation(args, result, *, mode="readiness", failed=False):
+ """Normalize only allowlisted facts already supplied by the caller.
+
+ Declaration never proves eligibility, and a snapshot never proves delivery.
+ The returned object is internal and is not a new JSON/wire contract.
+ """
+ if mode not in ("readiness", "doctor", "existing_intent"):
+ raise ValueError("unsupported delivery explanation mode")
+ paired = bool(getattr(args, "device", None))
+ context = validated_delivery_context(getattr(args, "_delivery_explanation_context", None)) if paired else None
+ requested = str(getattr(args, "to", "") or "")
+ uri = requested.lower().startswith("session-peer:")
+ agent = (context["agent"] if context else None) if paired else (
+ None if uri or not requested else "codex" if requested.startswith("codex:") else
+ "antigravity" if requested.startswith("antigravity:") else "claude")
+ caps = context["capabilities"] if context else None
+ if not paired and agent is not None:
+ caps = {key: getattr(AGENTS.get(agent).capabilities, key)
+ for key in ("send", "wake", "wait", "ack")}
+ home = context["home"] if context else None
+ if not paired and agent in ("codex", "antigravity"):
+ value = result.get("codexHome" if agent == "codex" else "antigravityHome")
+ home = value if isinstance(value, str) else None
+ facts = {"mode": mode, "paired": paired, "agent": agent,
+ "target": context["target"] if context else None, "home": home,
+ "capabilities": caps, "agents": [], "handoff": None,
+ "selectedRoute": result.get("route") if paired and result.get("route") in ("direct", "relay") else None,
+ "available": not failed, "historyContext": None}
+ if mode == "doctor":
+ declared = result.get("capabilities", {})
+ declared = declared.get("agents", {}) if isinstance(declared, dict) else {}
+ for name in ("claude", "codex", "antigravity"):
+ diagnostic = result.get(name, {})
+ status = diagnostic.get("status") if isinstance(diagnostic, dict) else None
+ allowed_status = {"available", "unavailable", "missing_home", "permission_denied",
+ "wrong_home", "missing_tool", "unknown", "disabled", "unsupported", "inbox_unavailable"}
+ declaration = declared.get(name, {}) if isinstance(declared, dict) else {}
+ home = diagnostic.get("selectedHome") if isinstance(diagnostic, dict) else None
+ facts["agents"].append({"agent": name, "status": status if isinstance(status, str) and status in allowed_status else "unknown",
+ "home": home if isinstance(home, str) else None,
+ "capabilities": {key: declaration[key] if isinstance(declaration, dict)
+ and type(declaration.get(key)) is bool else None
+ for key in ("send", "wake", "wait", "ack")}})
+ elif mode == "existing_intent":
+ query = result.get("handoffQuery")
+ if isinstance(query, dict) and query.get("context") in (
+ "ledger_missing", "ledger_corrupt", "id_unknown", "history_unavailable"):
+ facts["historyContext"] = query["context"]
+ public = result.get("handoff")
+ if isinstance(public, dict):
+ snapshot = {}
+ allowed = {"submission": {"not_attempted", "submitted", "refused", "unknown"},
+ "observation": {"unsupported", "not_requested", "pending", "observed", "failed"},
+ "ack": {"pending", "unsupported", "not_requested", "acknowledged"},
+ "wait": {"not_requested", "pending", "satisfied", "timed_out_unknown", "unsupported", "stopped", "failed"}}
+ for key, statuses in allowed.items():
+ value = public.get(key)
+ status = value.get("status") if isinstance(value, dict) else None
+ snapshot[key] = status if isinstance(status, str) and status in statuses else "unknown"
+ ack = public.get("ack", {})
+ observed = ack.get("receivedAtUtcMs") if isinstance(ack, dict) else None
+ snapshot["receivedAtUtcMs"] = observed if type(observed) is int and 0 <= observed <= 9007199254740991 else None
+ snapshot["late"] = ack.get("late") if isinstance(ack, dict) and type(ack.get("late")) is bool else None
+ assurance = ack.get("assurance") if isinstance(ack, dict) else None
+ snapshot["assurance"] = assurance if assurance in ("token_possession", "operator_confirmed") else None
+ if snapshot["ack"] != "acknowledged":
+ snapshot.update(receivedAtUtcMs=None, late=None, assurance=None)
+ wait = public.get("wait", {})
+ goal = wait.get("for") if isinstance(wait, dict) else None
+ snapshot["waitFor"] = goal if goal in ("none", "delivered", "acknowledged") else None
+ reason = wait.get("reason") if isinstance(wait, dict) else None
+ snapshot["waitReason"] = reason if reason in ("insufficient_budget", "deadline_before_effect", "evidence_unsupported", "evidence_failed", "history_unavailable", "stopped_by_operator", "invalid_handoff") else None
+ facts["handoff"] = snapshot
+ return facts
+
+
+def render_delivery_fact_lines(facts):
+ def capability(value):
+ return "supported (declared; eligibility unverified)" if value is True else "unsupported (declared)" if value is False else "unknown (not reported)"
+ if facts["mode"] == "doctor":
+ lines = ["Delivery capabilities (existing diagnostic snapshot)" if facts["available"] else
+ "Delivery capabilities unavailable (diagnostic snapshot not received)"]
+ for item in facts["agents"]:
+ basis = "diagnostic report" if item["status"] != "unknown" else "not reported"
+ lines.append(item["agent"] + " availability: " + item["status"] + " (" + basis + "; not per-target readiness)")
+ if item["agent"] in ("codex", "antigravity"):
+ lines.append(" Selected home: " + (explanation_text(item["home"]) if item["home"] else "unknown (not reported)"))
+ for key, value in item["capabilities"].items():
+ lines.append(" " + key.capitalize() + ": " + capability(value))
+ return lines + ["Injection observation: unsupported", "Task completion: unknown",
+ "Last delivery observation: unknown (not reported)",
+ "A diagnostic snapshot does not guarantee later delivery."]
+ if facts["mode"] == "existing_intent":
+ value = facts["handoff"] or {}
+ lines = ["Delivery evidence (existing intent snapshot)" if facts["available"] else
+ "Delivery evidence unavailable (history not read)"]
+ if facts["historyContext"]:
+ lines.append("History query: " + facts["historyContext"] + "; no history initialized or repaired.")
+ for key, label in (("submission", "Submission"), ("observation", "Injection observation"),
+ ("ack", "Explicit receipt"), ("wait", "Latest stored wait")):
+ lines.append(label + ": " + value.get(key, "unknown") + (" (stored observation)" if facts["available"] else " (not available)"))
+ observed = value.get("receivedAtUtcMs")
+ lines.append("Receipt recorded at (stored UTC milliseconds): " + (str(observed) if observed is not None else "unknown (not reported)"))
+ if value.get("assurance"):
+ lines.append("Receipt assurance: " + value["assurance"] + "; does not prove task completion")
+ if value["assurance"] == "token_possession":
+ lines.append("Receipt proves token possession, not original agent/model identity.")
+ if value.get("waitFor"):
+ lines.append("Latest stored wait goal: " + value["waitFor"])
+ if value.get("waitReason"):
+ lines.append("Latest stored wait reason: " + value["waitReason"])
+ if value.get("late") is not None:
+ lines.append("Receipt late relative to first wait: " + str(value["late"]).lower())
+ return lines + ["Task completion: unknown", "Stored evidence may be stale; this view does not refresh delivery.",
+ "Do not automatically resend after an unknown outcome."]
+ caps = facts["capabilities"] or {}
+ return ["Submission capability: " + capability(caps.get("send")),
+ "Wake capability: " + capability(caps.get("wake")),
+ "Injection observation: unsupported" if not facts["paired"] else "Injection observation: unknown (not reported)",
+ "Last delivery observation: unknown (not reported)",
+ "Task completion: unknown"]
diff --git a/session_peer_core/handoff.py b/session_peer_core/handoff.py
index f478d36..dac9493 100644
--- a/session_peer_core/handoff.py
+++ b/session_peer_core/handoff.py
@@ -1108,7 +1108,35 @@ def handoff_receipt_stdin(timeout=5):
raise handoff_error("invalid_receipt") from exc
+def cmd_handoff_explanation(args):
+ """Project existing history without saving retention or minting authority.
+
+ Reuses the same read-only ledger/clock primitives as #296 status/watch;
+ receipt schemas, wait history and ordinary output remain unchanged.
+ """
+ ledger = HandoffLedger()
+ correlation = handoff_uuid(args.correlation_id)
+ args._delivery_explanation_started = True
+ try:
+ epoch, record = ledger.record(correlation, read_only=True)
+ except CcPeerError as exc:
+ context = ("id_unknown" if exc.details.get("reason") == "handoff_id_unknown" else
+ "ledger_corrupt" if exc.details.get("reason") == "handoff_ledger_corrupt" else
+ "ledger_missing" if isinstance(exc.__cause__, FileNotFoundError) else "history_unavailable")
+ emit_delivery_explanation(args, {"handoffQuery": {"context": context}},
+ mode="existing_intent", failed=True, stream=sys.stderr)
+ return 1
+ active_clock = None
+ if record["capability"] and not record["capability"]["revoked"] and not record["ack"]:
+ active_clock = handoff_channel_epoch(ledger, read_only=True)
+ public = handoff_public(epoch, record, active_clock)
+ emit_delivery_explanation(args, {"handoff": public}, mode="existing_intent")
+ return 0
+
+
def cmd_handoff(args):
+ if args.handoff_action == "status" and getattr(args, "explain", False):
+ return cmd_handoff_explanation(args)
ledger = HandoffLedger()
action = args.handoff_action
if action == "init":
diff --git a/session_peer_core/output.py b/session_peer_core/output.py
index 1bf651c..8aaad38 100644
--- a/session_peer_core/output.py
+++ b/session_peer_core/output.py
@@ -43,6 +43,119 @@ def render_sessions(sessions: list[dict], where: str) -> str:
return f"Sessions on {where}:\n" + "\n".join(lines)
+def delivery_explanation_has_submission(result: dict) -> bool:
+ """Recognize reported submission evidence without changing native results."""
+ return (result.get("submitted") is True or result.get("dryRun") is False
+ or result.get("status") in ("queued", "posted", "submitted"))
+
+
+def render_delivery_explanation(args: argparse.Namespace, result: dict, *,
+ host: str | None = None, failed: bool = False,
+ mode: str = "readiness") -> str:
+ """Describe only existing dry-run facts; never inspect or execute a target.
+
+ This display-only allowlist deliberately excludes message bodies, native
+ errors, credentials, routing URIs and receipt authority. Missing remote
+ facts stay unknown rather than being filled from the sender's environment.
+ """
+ facts = build_delivery_explanation(args, result, mode=mode, failed=failed)
+ if mode != "readiness":
+ lines = render_delivery_fact_lines(facts)
+ if host:
+ lines.insert(1, "Destination host: " + explanation_text(host))
+ return "\n".join(lines)
+ requested = str(getattr(args, "to", ""))
+ unresolved_uri = requested.lower().startswith("session-peer:")
+ if unresolved_uri:
+ requested = "Reply-To URI (resolution unavailable)"
+ agent = facts["agent"]
+ if facts["paired"]:
+ requested = facts["target"] or "paired target (not reported by receiver)"
+ where = host or (("device:" + str(args.device)) if facts["paired"] else
+ (", ".join(getattr(args, "host", None) or []) or "this machine"))
+ lines = ["Delivery preview",
+ "Requested destination: " + explanation_text(requested) + " on " + explanation_text(where)]
+ if facts["paired"]:
+ lines.append("Selected paired route: " + (facts["selectedRoute"] + " (reported)" if facts["selectedRoute"] else "unknown (not reported)"))
+ target = None if facts["paired"] else result.get("target")
+ if isinstance(target, dict):
+ if agent == "claude" and type(target.get("pid")) is int:
+ resolved = "Claude PID " + str(target["pid"])
+ if isinstance(target.get("name"), str):
+ resolved += " (" + explanation_text(target["name"]) + ")"
+ lines.append("Resolved target: " + resolved)
+ elif agent == "codex" and isinstance(target.get("id"), str):
+ lines.append("Resolved target: codex:" + explanation_text(target["id"]))
+ if agent in ("codex", "antigravity"):
+ home = facts["home"]
+ lines.append(("Codex home: " if agent == "codex" else "Antigravity home: ")
+ + (explanation_text(home) if isinstance(home, str) else "unknown (not reported)"))
+ started = getattr(args, "_delivery_explanation_started", True)
+ unknown = result.get("status") == "unknown" or result.get("reason") in (
+ "outcome_unknown", "ssh_identity_evidence_unavailable") or (
+ "submitted" in result and result["submitted"] is None)
+ unexpected = delivery_explanation_has_submission(result)
+ dry_run_reported = result.get("dryRun") is True or result.get("status") in ("dry_run", "validated")
+ if not started:
+ lines.append("Destination checks: not run (command failed before destination checks)")
+ lines.append("Validation: not confirmed")
+ else:
+ lines.append("Validation: " + ("unexpected submission evidence; dry-run not confirmed" if unexpected else
+ "result unknown" if unknown else
+ "not confirmed" if failed else
+ "passed for this dry run" if dry_run_reported else
+ "not confirmed (dry-run evidence not reported)"))
+ if unexpected:
+ for key, label in (("dryRun", "dry-run"), ("submitted", "submitted")):
+ if type(result.get(key)) is bool:
+ lines.append("Reported " + label + ": " + str(result[key]).lower())
+ if result.get("status") in ("queued", "posted", "submitted", "unknown"):
+ lines.append("Reported native status: " + result["status"])
+ method = {"claude": "Claude Code inbox", "codex": "native Codex queue",
+ "antigravity": "registered Antigravity bridge"}.get(agent)
+ lines.append("Delivery method: " + (method + " (planned)" if method else "unknown (not resolved)"))
+ if not getattr(args, "wake", False):
+ wake = "not requested; eligibility not checked"
+ else:
+ fact = result.get("wake")
+ wake = ("preflight passed for this dry run; no turn started"
+ if not unknown and not unexpected and isinstance(fact, dict) and fact.get("status") == "validated" else
+ "requested; preflight not confirmed")
+ lines.extend(["Wake: " + wake,
+ ("Receipt ACK: unknown (receiver support not reported)" if facts["paired"]
+ and facts["capabilities"] is None else "Receipt ACK: unavailable for ordinary sends"),
+ "Consumption / completion: not observed",
+ "A later send repeats validation.",
+ "Do not automatically resend after an unknown outcome."])
+ if failed or unexpected:
+ lines.append("Next step: inspect the target and existing evidence before deciding what to do."
+ if unknown or unexpected else "Next step: resolve the validation failure before attempting a send.")
+ lines.extend(render_delivery_fact_lines(facts))
+ if facts["paired"] and agent is None:
+ lines.append("Selected agent / home: unknown (receiver context not reported)")
+ generation = getattr(args, "target_generation", None)
+ if generation is not None:
+ lines.append("Target generation: requested; existing validation result applies")
+ else:
+ lines.append("Target generation: not requested" if agent == "claude" and not facts["paired"]
+ else "Target generation: unsupported for this route")
+ return "\n".join(lines)
+
+
+def emit_delivery_explanation(args: argparse.Namespace, result: dict, *,
+ host: str | None = None, failed: bool = False,
+ stream=None, mode: str = "readiness") -> None:
+ explanation = render_delivery_explanation(args, result, host=host, failed=failed, mode=mode)
+ try:
+ generated = int(time.time() * 1000)
+ if 0 <= generated <= 9007199254740991:
+ explanation += "\nExplanation generated at (UTC milliseconds): " + str(generated) + "; not a delivery observation."
+ except Exception:
+ pass
+ print(explanation,
+ file=stream or (sys.stderr if failed or delivery_explanation_has_submission(result) else sys.stdout))
+
+
JSON_RESPONSE_SCHEMA_VERSION = 1
diff --git a/session_peer_core/sending.py b/session_peer_core/sending.py
index 2a34917..7f9d3fa 100644
--- a/session_peer_core/sending.py
+++ b/session_peer_core/sending.py
@@ -78,9 +78,13 @@ def cmd_send(args: argparse.Namespace) -> int:
validate_agent_send(adapter, args, text)
if not args.host:
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
result = LocalTransport().execute("send", adapter, args, text)
result.update(routing_metadata)
emit(args.json, result, adapter.submission_text(result, "this machine"), command="send")
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, result, failed=result.get("ok") is False)
if advertised_route and advertised_route["status"] == "unverified" and not args.json:
print(
"session-peer: reverse SSH reply route was not checked; "
@@ -93,9 +97,14 @@ def cmd_send(args: argparse.Namespace) -> int:
for requested_host in args.host:
host = requested_host
try:
+ if getattr(args, "explain", False):
+ args._delivery_explanation_started = True
+ args._delivery_explanation_host = requested_host
transport = SshTransport(requested_host, args, tailnet_status)
host, ssh_opts = transport.host, transport.ssh_opts
shown_host = display_host(requested_host, host)
+ if getattr(args, "explain", False):
+ args._delivery_explanation_host = shown_host
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)
@@ -123,6 +132,11 @@ def cmd_send(args: argparse.Namespace) -> int:
exit_code = EXIT_ERROR
if not args.json:
print(adapter.submission_text(payload, shown_host))
+ if getattr(args, "explain", False):
+ # Explain the received facts, before legacy adapter shaping
+ # replaces fields such as Claude's dryRun with request data.
+ emit_delivery_explanation(args, result, host=shown_host,
+ failed=payload.get("ok") is False or result.get("ok") is False)
except CcPeerError as exc:
exit_code = EXIT_ERROR
all_results.append(json_result(
@@ -132,6 +146,10 @@ def cmd_send(args: argparse.Namespace) -> int:
))
if not args.json:
print(human_text(f"session-peer: {requested_host}: {exc}"), file=sys.stderr)
+ if getattr(args, "explain", False):
+ emit_delivery_explanation(args, exc.details,
+ host=display_host(requested_host, host),
+ failed=True, stream=sys.stderr)
if args.json:
emit_json_results(all_results)
diff --git a/session_peer_relay/app.py b/session_peer_relay/app.py
index a7a8c48..ca5c8e0 100644
--- a/session_peer_relay/app.py
+++ b/session_peer_relay/app.py
@@ -113,7 +113,7 @@ async def dispatch(self, peer, value):
raise Rejected('key_rotation_restricted')
if op == 'probe':
return {'ok': True, 'device': self.store.device, 'receiverReady': True,
- 'features': ['identity-rotation-v1']}
+ 'features': ['identity-rotation-v1', 'delivery-explanation-v1']}
if op == 'status':
return self.store.status(peer, value['body'])
if op == 'rotation.prepare':
@@ -139,8 +139,11 @@ async def dispatch(self, peer, value):
for row in part.get('sessions', []))
return {'ok': not errors, 'sessions': sessions, 'errors': errors}
body = value['body']
- if not isinstance(body, dict) or set(body) != {'target', 'message'}:
+ if (not isinstance(body, dict) or set(body) not in (
+ {'target', 'message'}, {'target', 'message', 'explain'})
+ or ('explain' in body and (op != 'resolve' or body['explain'] is not True))):
raise Rejected('invalid_body')
+ explanation = {'explain': True} if body.get('explain') is True else {}
alias, text = body['target'], body['message']
if not isinstance(alias, str):
raise Rejected('invalid_target')
@@ -155,11 +158,11 @@ async def dispatch(self, peer, value):
raise Rejected('framed_message_too_large')
# Binding overhead must also fit, before resolve or journal begin.
try:
- worker_frame(binding, 'resolve' if op == 'resolve' else 'send', text, key)
+ worker_frame(binding, 'resolve' if op == 'resolve' else 'send', text, key, **explanation)
except FrameRejected as exc:
raise Rejected(str(exc)) from None
if op == 'resolve':
- result = await self.native.invoke(binding, 'resolve', text, peer_fingerprint=key)
+ result = await self.native.invoke(binding, 'resolve', text, peer_fingerprint=key, **explanation)
if result.get('reason') == 'codex_executable_not_found':
self.lifecycle_event('native_preflight_refused', operation='resolve',
reason='codex_executable_not_found')
@@ -488,7 +491,7 @@ async def finish_pair(store, invite, route, credential, *, offer=False):
raise
-async def exchange(store, peer, op, body=None, ident=None, route='auto', credential=None):
+async def exchange(store, peer, op, body=None, ident=None, route='auto', credential=None, *, explain=False):
if store.recovery_required() and op not in ('probe', 'status', 'rotation.status'):
raise Rejected('recovery_required')
record = store.peer(peer)
@@ -517,7 +520,9 @@ async def ready_once(kind, attempt_id):
raise Rejected('peer_not_ready')
if op.startswith('rotation.') and 'identity-rotation-v1' not in response.get('features', []):
raise Rejected('rotation_unsupported')
- return kind, channel, attempt_id
+ features = response.get('features')
+ supported = isinstance(features, list) and 'delivery-explanation-v1' in features
+ return kind, channel, attempt_id, supported
except TimeoutError:
with contextlib.suppress(Exception):
await channel.close()
@@ -584,7 +589,7 @@ async def ready(kind):
if (not task.cancelled() and task.exception() is None
and (winner is None or task.result()[1] is not winner[1])):
await task.result()[1].close()
- kind, channel, selected_attempt_id = winner
+ kind, channel, selected_attempt_id, explanation_supported = winner
setup = ({'setupAttempts': setup_attempts[kind], 'setupDegraded': True,
'setupFailureHistory': history[kind]} if setup_attempts[kind] > 1 else {})
submitted = False
@@ -593,10 +598,14 @@ async def ready(kind):
# Preserve the preflighted request ID/body; Secure.send rechecks framing
# after refreshing the expiry, before any application bytes are written.
value['expires'] = time.time() + 60
+ if op == 'resolve' and explain is True and explanation_supported:
+ value['body'] = {**body, 'explain': True}
# Exactly one application submission, after path selection. No failover resend.
await channel.send(value)
submitted = True
result = await channel.recv()
+ if not (op == 'resolve' and explain is True and explanation_supported):
+ result = {key: item for key, item in result.items() if key != 'deliveryContext'}
return {**result, 'route': kind, 'receiverAccepted': result.get('ok', False),
'relayAttached': kind == 'relay', 'requestId': value['id'],
'attemptId': selected_attempt_id, **setup}
diff --git a/session_peer_relay/cli.py b/session_peer_relay/cli.py
index d904d1e..dff7d4a 100644
--- a/session_peer_relay/cli.py
+++ b/session_peer_relay/cli.py
@@ -22,6 +22,7 @@
from .transport_errors import NoAuthenticatedRoute, TransportFailure
from .guidance import with_guidance
from .pairing_ui import OfferOutcomeUnknown, offer_failure
+from .framing import valid_delivery_context
def connection_diagnostics(exc):
@@ -460,9 +461,13 @@ async def core_exchange(args):
text = core.wrap_message(text, with_from=True, with_reply=False, explicit_host=None)
body = {'target': args.to, 'message': text}
operation = 'resolve' if args.command == 'send' and args.dry_run else args.command
+ credential = device_credential(args, store, args.device)
+ if operation == 'resolve' and getattr(args, 'explain', False):
+ args._delivery_explanation_started = True
result = await exchange(store, args.device, operation, body,
getattr(args, 'request_id', None), args.device_route,
- device_credential(args, store, args.device))
+ credential,
+ **({'explain': True} if operation == 'resolve' and getattr(args, 'explain', False) else {}))
if args.command == 'list' and args.agent:
result['sessions'] = [row for row in result.get('sessions', []) if row.get('agent') == args.agent]
return result
@@ -477,6 +482,13 @@ def invoke_core(args):
except Exception as exc:
result = {'ok': False, 'reason': str(exc) if isinstance(exc, Rejected) else type(exc).__name__,
'retryAllowed': False, 'consumptionConfirmed': False, **connection_diagnostics(exc)}
+ delivery_context = result.pop('deliveryContext', None)
+ explain = (args.command == 'send' and getattr(args, 'dry_run', False)
+ and getattr(args, 'explain', False))
+ if hasattr(args, '_delivery_explanation_context'):
+ del args._delivery_explanation_context
+ if explain and valid_delivery_context(delivery_context):
+ args._delivery_explanation_context = delivery_context
result = with_guidance(result)
result.update(device=args.device, host='device:'+args.device, transport='paired_device')
display = core.human_text(result)
@@ -488,4 +500,6 @@ def invoke_core(args):
str(row.get('target', ''))+' '+str(row.get('agent', ''))+' '+str(row.get('id', row.get('pid', '')))+' '+str(row.get('status', 'unknown'))
for row in display.get('sessions', []))
core.emit(args.json, result, human, command=args.command)
+ if explain:
+ core.emit_delivery_explanation(args, result, failed=not result.get('ok'))
return 0 if result.get('ok') else 1
diff --git a/session_peer_relay/framing.py b/session_peer_relay/framing.py
index 5e4025c..9e03d30 100644
--- a/session_peer_relay/framing.py
+++ b/session_peer_relay/framing.py
@@ -1,10 +1,51 @@
"""Byte budgets checked before application or native-worker effects."""
import json
+import re
+from pathlib import PurePosixPath, PureWindowsPath
MAX_TEXT_BYTES = 32 * 1024
MAX_APPLICATION_BYTES = 64 * 1024
# Includes binding, operation and JSON escaping, not just the raw message.
MAX_WORKER_BYTES = 256 * 1024
+MAX_DELIVERY_CONTEXT_BYTES = 8 * 1024
+
+
+def valid_delivery_context(value):
+ """Accept only bounded receiver facts; malformed context never spoils a result."""
+ if (type(value) is not dict
+ or set(value) != {'schemaVersion', 'agent', 'target', 'home', 'capabilities'}
+ or type(value['schemaVersion']) is not int or value['schemaVersion'] != 1
+ or value['agent'] not in ('claude', 'codex', 'antigravity')):
+ return False
+ for key in ('target', 'home'):
+ item = value[key]
+ if item is not None:
+ if not isinstance(item, str) or not item or any(ord(char) < 32 or 127 <= ord(char) <= 159 for char in item):
+ return False
+ try:
+ if len(item.encode('utf-8')) > 4096:
+ return False
+ except UnicodeError:
+ return False
+ if value['target'] is not None and not value['target'].startswith(value['agent'] + ':'):
+ return False
+ if value['target'] is not None:
+ identifier = value['target'].split(':', 1)[1]
+ pattern = r'[1-9][0-9]*' if value['agent'] == 'claude' else r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}'
+ if re.fullmatch(pattern, identifier) is None:
+ return False
+ if value['home'] is not None and not (
+ PurePosixPath(value['home']).is_absolute() or PureWindowsPath(value['home']).is_absolute()):
+ return False
+ capabilities = value['capabilities']
+ if (type(capabilities) is not dict or set(capabilities) != {'send', 'wake', 'wait', 'ack'}
+ or any(type(item) is not bool for item in capabilities.values())):
+ return False
+ try:
+ encoded_frame(value, MAX_DELIVERY_CONTEXT_BYTES, 'delivery_context_too_large')
+ except FrameRejected:
+ return False
+ return True
class FrameRejected(ValueError):
@@ -25,10 +66,14 @@ def application_frame(value):
return encoded_frame(value, MAX_APPLICATION_BYTES, 'application_frame_too_large')
-def worker_frame(binding, operation, text, peer_fingerprint=None):
+def worker_frame(binding, operation, text, peer_fingerprint=None, *, explain=False):
value = {'binding': binding, 'operation': operation, 'text': text}
if peer_fingerprint is not None:
value['peerFingerprint'] = peer_fingerprint
+ if explain:
+ if operation != 'resolve' or explain is not True:
+ raise FrameRejected('invalid_explain')
+ value['explain'] = True
return encoded_frame(value,
MAX_WORKER_BYTES, 'native_worker_frame_too_large')
diff --git a/session_peer_relay/native.py b/session_peer_relay/native.py
index a59206f..b76f770 100644
--- a/session_peer_relay/native.py
+++ b/session_peer_relay/native.py
@@ -229,9 +229,10 @@ class Native:
def __init__(self):
self.slots = asyncio.Semaphore(2)
- async def invoke(self, binding, operation, text=None, *, peer_fingerprint=None):
+ async def invoke(self, binding, operation, text=None, *, peer_fingerprint=None, explain=False):
try:
- payload = worker_frame(binding, operation, text, peer_fingerprint)
+ payload = worker_frame(binding, operation, text, peer_fingerprint,
+ **({'explain': True} if explain else {}))
except FrameRejected as exc:
return {'ok': False, 'status': 'refused', 'submitted': False,
'reason': str(exc), 'retryAllowed': False, 'consumptionConfirmed': False}
diff --git a/session_peer_relay/worker.py b/session_peer_relay/worker.py
index 6100125..390e095 100644
--- a/session_peer_relay/worker.py
+++ b/session_peer_relay/worker.py
@@ -5,11 +5,33 @@
import session_peer as core
from .native import Policy, options, invoke_windows_codex
-from .framing import MAX_WORKER_BYTES, valid_text
+from .framing import MAX_WORKER_BYTES, valid_text, valid_delivery_context
+
+
+def delivery_context(binding, adapter, result):
+ """Use the executing registry and actual native resolution, never guessed homes."""
+ agent = binding['agent']
+ target = None
+ native_target = result.get('target')
+ if result.get('ok') is True:
+ if isinstance(native_target, dict):
+ identifier = native_target.get('pid' if agent == 'claude' else 'id')
+ if (agent == 'claude' and type(identifier) is int and identifier > 0
+ or agent != 'claude' and isinstance(identifier, str) and identifier):
+ # Claude's public CLI adapter uses an unprefixed PID, whereas
+ # this private cross-agent context always carries an agent tag.
+ target = agent + ':' + str(identifier).lower()
+ home = result.get('codexHome' if agent == 'codex' else 'antigravityHome') if agent != 'claude' else None
+ value = {'schemaVersion': 1, 'agent': agent, 'target': target, 'home': home,
+ 'capabilities': {key: getattr(adapter.capabilities, key)
+ for key in ('send', 'wake', 'wait', 'ack')}}
+ return value if valid_delivery_context(value) else None
def main():
attempted = False
+ explain = False
+ binding = adapter = None
try:
raw = sys.stdin.buffer.read(MAX_WORKER_BYTES + 1)
if len(raw) > MAX_WORKER_BYTES:
@@ -20,6 +42,9 @@ def main():
Policy({'targets': {'endpoint': binding}, 'peers': {}})
if op not in ('list', 'send', 'resolve'):
raise ValueError('invalid_operation')
+ if 'explain' in value and (op != 'resolve' or value['explain'] is not True):
+ raise ValueError('invalid_explain')
+ explain = value.get('explain') is True
adapter = core.AGENTS.get(binding['agent'])
args = options(binding)
args._peer_fingerprint = peer_fingerprint
@@ -72,7 +97,16 @@ def main():
result = {'ok': False, 'reason': 'native_outcome_unknown' if attempted else 'native_refused',
'status': 'unknown' if attempted else 'refused',
'consumptionConfirmed': False, 'retryAllowed': False}
+ if explain and adapter is not None:
+ context = delivery_context(binding, adapter, result)
+ if context is not None:
+ result['deliveryContext'] = context
encoded = json.dumps(result, ensure_ascii=False)
+ if len(encoded.encode()) > 60*1024 and 'deliveryContext' in result:
+ # Optional explanation metadata cannot erase independently usable
+ # native facts merely by taking the result over the legacy byte cap.
+ result.pop('deliveryContext', None)
+ encoded = json.dumps(result, ensure_ascii=False)
if len(encoded.encode()) > 60*1024:
encoded = json.dumps({'ok': False, 'reason': 'native_result_too_large',
'status': 'unknown', 'retryAllowed': False, 'consumptionConfirmed': False})
diff --git a/tests/core/test_delivery_explanation.py b/tests/core/test_delivery_explanation.py
new file mode 100644
index 0000000..2d38ab2
--- /dev/null
+++ b/tests/core/test_delivery_explanation.py
@@ -0,0 +1,421 @@
+"""Offline display-only previews preserve ordinary submission contracts."""
+import argparse
+import contextlib
+import io
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import session_peer as peer
+
+
+THREAD = "01900000-0000-7000-8000-000000000001"
+BODY = "PRIVATE-BODY-MUST-NOT-BE-DISPLAYED"
+
+
+class DeliveryExplanation(unittest.TestCase):
+ def invoke(self, *flags, target="worker"):
+ out, err = io.StringIO(), io.StringIO()
+ argv = ["send", "--to", target, "--no-from", "--no-reply-to",
+ "--message", BODY, *flags]
+ with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+ code = peer.main(argv)
+ return code, out.getvalue(), err.getvalue()
+
+ def test_invalid_combinations_fail_before_updates_input_or_dispatch(self):
+ variants = [[], ["--dry-run", "--json"],
+ ["--dry-run", "--output-format", "json"],
+ ["--dry-run", "--correlation-id", "00000000-0000-4000-8000-000000000001"],
+ ["--dry-run", "--request-ack"],
+ ["--dry-run", "--observe-delivery"],
+ ["--dry-run", "--wait-for", "acknowledged"],
+ ["--dry-run", "--_handoff-native-context", "private"],
+ ["--dry-run", "--_handoff-native-cutoff-ms", "1"],
+ ["--dry-run", "--_handoff-native-total-ms", "1"]]
+ for flags in variants:
+ diagnostic = io.StringIO()
+ with self.subTest(flags=flags), \
+ mock.patch.object(peer, "prepare_client_update") as updates, \
+ mock.patch.object(peer, "prepare_skill_updates") as skills, \
+ mock.patch.object(peer, "read_message") as read, \
+ mock.patch.object(peer, "cmd_send") as send, \
+ mock.patch.object(peer.subprocess, "Popen") as child, \
+ mock.patch.object(sys, "argv", ["session-peer", "send", "--to", "worker",
+ "--explain", *flags]), \
+ contextlib.redirect_stderr(diagnostic), self.assertRaises(SystemExit) as exc:
+ peer.main()
+ self.assertEqual(exc.exception.code, 2)
+ expected = ("--explain requires --dry-run" if not flags else
+ "--explain requires text output" if "--json" in flags or "--output-format" in flags else
+ "--explain does not support experimental handoff sends")
+ self.assertIn("session-peer: error: " + expected, diagnostic.getvalue())
+ for action in (updates, skills, read, send, child):
+ action.assert_not_called()
+
+ def test_claude_preview_reuses_dry_run_without_native_post_or_extra_discovery(self):
+ session = {"pid": 42, "name": "worker", "alive": True,
+ "reachable": True, "socket": "/fixture.sock"}
+ outputs = []
+ for extra in ([], ["--explain"]):
+ with mock.patch.object(peer, "discover", return_value=[session]) as discover, \
+ mock.patch.object(peer, "post_to_socket") as post, \
+ mock.patch.object(peer.subprocess, "run") as run:
+ code, out, err = self.invoke("--dry-run", *extra)
+ self.assertEqual(code, 0)
+ self.assertEqual(err, "")
+ discover.assert_called_once_with(include_unreachable=True)
+ post.assert_not_called()
+ run.assert_not_called()
+ outputs.append(out)
+ self.assertTrue(outputs[1].startswith(outputs[0]))
+ self.assertIn("Resolved target: Claude PID 42 (worker)", outputs[1])
+ self.assertIn("Wake: not requested; eligibility not checked", outputs[1])
+ self.assertNotIn(BODY, outputs[1])
+
+ def test_existing_json_and_absent_claude_fields_are_unchanged(self):
+ payload = {"ok": True, "target": {"pid": 42, "name": None},
+ "chars": 99, "dryRun": True}
+ before = json.dumps(payload, sort_keys=True)
+ with mock.patch.object(peer.LocalTransport, "execute", return_value=payload) as execute:
+ code, out, _ = self.invoke("--dry-run", "--json")
+ self.assertEqual(code, 0)
+ result = json.loads(out)
+ self.assertEqual(result, {"schemaVersion": 1, "ok": True, "host": peer.local_host(),
+ "command": "send", "target": {"pid": 42, "name": None},
+ "chars": 99, "dryRun": True})
+ self.assertEqual(json.dumps(payload, sort_keys=True), before)
+ self.assertNotIn("submitted", result)
+ self.assertNotIn("status", result)
+ execute.assert_called_once()
+
+ def test_codex_reported_home_and_wake_use_only_native_result(self):
+ for wake, fact, expected in (
+ (False, None, "not requested; eligibility not checked"),
+ (True, {"status": "validated", "cwd": "/hidden-cwd"},
+ "preflight passed for this dry run; no turn started"),
+ (True, None, "requested; preflight not confirmed")):
+ with self.subTest(wake=wake, fact=fact):
+ payload = {"ok": True, "target": {"agent": "codex", "id": THREAD},
+ "chars": 12, "dryRun": True, "status": "validated", "submitted": False,
+ "consumptionConfirmed": False, "codexHome": "/receiver/codex"}
+ if fact is not None:
+ payload["wake"] = fact
+ with mock.patch.object(peer, "queue_codex", return_value=payload) as queue:
+ code, out, _ = self.invoke("--dry-run", "--explain",
+ *(["--wake"] if wake else []), target="codex:" + THREAD)
+ self.assertEqual(code, 0)
+ queue.assert_called_once()
+ self.assertIn("Codex home: /receiver/codex", out)
+ self.assertIn("Wake: " + expected, out)
+ self.assertNotIn("/hidden-cwd", out)
+
+ def test_antigravity_dry_run_needs_no_target_or_dry_run_fields(self):
+ payload = {"ok": True, "status": "dry_run", "submitted": False,
+ "consumptionConfirmed": False, "generation": "not-for-display",
+ "antigravityHome": "/receiver/agy"}
+ with mock.patch.object(peer.AntigravityAdapter, "submit", return_value=payload) as submit:
+ code, out, _ = self.invoke("--dry-run", "--explain", target="antigravity:" + THREAD)
+ self.assertEqual(code, 0)
+ submit.assert_called_once()
+ self.assertIn("registered Antigravity bridge", out)
+ self.assertIn("Antigravity home: /receiver/agy", out)
+ self.assertNotIn("not-for-display", out)
+
+ def test_remote_preview_is_sender_only_and_does_not_guess_remote_home(self):
+ payload = {"ok": True, "target": {"agent": "codex", "id": THREAD},
+ "chars": 12, "dryRun": True, "status": "validated", "submitted": False,
+ "consumptionConfirmed": False}
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport") as transport:
+ transport.return_value.host = "receiver"
+ transport.return_value.ssh_opts = []
+ transport.return_value.execute.return_value = payload
+ code, out, _ = self.invoke("--host", "receiver", "--dry-run", "--explain",
+ "--codex-home", "/requested/home", target="codex:" + THREAD)
+ self.assertEqual(code, 0)
+ transport.assert_called_once()
+ transport.return_value.execute.assert_called_once()
+ argv = transport.return_value.execute.call_args.args[0]
+ self.assertNotIn("--explain", argv)
+ self.assertIn("--dry-run", argv)
+ self.assertIn("/requested/home", argv)
+ self.assertIn("Codex home: unknown (not reported)", out)
+ self.assertNotIn("Codex home: /requested/home", out)
+
+ def test_multiple_hosts_keep_partial_failure_attribution_and_streams(self):
+ good = mock.Mock(host="first", ssh_opts=[])
+ good.execute.return_value = {"target": {"pid": 42, "name": "worker"},
+ "chars": 12, "dryRun": True}
+ bad = mock.Mock(host="second", ssh_opts=[])
+ bad.execute.side_effect = peer.CcPeerError("fixture diagnostic", {
+ "status": "unknown", "submitted": None, "retryAllowed": False,
+ "error": BODY, "receipt": BODY})
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport", side_effect=[good, bad]):
+ code, out, err = self.invoke("--host", "first", "--host", "second",
+ "--dry-run", "--explain")
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertIn("on first", out)
+ self.assertNotIn("on second", out)
+ self.assertIn("on second", err)
+ self.assertIn("session-peer: second: fixture diagnostic", err)
+ self.assertIn("Do not automatically resend after an unknown outcome.", err)
+ self.assertNotIn(BODY, out + err)
+ good.execute.assert_called_once()
+ bad.execute.assert_called_once()
+
+ def test_local_failure_keeps_exit_and_sanitized_diagnostic(self):
+ error = peer.NoTargetError("no reachable session: fixture\x1b[31m")
+ with mock.patch.object(peer.LocalTransport, "execute", side_effect=error) as execute:
+ code, out, err = self.invoke("--dry-run", "--explain")
+ self.assertEqual(code, peer.EXIT_NO_TARGET)
+ self.assertEqual(out, "")
+ self.assertIn("Validation: not confirmed", err)
+ self.assertIn("fixture\\x1b[31m", err)
+ self.assertNotIn("\x1b", err)
+ execute.assert_called_once()
+
+ def test_unknown_local_and_ssh_exceptions_preserve_uncertainty(self):
+ for remote in (False, True):
+ with self.subTest(remote=remote), contextlib.ExitStack() as stack:
+ error = peer.CcPeerError("fixture unknown outcome", {
+ "status": "unknown", "submitted": None, "retryAllowed": False})
+ stack.enter_context(mock.patch.object(peer, "tailscale_status", return_value={}))
+ if remote:
+ transport = stack.enter_context(mock.patch.object(peer, "SshTransport"))
+ transport.return_value.host = "receiver"
+ transport.return_value.ssh_opts = []
+ execute = transport.return_value.execute
+ execute.side_effect = error
+ else:
+ execute = stack.enter_context(mock.patch.object(
+ peer.LocalTransport, "execute", side_effect=error))
+ code, out, err = self.invoke("--dry-run", "--explain",
+ *(["--host", "receiver"] if remote else []))
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertEqual(out, "")
+ self.assertIn("Validation: result unknown", err)
+ self.assertIn("inspect the target and existing evidence", err)
+ self.assertIn("Do not automatically resend", err)
+ self.assertNotIn("nothing submitted", err)
+ self.assertNotIn("Destination checks: not run", err)
+ execute.assert_called_once()
+
+ def test_local_transport_invalid_result_and_adapter_exception_stay_unknown(self):
+ for kwargs in ({"return_value": None}, {"side_effect": RuntimeError(BODY)}):
+ with self.subTest(case=tuple(kwargs)), \
+ mock.patch.object(peer.ClaudeAdapter, "submit", **kwargs) as submit:
+ code, out, err = self.invoke("--dry-run", "--explain")
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertEqual(out, "")
+ self.assertIn("Validation: result unknown", err)
+ self.assertIn("inspect the target and existing evidence", err)
+ self.assertIn("Do not automatically resend", err)
+ self.assertNotIn(BODY, err)
+ self.assertNotIn("nothing submitted", err)
+ submit.assert_called_once()
+
+ def test_returned_failed_payload_keeps_old_stdout_and_explanation_on_stderr(self):
+ for remote in (False, True):
+ for submitted in (None, False):
+ outputs = []
+ for explain in (False, True):
+ with self.subTest(remote=remote, submitted=submitted, explain=explain), \
+ contextlib.ExitStack() as stack:
+ payload = {"ok": False, "target": {"agent": "codex", "id": THREAD},
+ "dryRun": True, "chars": 12, "status": "unknown",
+ "submitted": submitted, "consumptionConfirmed": False}
+ stack.enter_context(mock.patch.object(peer, "tailscale_status", return_value={}))
+ if remote:
+ transport = stack.enter_context(mock.patch.object(peer, "SshTransport"))
+ transport.return_value.host = "receiver"
+ transport.return_value.ssh_opts = []
+ execute = transport.return_value.execute
+ execute.return_value = payload
+ else:
+ execute = stack.enter_context(mock.patch.object(
+ peer.LocalTransport, "execute", return_value=payload))
+ code, out, err = self.invoke("--dry-run", *(["--explain"] if explain else []),
+ *(["--host", "receiver"] if remote else []),
+ target="codex:" + THREAD)
+ self.assertEqual(code, peer.EXIT_ERROR)
+ execute.assert_called_once()
+ outputs.append(out)
+ if explain:
+ self.assertIn("Delivery preview\n", err)
+ self.assertIn("Validation: result unknown", err)
+ self.assertNotIn("nothing submitted", err)
+ else:
+ self.assertEqual(err, "")
+ self.assertEqual(outputs[0], outputs[1])
+
+ def test_common_multihost_failures_do_not_claim_destination_checks(self):
+ for stage in ("read_message", "validate_ssh_identity_options"):
+ with self.subTest(stage=stage), \
+ mock.patch.object(peer, stage, side_effect=peer.CcPeerError("common command failure")), \
+ mock.patch.object(peer, "SshTransport") as transport, \
+ mock.patch.object(peer.LocalTransport, "execute") as local:
+ code, out, err = self.invoke("--host", "first", "--host", "second",
+ "--dry-run", "--explain")
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertEqual(out, "")
+ self.assertEqual(err.count("Delivery preview\n"), 1)
+ self.assertIn("Destination checks: not run", err)
+ self.assertIn("Validation: not confirmed", err)
+ self.assertNotIn("passed for this dry run", err)
+ transport.assert_not_called()
+ local.assert_not_called()
+
+ def test_second_host_unexpected_error_names_only_current_destination(self):
+ good = mock.Mock(host="first", ssh_opts=[])
+ good.execute.return_value = {"target": {"pid": 42, "name": "worker"},
+ "chars": 12, "dryRun": True}
+ bad = mock.Mock(host="second", ssh_opts=[])
+ bad.execute.side_effect = OSError("fixture transport error")
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport", side_effect=[good, bad]):
+ code, out, err = self.invoke("--host", "first", "--host", "second",
+ "--dry-run", "--explain")
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertIn("on first", out)
+ self.assertIn("on second", err)
+ self.assertNotIn("first", err)
+ self.assertNotIn("Destination checks: not run", err)
+ self.assertIn("Validation: result unknown", err)
+ self.assertIn("inspect the target and existing evidence", err)
+ good.execute.assert_called_once()
+ bad.execute.assert_called_once()
+
+ def test_remote_submission_evidence_is_not_presented_as_validated_dry_run(self):
+ for agent in ("claude", "codex"):
+ for evidence in ({"submitted": True}, {"dryRun": False},
+ {"status": "queued"}, {"status": "posted"}):
+ with self.subTest(agent=agent, evidence=evidence):
+ payload = {"ok": True, "target": {"pid": 42, "name": "worker"}
+ if agent == "claude" else {"agent": "codex", "id": THREAD},
+ "chars": 12, "dryRun": True, **evidence}
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport") as transport:
+ transport.return_value.host = "receiver"
+ transport.return_value.ssh_opts = []
+ transport.return_value.execute.return_value = payload
+ code, out, err = self.invoke("--host", "receiver", "--dry-run", "--explain",
+ target="worker" if agent == "claude" else "codex:" + THREAD)
+ self.assertEqual(code, 0)
+ self.assertNotIn("passed for this dry run", out + err)
+ self.assertIn("unexpected submission evidence; dry-run not confirmed", err)
+ self.assertIn("Reported ", err)
+ self.assertIn("Do not automatically resend", err)
+ self.assertNotIn("nothing submitted", out + err)
+ transport.return_value.execute.assert_called_once()
+
+ def test_display_allowlist_does_not_mutate_or_reflect_payload_secrets(self):
+ args = argparse.Namespace(to="codex:" + THREAD, host=[], wake=False,
+ codex_home="/sender/home")
+ payload = {"codexHome": "/receiver/\n\x1b", "target": {"id": THREAD + "\r"},
+ "error": BODY, "message": BODY, "receipt": BODY,
+ "handoff": {"capability": BODY}, "reason": BODY,
+ "replyRoute": {"uri": BODY}}
+ original = json.dumps(payload)
+ rendered = peer.render_delivery_explanation(args, payload, host="host\x00")
+ self.assertNotIn(BODY, rendered)
+ for char in ("\x1b", "\r", "\x00"):
+ self.assertNotIn(char, rendered)
+ self.assertIn("\\x0a\\x1b", rendered)
+ self.assertEqual(json.dumps(payload), original)
+
+ def test_ok_true_does_not_promote_unknown_evidence_to_validation(self):
+ args = argparse.Namespace(to="codex:" + THREAD, host=[], wake=False)
+ for evidence in ({"status": "unknown"}, {"submitted": None}):
+ rendered = peer.render_delivery_explanation(args, {"ok": True, **evidence})
+ self.assertIn("Validation: result unknown", rendered)
+ self.assertNotIn("passed for this dry run", rendered)
+
+ def test_connection_identity_failure_reason_preserves_uncertainty(self):
+ args = argparse.Namespace(to="worker", host=["receiver"], wake=False)
+ rendered = peer.render_delivery_explanation(args, {
+ "reason": "ssh_identity_evidence_unavailable", "dryRun": True}, failed=True)
+ self.assertIn("Validation: result unknown", rendered)
+ self.assertIn("inspect the target and existing evidence", rendered)
+ self.assertNotIn("nothing submitted", rendered)
+
+ def test_passed_validation_requires_adapter_dry_run_evidence(self):
+ args = argparse.Namespace(to="worker", host=[], wake=False)
+ for result, passed in (({"ok": True}, False), ({"ok": True, "submitted": False}, False),
+ ({"ok": True, "dryRun": True}, True),
+ ({"ok": True, "status": "validated"}, True),
+ ({"ok": True, "status": "dry_run"}, True)):
+ with self.subTest(result=result):
+ rendered = peer.render_delivery_explanation(args, result)
+ if passed:
+ self.assertIn("Validation: passed for this dry run", rendered)
+ else:
+ self.assertIn("Validation: not confirmed (dry-run evidence not reported)", rendered)
+ self.assertNotIn("passed for this dry run", rendered)
+
+ def test_ambiguous_or_submitted_response_does_not_claim_no_turn_started(self):
+ args = argparse.Namespace(to="codex:" + THREAD, host=[], wake=True)
+ for evidence in ({"status": "unknown"}, {"submitted": None},
+ {"submitted": True}, {"dryRun": False}, {"status": "queued"}):
+ rendered = peer.render_delivery_explanation(args, {
+ "ok": True, "wake": {"status": "validated"}, **evidence})
+ self.assertIn("Wake: requested; preflight not confirmed", rendered)
+ self.assertNotIn("no turn started", rendered)
+
+ def test_unresolved_reply_uri_is_not_reflected(self):
+ for scheme in ("session-peer", "SeSsIoN-PeEr"):
+ args = argparse.Namespace(to=scheme + "://v1/reply?secret=" + BODY,
+ host=[], wake=False)
+ rendered = peer.render_delivery_explanation(args, {}, failed=True)
+ self.assertNotIn(BODY, rendered)
+ self.assertIn("Delivery method: unknown (not resolved)", rendered)
+ self.assertNotIn("Claude Code inbox", rendered)
+
+ @unittest.skipIf(peer.IS_WINDOWS, "private POSIX message-file input only")
+ def test_private_message_file_remains_an_ordinary_supported_input(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "private-message"
+ path.write_text(BODY, encoding="utf-8")
+ path.chmod(0o600)
+ out = io.StringIO()
+ payload = {"ok": True, "target": {"pid": 42, "name": "worker"},
+ "chars": 12, "dryRun": True}
+ with mock.patch.object(peer.LocalTransport, "execute", return_value=payload) as execute, \
+ contextlib.redirect_stdout(out):
+ code = peer.main(["send", "--to", "worker", "--no-from", "--no-reply-to",
+ "--message-file", str(path), "--dry-run", "--explain"])
+ self.assertEqual(code, 0)
+ self.assertEqual(execute.call_args.args[-1], BODY)
+ self.assertNotIn(BODY, out.getvalue())
+
+ def test_windows_private_message_file_refusal_remains_before_destination_checks(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "private-message"
+ path.write_text(BODY, encoding="utf-8")
+ out, err = io.StringIO(), io.StringIO()
+ with mock.patch.object(peer, "IS_WINDOWS", True), \
+ mock.patch.object(peer.LocalTransport, "execute") as execute, \
+ mock.patch.object(peer, "SshTransport") as ssh, \
+ mock.patch.object(peer.subprocess, "Popen") as child, \
+ contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+ with self.assertRaises(peer.CcPeerError) as refusal:
+ peer.handoff_private_stat(path)
+ self.assertEqual(refusal.exception.details["reason"], "handoff_platform_unsupported")
+ code = peer.main(["send", "--to", "worker", "--no-from", "--no-reply-to",
+ "--message-file", str(path), "--dry-run", "--explain"])
+ self.assertEqual(code, peer.EXIT_ERROR)
+ self.assertEqual(out.getvalue(), "")
+ self.assertIn("session-peer: Handoff operation refused", err.getvalue())
+ self.assertNotIn("handoff_platform_unsupported", err.getvalue())
+ self.assertIn("Destination checks: not run", err.getvalue())
+ self.assertNotIn(BODY, err.getvalue())
+ execute.assert_not_called()
+ ssh.assert_not_called()
+ child.assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/core/test_explanation_context.py b/tests/core/test_explanation_context.py
new file mode 100644
index 0000000..1e9703e
--- /dev/null
+++ b/tests/core/test_explanation_context.py
@@ -0,0 +1,167 @@
+"""Offline facts-only explanation context and privacy boundaries."""
+import argparse
+import copy
+import io
+import unittest
+from unittest import mock
+
+import session_peer as peer
+
+
+class ExplanationContextTests(unittest.TestCase):
+ def args(self, **kwargs):
+ return argparse.Namespace(**{**dict(to="worker", host=[], device=None, wake=False), **kwargs})
+
+ def context(self, **kwargs):
+ return {**dict(schemaVersion=1, agent="codex", target="codex:00000000-0000-4000-8000-000000000001", home="/receiver/home",
+ capabilities=dict(send=True, wake=True, wait=False, ack=False)), **kwargs}
+
+ def test_context_closed_versioned_and_boolean_contract(self):
+ valid = self.context()
+ self.assertEqual(peer.validated_delivery_context(valid), valid)
+ bad = [self.context(schemaVersion=True), self.context(schemaVersion=2), self.context(agent="other"),
+ self.context(target="session-peer://secret"), self.context(home=1),
+ self.context(capabilities={"send": 1, "wake": True, "wait": False, "ack": False}),
+ {**valid, "capability": "SECRET"}]
+ for value in bad:
+ with self.subTest(value=value):
+ self.assertIsNone(peer.validated_delivery_context(value))
+ normalized = peer.validated_delivery_context(valid)
+ normalized["capabilities"]["send"] = False
+ self.assertTrue(valid["capabilities"]["send"])
+
+ def test_paired_missing_context_never_guesses_from_alias_or_sender(self):
+ args = self.args(device="fingerprint", to="codex:sender-alias", codex_home="/sender/home")
+ result = {"dryRun": True, "codexHome": "/unvalidated/home", "target": {"id": "unvalidated"}}
+ facts = peer.build_delivery_explanation(args, result)
+ self.assertIsNone(facts["agent"])
+ self.assertIsNone(facts["capabilities"])
+ rendered = peer.render_delivery_explanation(args, result)
+ for secret in ("/sender/home", "/unvalidated/home", "codex:sender-alias"):
+ self.assertNotIn(secret, rendered)
+ self.assertIn("Selected agent / home: unknown", rendered)
+ self.assertIn("Receipt ACK: unknown", rendered)
+
+ def test_paired_uses_only_receiver_declarations(self):
+ args = self.args(device="fingerprint", to="opaque-alias", _delivery_explanation_context=self.context())
+ rendered = peer.render_delivery_explanation(args, {"dryRun": True, "codexHome": "/wrong/home"})
+ self.assertIn("codex:00000000-0000-4000-8000-000000000001", rendered)
+ self.assertIn("/receiver/home", rendered)
+ self.assertNotIn("/wrong/home", rendered)
+ self.assertIn("Wake capability: supported (declared; eligibility unverified)", rendered)
+ self.assertIn("Wake: not requested; eligibility not checked", rendered)
+
+ def test_declarations_never_become_observed_eligibility(self):
+ rendered = peer.render_delivery_explanation(self.args(to="codex:id"), {"ok": True})
+ self.assertIn("Submission capability: supported (declared; eligibility unverified)", rendered)
+ self.assertIn("Validation: not confirmed", rendered)
+ self.assertIn("Task completion: unknown", rendered)
+ self.assertIn("Last delivery observation: unknown", rendered)
+
+ def test_paired_route_uses_only_existing_report_not_requested_auto(self):
+ args = self.args(device="fingerprint", to="opaque-alias", device_route="auto")
+ for route in ("direct", "relay", None, "auto"):
+ output = peer.render_delivery_explanation(args, {"dryRun": True, "route": route})
+ self.assertIn("Selected paired route: " + (route + " (reported)" if route in ("direct", "relay") else "unknown (not reported)"), output)
+
+ def test_doctor_reuses_only_supplied_diagnostics(self):
+ result = {"codex": {"status": "available", "selectedHome": "/receiver/\x1bhome"},
+ "claude": {"status": "PRIVATE-ERROR"},
+ "capabilities": {"agents": {"codex": {"send": True, "wake": True, "wait": False, "ack": False}}}}
+ original = copy.deepcopy(result)
+ with mock.patch.object(peer, "doctor_payload", side_effect=AssertionError("new diagnostic")), \
+ mock.patch.object(peer, "collect_listing", side_effect=AssertionError("new discovery")), \
+ mock.patch.object(peer.subprocess, "run", side_effect=AssertionError("new process")):
+ rendered = peer.render_delivery_explanation(self.args(), result, mode="doctor", host="worker\n")
+ self.assertIn("codex availability: available", rendered)
+ self.assertIn("not per-target readiness", rendered)
+ self.assertIn("Selected home: /receiver/\\x1bhome", rendered)
+ self.assertNotIn("PRIVATE-ERROR", rendered)
+ self.assertNotIn("\x1b", rendered)
+ self.assertIn("Destination host: worker\\x0a", rendered)
+ self.assertEqual(result, original)
+
+ def test_doctor_unknown_capabilities_not_filled_from_local_registry(self):
+ rendered = peer.render_delivery_explanation(self.args(), {"codex": {"status": "available"}}, mode="doctor")
+ self.assertIn("Wake: unknown (not reported)", rendered)
+ self.assertNotIn(": supported (declared", rendered)
+
+ def test_existing_intent_keeps_receipt_latest_wait_and_completion_separate(self):
+ result = {"handoff": {"submission": {"status": "submitted"},
+ "observation": {"status": "unsupported", "injectionObserved": False},
+ "ack": {"status": "acknowledged", "assurance": "token_possession", "receivedAtUtcMs": 12345, "late": True},
+ "wait": {"status": "satisfied", "for": "acknowledged"}}}
+ rendered = peer.render_delivery_explanation(self.args(), result, mode="existing_intent")
+ self.assertIn("Explicit receipt: acknowledged", rendered)
+ self.assertIn("Latest stored wait: satisfied", rendered)
+ self.assertIn("Receipt assurance: token_possession", rendered)
+ self.assertIn("Receipt late relative to first wait: true", rendered)
+ self.assertIn("UTC milliseconds): 12345", rendered)
+ self.assertIn("Task completion: unknown", rendered)
+ self.assertIn("Stored evidence may be stale", rendered)
+
+ def test_non_acknowledged_timestamp_cannot_become_receipt_observation(self):
+ result = {"handoff": {"ack": {"status": "unsupported", "receivedAtUtcMs": 12345, "late": True}}}
+ rendered = peer.render_delivery_explanation(self.args(), result, mode="existing_intent")
+ self.assertIn("Explicit receipt: unsupported", rendered)
+ self.assertNotIn("12345", rendered)
+ self.assertNotIn("late relative", rendered)
+
+ def test_existing_intent_history_failure_is_unknown_not_new_observation(self):
+ result = {"error": "SECRET", "handoffQuery": {"status": "unknown", "context": "history_unavailable"}}
+ with mock.patch.object(peer, "HandoffLedger", side_effect=AssertionError("new ledger read")), \
+ mock.patch.object(peer, "handoff_ensure_collector", side_effect=AssertionError("new collector")):
+ rendered = peer.render_delivery_explanation(self.args(), result, mode="existing_intent", failed=True)
+ self.assertIn("Submission: unknown", rendered)
+ self.assertNotIn("SECRET", rendered)
+ self.assertIn("Do not automatically resend", rendered)
+
+ def test_private_fields_and_control_characters_never_escape_allowlist(self):
+ result = {"handoff": {"capability": "SECRET", "targetGeneration": "SECRET", "native": "SECRET",
+ "ack": {"status": "pending", "receiptId": "SECRET", "assurance": "SECRET"}},
+ "message": "SECRET", "error": "SECRET", "reason": "SECRET"}
+ rendered = peer.render_delivery_explanation(self.args(), result, mode="existing_intent")
+ self.assertNotIn("SECRET", rendered)
+
+ def test_emitter_mode_and_stream_are_backward_compatible(self):
+ output = io.StringIO()
+ peer.emit_delivery_explanation(self.args(), {}, stream=output, mode="doctor")
+ self.assertIn("Delivery capabilities", output.getvalue())
+
+ def test_malformed_status_values_are_unknown(self):
+ rendered = peer.render_delivery_explanation(self.args(), {"codex": {"status": []}}, mode="doctor")
+ self.assertIn("codex availability: unknown", rendered)
+ rendered = peer.render_delivery_explanation(self.args(), {"handoff": {"ack": {"status": []}}}, mode="existing_intent")
+ self.assertIn("Explicit receipt: unknown", rendered)
+
+
+ def test_bidi_is_escaped_only_in_explanations(self):
+ home = "/receiver/\u202eprofile"
+ args = self.args(device="fingerprint", to="alias", _delivery_explanation_context=self.context(home=home))
+ output = peer.render_delivery_explanation(args, {"dryRun": True})
+ self.assertNotIn("\u202e", output)
+ self.assertIn("\\u202e", output)
+ self.assertEqual(args._delivery_explanation_context["home"], home)
+
+ def test_unavailable_history_is_not_presented_as_read_observation(self):
+ output = peer.render_delivery_explanation(self.args(), {"handoffQuery": {"context": "ledger_missing"}},
+ mode="existing_intent", failed=True)
+ self.assertIn("history not read", output)
+ self.assertIn("History query: ledger_missing", output)
+ self.assertNotIn("stored observation", output)
+
+ def test_receipt_time_is_recorded_not_assumed_physical_arrival(self):
+ public = {"ack": {"status": "acknowledged", "assurance": "operator_confirmed",
+ "receivedAtUtcMs": 123, "late": True}}
+ output = peer.render_delivery_explanation(self.args(), {"handoff": public}, mode="existing_intent")
+ self.assertIn("Receipt recorded at (stored UTC milliseconds): 123", output)
+ self.assertNotIn("Receipt observed at", output)
+ stream = io.StringIO()
+ with mock.patch.object(peer.time, "time", side_effect=OSError("private clock diagnostic")):
+ peer.emit_delivery_explanation(self.args(), {"handoff": public}, mode="existing_intent", stream=stream)
+ self.assertIn("Explicit receipt: acknowledged", stream.getvalue())
+ self.assertNotIn("private clock", stream.getvalue())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/core/test_explanation_modes.py b/tests/core/test_explanation_modes.py
new file mode 100644
index 0000000..1770ad1
--- /dev/null
+++ b/tests/core/test_explanation_modes.py
@@ -0,0 +1,137 @@
+"""New modes reuse existing facts; no extra probes or native submission."""
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest import mock
+
+import session_peer as peer
+
+
+class ExplanationModes(unittest.TestCase):
+ def invoke(self, argv):
+ out, err = io.StringIO(), io.StringIO()
+ with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+ code = peer.main(argv)
+ return code, out.getvalue(), err.getvalue()
+
+ def test_doctor_reuses_exactly_one_existing_payload_and_keeps_default_json(self):
+ payload = {"status": "partial", "claude": {"status": "available"},
+ "capabilities": {"agents": {"claude": {"send": True, "wake": False, "wait": False, "ack": False}}}}
+ with mock.patch.object(peer, "doctor_payload", return_value=payload) as doctor, \
+ mock.patch.object(peer, "render_doctor", return_value="Existing diagnostics"), \
+ mock.patch.object(peer.subprocess, "Popen", side_effect=AssertionError("extra process")):
+ code, plain, _ = self.invoke(["doctor", "--no-update-notice"])
+ self.assertEqual(code, 0)
+ self.assertNotIn("Delivery capabilities", plain)
+ code, explained, _ = self.invoke(["doctor", "--explain", "--no-update-notice"])
+ self.assertEqual(code, 0)
+ self.assertIn("Delivery capabilities", explained)
+ self.assertIn("not per-target readiness", explained)
+ self.assertIn("not a delivery observation", explained)
+ self.assertEqual(doctor.call_count, 2)
+ code, encoded, _ = self.invoke(["doctor", "--json", "--no-update-notice"])
+ self.assertEqual(code, 0)
+ self.assertNotIn("deliveryExplanation", json.loads(encoded))
+ self.assertNotIn("Delivery capabilities", encoded)
+
+ def test_remote_doctor_explains_each_existing_response_without_extra_ssh(self):
+ payload = {"status": "partial", "capabilities": {"agents": {}}, "host": "remote"}
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport") as transport, \
+ mock.patch.object(peer, "render_doctor", return_value="Existing remote diagnostics"):
+ transport.return_value.host = "resolved"
+ transport.return_value.ssh_opts = []
+ transport.return_value.execute.return_value = payload
+ code, output, _ = self.invoke(["doctor", "--host", "one", "--host", "two", "--explain", "--no-update-notice"])
+ self.assertEqual(code, 0)
+ self.assertEqual(transport.call_count, 2)
+ self.assertEqual(transport.return_value.execute.call_count, 2)
+ self.assertTrue(all("--explain" not in c.args[0] for c in transport.return_value.execute.call_args_list))
+ self.assertEqual(output.count("Delivery capabilities"), 2)
+
+ def test_new_json_combinations_fail_before_advisory_or_destination_work(self):
+ for argv in (["doctor", "--explain", "--json"],
+ ["handoff", "status", "--correlation-id", "00000000-0000-4000-8000-000000000001", "--explain", "--json"]):
+ with mock.patch.object(peer, "prepare_client_update") as updates, \
+ mock.patch.object(peer, "prepare_skill_updates") as skills, \
+ mock.patch.object(peer, "HandoffLedger") as ledger, \
+ mock.patch.object(peer, "doctor_payload") as doctor, \
+ mock.patch.object(peer.sys, "argv", ["session-peer", *argv]), \
+ contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error:
+ peer.main()
+ self.assertEqual(error.exception.code, 2)
+ for operation in (updates, skills, ledger, doctor):
+ operation.assert_not_called()
+
+ def test_remote_doctor_partial_failure_still_explains_each_host(self):
+ payload = {"status": "partial", "capabilities": {"agents": {}}}
+ with mock.patch.object(peer, "tailscale_status", return_value={}), \
+ mock.patch.object(peer, "SshTransport") as transport, \
+ mock.patch.object(peer, "render_doctor", return_value="Existing diagnostics"):
+ transport.return_value.host = "resolved"
+ transport.return_value.ssh_opts = []
+ transport.return_value.execute.side_effect = [payload, peer.CcPeerError("fixture transport failed")]
+ code, output, err = self.invoke(["doctor", "--host", "one", "--host", "two", "--explain", "--no-update-notice"])
+ self.assertEqual(code, 1)
+ self.assertEqual(transport.return_value.execute.call_count, 2)
+ self.assertEqual(output.count("Delivery capabilities"), 1)
+ self.assertEqual(err.count("Delivery capabilities"), 1)
+ self.assertIn("unknown (not reported", err)
+
+ @unittest.skipIf(peer.IS_WINDOWS or peer.fcntl is None, "local POSIX ledger")
+ def test_status_explain_clock_is_readonly_and_legacy_failure_never_retries(self):
+ epoch = "00000000-0000-4000-8000-000000000001"
+ correlation = "00000000-0000-4000-8000-000000000002"
+ public = {"submission": {"status": "submitted"}, "observation": {"status": "unsupported"},
+ "ack": {"status": "pending"}, "wait": {"status": "not_requested"}}
+ record = {"capability": {"revoked": False}, "ack": None}
+ for response in ({"ok": True, "clockEpoch": epoch}, peer.handoff_error("receipt_operation_refused")):
+ with mock.patch.object(peer.HandoffLedger, "record", return_value=(epoch, record)) as read, \
+ mock.patch.object(peer.HandoffLedger, "_save", side_effect=AssertionError("history write")), \
+ mock.patch.object(peer, "handoff_public", return_value=public) as projection, \
+ mock.patch.object(peer, "handoff_ipc", side_effect=response if isinstance(response, Exception) else None,
+ return_value=response) as ipc, \
+ mock.patch.object(peer, "handoff_collector", side_effect=AssertionError("collector start")):
+ code, _, _ = self.invoke(["handoff", "status", "--correlation-id", correlation, "--explain"])
+ self.assertEqual(code, 0)
+ read.assert_called_once_with(correlation, read_only=True)
+ self.assertEqual(ipc.call_count, 1)
+ self.assertEqual(ipc.call_args.args[1], {"op": "clock", "readOnly": True})
+ self.assertEqual(projection.call_args.args[2], None if isinstance(response, Exception) else epoch)
+
+ @unittest.skipIf(peer.IS_WINDOWS or peer.fcntl is None, "local POSIX ledger")
+ def test_handoff_explanation_never_persists_retention_or_creates_authority(self):
+ with tempfile.TemporaryDirectory(prefix="codex-explain-", dir=os.environ.get("SESSION_PEER_TEST_TMP")) as directory:
+ root = Path(directory).resolve() / "ledger"
+ ledger = peer.HandoffLedger(root)
+ ledger.initialize()
+ _, record = ledger.prepare({"agent": "claude", "destination": ["local"],
+ "target": str(os.getpid()), "home": None, "payloadDigest": "a" * 64}, "tg1:" + "b" * 64)
+ before = {p.name: (p.read_bytes(), p.stat().st_ino, p.stat().st_mtime_ns) for p in root.iterdir()}
+ with mock.patch.object(peer, "handoff_root", return_value=root), \
+ mock.patch.object(peer.HandoffLedger, "_save", side_effect=AssertionError("history write")), \
+ mock.patch.object(peer, "handoff_collector", side_effect=AssertionError("collector launch")), \
+ mock.patch.object(peer, "handoff_ipc", side_effect=AssertionError("unneeded IPC")), \
+ mock.patch.object(peer, "post_to_socket", side_effect=AssertionError("native post")), \
+ mock.patch.object(peer.subprocess, "Popen", side_effect=AssertionError("child process")):
+ code, output, _ = self.invoke(["handoff", "status", "--correlation-id", record["id"], "--explain"])
+ self.assertEqual(code, 0)
+ self.assertIn("existing intent snapshot", output)
+ self.assertIn("Task completion: unknown", output)
+ self.assertIn("Submission: not_attempted", output)
+ after = {p.name: (p.read_bytes(), p.stat().st_ino, p.stat().st_mtime_ns) for p in root.iterdir()}
+ self.assertTrue(before == after, "explanation mutated history")
+
+ @unittest.skipIf(peer.IS_WINDOWS or peer.fcntl is None, "local POSIX ledger")
+ def test_missing_handoff_explanation_does_not_initialize_history(self):
+ with tempfile.TemporaryDirectory(prefix="codex-explain-", dir=os.environ.get("SESSION_PEER_TEST_TMP")) as directory:
+ root = Path(directory).resolve() / "missing"
+ with mock.patch.object(peer, "handoff_root", return_value=root):
+ code, _, err = self.invoke(["handoff", "status", "--correlation-id", "00000000-0000-4000-8000-000000000001", "--explain"])
+ self.assertEqual(code, 1)
+ self.assertIn("unknown", err)
+ self.assertFalse(root.exists())
diff --git a/tests/relay/test_delivery_explanation.py b/tests/relay/test_delivery_explanation.py
new file mode 100644
index 0000000..bdf3ebd
--- /dev/null
+++ b/tests/relay/test_delivery_explanation.py
@@ -0,0 +1,210 @@
+"""Offline paired explanation negotiation; no native or model submission."""
+import contextlib
+import copy
+import io
+import json
+import sys
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+from tests.relay import test_native_relay as platform_guard
+from session_peer_relay import app, cli, worker
+from session_peer_relay.framing import valid_delivery_context, worker_frame
+
+UUID = '01900000-0000-7000-8000-000000000001'
+CONTEXT = {'schemaVersion': 1, 'agent': 'codex', 'target': 'codex:' + UUID,
+ 'home': '/actual', 'capabilities': {'send': True, 'wake': True, 'wait': False, 'ack': False}}
+
+
+class ContextValidation(unittest.TestCase):
+ def test_strict_bounded_optional_context(self):
+ self.assertTrue(valid_delivery_context(CONTEXT))
+ self.assertTrue(valid_delivery_context({**CONTEXT, 'target': None, 'home': None}))
+ for change in ({'schemaVersion': True}, {'schemaVersion': 2}, {'agent': 'other'},
+ {'home': 'relative'}, {'home': '/' + 'x' * 4096}, {'home': '/\ud800'},
+ {'target': 'claude:1'}, {'target': 'codex:alias'},
+ {'capabilities': {**CONTEXT['capabilities'], 'wake': 1}}, {'extra': True}):
+ self.assertFalse(valid_delivery_context({**CONTEXT, **change}), change)
+
+ def test_context_only_explicit_resolve_and_actual_returned_facts(self):
+ binding = {'agent': 'codex', 'target': 'codex:' + UUID, 'codexHome': '/guessed'}
+ for explain in (False, True):
+ output = io.StringIO()
+ native = {'ok': True, 'target': {'agent': 'codex', 'id': UUID}, 'codexHome': '/actual'}
+ with mock.patch.object(sys, 'stdin', SimpleNamespace(buffer=io.BytesIO(
+ worker_frame(binding, 'resolve', 'fixture', explain=explain)))), \
+ mock.patch.object(worker.core, 'codex_executable', return_value='fixture'), \
+ mock.patch.object(worker.core.LocalTransport, 'execute', return_value=native) as execute, \
+ contextlib.redirect_stdout(output):
+ worker.main()
+ result = json.loads(output.getvalue())
+ self.assertEqual(execute.call_count, 1)
+ if explain:
+ self.assertEqual(result['deliveryContext'], CONTEXT)
+ else:
+ self.assertNotIn('deliveryContext', result)
+ no_facts = worker.delivery_context(binding, worker.core.AGENTS.get('codex'), {'ok': False})
+ self.assertIsNone(no_facts['target'])
+ self.assertIsNone(no_facts['home'])
+ agy = worker.delivery_context({'agent': 'antigravity', 'target': 'antigravity:' + UUID},
+ worker.core.AGENTS.get('antigravity'),
+ {'ok': True, 'antigravityHome': '/actual'})
+ self.assertIsNone(agy['target'])
+ self.assertEqual(agy['home'], '/actual')
+ claude = worker.delivery_context({'agent': 'claude', 'target': 'worker'},
+ worker.core.AGENTS.get('claude'),
+ {'ok': True, 'target': {'pid': 42, 'name': 'worker'}})
+ self.assertEqual(claude['target'], 'claude:42')
+ self.assertIsNone(claude['home'])
+
+ def test_invoke_strips_context_and_preserves_native_facts(self):
+ for explain in (False, True):
+ for context in (CONTEXT, {**CONTEXT, 'schemaVersion': 99}, None):
+ args = SimpleNamespace(device='a' * 64, command='send', dry_run=True, json=True, explain=explain)
+ native = {'ok': True, 'target': {'id': UUID}, 'codexHome': '/actual',
+ 'route': 'direct', 'deliveryContext': copy.deepcopy(context)}
+ with mock.patch.object(cli, 'core_exchange', mock.AsyncMock(return_value=native)), \
+ mock.patch.object(cli.core, 'emit') as emit, \
+ mock.patch.object(cli.core, 'emit_delivery_explanation') as explanation:
+ self.assertEqual(cli.invoke_core(args), 0)
+ emitted = emit.call_args.args[1]
+ self.assertNotIn('deliveryContext', emitted)
+ self.assertEqual(emitted['target'], {'id': UUID})
+ self.assertEqual(emitted['codexHome'], '/actual')
+ self.assertEqual(hasattr(args, '_delivery_explanation_context'), explain and context == CONTEXT)
+ if explain:
+ explanation.assert_called_once_with(args, emitted, failed=False)
+ else:
+ explanation.assert_not_called()
+
+ def test_optional_worker_context_cannot_displace_valid_native_result_at_cap(self):
+ binding = {'agent': 'codex', 'target': 'codex:' + UUID, 'codexHome': '/configured'}
+ native = {'ok': True, 'target': {'agent': 'codex', 'id': UUID}, 'codexHome': '/actual',
+ 'padding': 'x' * (60 * 1024 - 280)}
+ output = io.StringIO()
+ with mock.patch.object(sys, 'stdin', SimpleNamespace(buffer=io.BytesIO(
+ worker_frame(binding, 'resolve', 'fixture', explain=True)))), \
+ mock.patch.object(worker.core, 'codex_executable', return_value='fixture'), \
+ mock.patch.object(worker.core.LocalTransport, 'execute', return_value=native), \
+ contextlib.redirect_stdout(output):
+ worker.main()
+ result = json.loads(output.getvalue())
+ self.assertTrue(result['ok'])
+ self.assertEqual(result['target']['id'], UUID)
+ self.assertEqual(result['padding'], native['padding'])
+ self.assertNotIn('deliveryContext', result)
+
+ def test_default_json_has_no_context_or_extra_output(self):
+ args = SimpleNamespace(device='a' * 64, command='send', dry_run=True, json=True)
+ native = {'ok': True, 'target': {'id': UUID}, 'codexHome': '/actual',
+ 'deliveryContext': CONTEXT}
+ output = io.StringIO()
+ with mock.patch.object(cli, 'core_exchange', mock.AsyncMock(return_value=native)), \
+ mock.patch.object(cli.core, 'emit_delivery_explanation') as explanation, \
+ contextlib.redirect_stdout(output):
+ self.assertEqual(cli.invoke_core(args), 0)
+ result = json.loads(output.getvalue())
+ self.assertNotIn('deliveryContext', result)
+ self.assertEqual(result['target'], {'id': UUID})
+ self.assertEqual(result['codexHome'], '/actual')
+ self.assertFalse(hasattr(args, '_delivery_explanation_context'))
+ explanation.assert_not_called()
+
+ def test_real_cli_reports_device_validation_stage(self):
+ # Exercise main -> cmd_send -> relay -> actual explanation renderer.
+ argv = ['send', '--device', 'a' * 64, '--to', 'alias', '--message', 'fixture',
+ '--dry-run', '--explain', '--no-from', '--no-update-notice']
+ for result, expected in (({'ok': True, 'status': 'validated', 'dryRun': True,
+ 'deliveryContext': CONTEXT}, 'Validation: passed for this dry run'),
+ ({'ok': False, 'status': 'unknown', 'retryAllowed': False},
+ 'Validation: result unknown')):
+ out, err = io.StringIO(), io.StringIO()
+ with mock.patch.object(cli, 'Store'), \
+ mock.patch.object(cli, 'device_credential', return_value=None), \
+ mock.patch.object(cli, 'exchange', mock.AsyncMock(return_value=result)) as exchange, \
+ contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+ code = cli.core.main(argv)
+ self.assertEqual(code, 0 if result['ok'] else 1)
+ self.assertIn(expected, out.getvalue() + err.getvalue())
+ self.assertNotIn('Destination checks: not run', out.getvalue() + err.getvalue())
+ exchange.assert_awaited_once()
+ for rejected in ('message', 'credential', 'options'):
+ out, err = io.StringIO(), io.StringIO()
+ options = [*argv, '--codex-home', '/sender'] if rejected == 'options' else argv
+ with mock.patch.object(cli, 'Store'), \
+ mock.patch.object(cli.core, 'check_message',
+ side_effect=cli.Rejected('invalid_message') if rejected == 'message' else None), \
+ mock.patch.object(cli, 'device_credential',
+ side_effect=cli.Rejected('login_missing') if rejected == 'credential' else None), \
+ mock.patch.object(cli, 'exchange', mock.AsyncMock()) as exchange, \
+ contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+ self.assertEqual(cli.core.main(options), 1)
+ self.assertIn('Destination checks: not run', out.getvalue() + err.getvalue())
+ exchange.assert_not_awaited()
+
+
+class Negotiation(unittest.IsolatedAsyncioTestCase):
+ async def test_core_exchange_requests_context_only_for_explicit_dry_run(self):
+ for command, dry_run, explain in (('send', True, True), ('send', True, False),
+ ('send', False, True), ('list', False, True)):
+ args = SimpleNamespace(command=command, dry_run=dry_run, explain=explain,
+ device='a' * 64, host=None, wake=False, reply_to=None,
+ b64=False, no_from=True, to='alias', device_state=None,
+ device_route='auto', agent=None)
+ store = mock.Mock()
+ with mock.patch.object(cli, 'Store', return_value=store), \
+ mock.patch.object(cli.core, 'read_message', return_value='fixture'), \
+ mock.patch.object(cli.core, 'check_message'), \
+ mock.patch.object(cli, 'device_credential', return_value=None), \
+ mock.patch.object(cli, 'exchange', mock.AsyncMock(return_value={'ok': True})) as exchange:
+ await cli.core_exchange(args)
+ self.assertEqual(exchange.call_args.kwargs,
+ {'explain': True} if command == 'send' and dry_run and explain else {})
+ store.close.assert_called_once()
+
+ async def test_only_explicit_resolve_on_advertising_receiver_gets_flag(self):
+ store = mock.Mock(device='a' * 64)
+ store.recovery_required.return_value = False
+ store.peer.return_value = {'status': 'paired', 'certificate': 'fixture',
+ 'routes': {'direct': '127.0.0.1:1'}}
+ peer = 'b' * 64
+ for op in ('resolve', 'send', 'list'):
+ for explain in (False, True, 1):
+ for feature in (False, True):
+ probe = {'ok': True, 'device': peer}
+ if feature:
+ probe['features'] = ['delivery-explanation-v1']
+ native = {'ok': True, 'target': {'id': UUID}, 'codexHome': '/actual',
+ 'deliveryContext': CONTEXT}
+ channel = SimpleNamespace(send=mock.AsyncMock(), close=mock.AsyncMock(),
+ recv=mock.AsyncMock(side_effect=[probe, native]))
+ body = None if op == 'list' else {'target': 'alias', 'message': 'fixture'}
+ with mock.patch.object(app, 'open_channel', mock.AsyncMock(return_value=channel)) as connect:
+ result = await app.exchange(store, peer, op, body, route='direct', explain=explain)
+ connect.assert_awaited_once()
+ sent = [call.args[0] for call in channel.send.await_args_list]
+ self.assertEqual([item['op'] for item in sent], ['probe', op])
+ expected = {**body, 'explain': True} if op == 'resolve' and explain is True and feature else body
+ self.assertEqual(sent[1]['body'], expected)
+ self.assertEqual(result['target'], {'id': UUID})
+ self.assertEqual(result['codexHome'], '/actual')
+ negotiated = op == 'resolve' and explain is True and feature
+ self.assertEqual('deliveryContext' in result, negotiated)
+ if negotiated:
+ self.assertEqual(result['deliveryContext'], CONTEXT)
+
+ async def test_receiver_resolve_only_optional_flag(self):
+ from tests.relay.test_message_framing import EffectBoundaries, SENDER, PEER, IDENT
+ fixture = EffectBoundaries()
+ for op, flag, accepted in (('resolve', True, True), ('resolve', False, False), ('send', True, False)):
+ receiver = fixture.receiver()
+ body = {'target': 'fixture', 'message': 'fixture', 'explain': flag}
+ value = app.request(fixture.store(), PEER, op, body, IDENT)
+ if accepted:
+ await receiver.dispatch(SENDER, value)
+ self.assertTrue(receiver.native.invoke.call_args.kwargs['explain'])
+ else:
+ with self.assertRaisesRegex(app.Rejected, 'invalid_body'):
+ await receiver.dispatch(SENDER, value)
+ receiver.native.invoke.assert_not_called()
diff --git a/tools/generate_session_peer.py b/tools/generate_session_peer.py
index 6deaa08..d1575b6 100644
--- a/tools/generate_session_peer.py
+++ b/tools/generate_session_peer.py
@@ -23,6 +23,7 @@
"ssh_config.py",
"ssh.py",
"ssh_identity.py",
+ "explanation.py",
"output.py",
"diagnostics.py",
"adapters.py",