diff --git a/README.md b/README.md
index 9cb8a67..37c00a2 100644
--- a/README.md
+++ b/README.md
@@ -80,6 +80,10 @@ Windows以外では定期実行の登録に未対応です。Codexが終了し
 
 ### 通常Chatの一時Chat（GPT-6 Pro）へ渡す（任意）
 
+0.7.0では、確認済み接続の操作・範囲・有効期限をローカル台帳に保持し、`chat_plan.py` が必要な項目だけ機械照合します。分類専用のAI呼出しや全接続の再調査を避け、`chat_transfer.py` の一回限りのローカルフォームで、準備済み本文と最終回答をBrowser内の変数のまま受け渡せます。親のモデルへ全文を出力して再転記する必要を減らします。[接続と分類](plugins/codex-task-routing/skills/task-routing/references/chat-capabilities.md)、[転送補助](plugins/codex-task-routing/skills/task-routing/references/chat-transfer.md)。これらは明示実行の補助で、フックから起動・送信しません。
+
+通常ChatのGPT-6 ProはWork/Codexと別枠ですが、親の準備・操作・検収はCodex側に残ります。6 Proには契約別の上限があり、無制限ではありません。モデルや残量は実際のUIで確認し、第三者ツール固有の料金まで無料と扱いません。[公式の利用枠](https://help.openai.com/en/articles/20001354)、[Work/Codexの共有枠](https://learn.chatgpt.com/docs/pricing)。
+
 0.6.0では、この経路を有効にしたrootが、まとまった独立工程の担当選定時にまずChatへの適合を検討します。材料と確認済みツールで完結し、準備・待機・検収・手戻りが見合う比較・分析・設計・文案・レビューは6 Proを優先します。開発も証拠取得→Chatで分析→Codexで実装・検証と分けられます。短い仕事や頻繁な現物照会はCodexに残し、親にも独立作業がある条件を維持します。経路自体は既定で無効、子のmodel・effort表は従来どおりです。有効化後は毎回プラグインを指定する必要はありません。
 
 新規依頼には、目的・材料・合格条件に加え、`handoff` で必要情報、出典要件、鮮度、許可ツール、返却形式、停止条件を渡します。MCPはChat側で必要な操作が使えるかを確認し、回答に実取得の出典・版や日時・取得範囲を求めます。設定済み・選択済みだけを取得成功と扱いません。本文だけの仕事はツールなしで完結させ、承認済み範囲の確認を繰り返しません。
diff --git a/docs/validation.md b/docs/validation.md
index 50e08ee..4c68ba6 100644
--- a/docs/validation.md
+++ b/docs/validation.md
@@ -1,5 +1,14 @@
 # 検証記録
 
+## 0.7.1: PR #9の境界値レビュー
+
+2026-09-15、PR #9のローカル転送、接続台帳、キャッシュ公開の差分を確認しました。
+
+- JSONで表現できてもUTF-8へ保存できない単独サロゲートを、共通の文字列検証で拒否します。依頼のhash生成前・回答の保存前に安全なプロトコルエラーを返し、修正した回答は同じ転送で受理できます。通常の補助平面Unicodeは保持します。
+- 接続期限の上限判定を日時への加算から差分比較へ変更し、UTC変換の範囲外も安全な入力エラーとして扱います。
+- CIから取得したPRソースをLinux/Python 3.13.5で検証し、104件の単体テストが成功（環境依存2件はskip）、package検査とdiffの空白検査も通過しました。初回の旧101件では既存の子プロセス起動テストが1件失敗しましたが、対象再実行と修正後の全件で成功しました。タイミング依存の可能性は残します。
+- Browser実画面、Windowsの実プラグイン導入、実行モデル・利用量の測定は今回追加していません。3 OSの最終コミットの結果はPRのCIを参照してください。
+
 ## 0.6.0: Chat優先判定と引継ぎ契約
 
 有効なrootの工程選定にChat適合の優先判定を加え、分類表・スキル・起動時の追加指示をそろえました。既定オフ、子からのChat再委譲禁止、親の独立作業、nativeモデルとeffortの範囲は維持しています。
diff --git a/plugins/codex-task-routing/.codex-plugin/plugin.json b/plugins/codex-task-routing/.codex-plugin/plugin.json
index 77c2c20..52ce040 100644
--- a/plugins/codex-task-routing/.codex-plugin/plugin.json
+++ b/plugins/codex-task-routing/.codex-plugin/plugin.json
@@ -1,6 +1,6 @@
 {
   "name": "codex-task-routing",
-  "version": "0.6.0",
+  "version": "0.7.1",
   "description": "Task delegation policies for research, documents, images, operations, and development in Codex.",
   "author": {
     "name": "Acecore",
diff --git a/plugins/codex-task-routing/scripts/chat_plan.py b/plugins/codex-task-routing/scripts/chat_plan.py
new file mode 100644
index 0000000..9e62520
--- /dev/null
+++ b/plugins/codex-task-routing/scripts/chat_plan.py
@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Local, deterministic suitability checks; never discover tools or send work."""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import json
+from pathlib import Path
+import sys
+
+from chatgpt_route import ChatRouteError, _load_json_object, _require_keys
+
+FLAGS = {"substantial", "parent_has_independent_work", "materials_approved",
+         "handoff_proportionate", "requires_repeated_local_access"}
+STATES = {"advertised", "verified", "blocked"}
+
+
+def _keys(value, required, purpose):
+    if not isinstance(value, dict):
+        raise ChatRouteError(f"{purpose} must be an object")
+    _require_keys(value, required=required, allowed=required, purpose=purpose)
+
+
+def _text(value):
+    if not isinstance(value, str) or not value.strip() or len(value) > 512:
+        raise ChatRouteError("metadata must be a nonempty string of at most 512 characters")
+    return value
+
+
+def _time(value):
+    try:
+        parsed = datetime.fromisoformat(_text(value).replace("Z", "+00:00"))
+        if parsed.tzinfo is None:
+            raise ValueError()
+        return parsed.astimezone(timezone.utc)
+    except (ValueError, OverflowError) as exc:
+        raise ChatRouteError("observation time requires an ISO timestamp with timezone") from exc
+
+
+def validate_inventory(value):
+    _keys(value, {"schema_version", "surface", "model", "installed_plugins", "capabilities"}, "inventory")
+    if type(value["schema_version"]) is not int or value["schema_version"] != 1:
+        raise ChatRouteError("unsupported inventory schema")
+    if value["surface"] != "chat-temporary" or value["model"] != "6 Pro":
+        raise ChatRouteError("inventory must describe Temporary Chat with 6 Pro")
+    installed, caps = value["installed_plugins"], value["capabilities"]
+    if not isinstance(installed, list) or len(installed) > 128:
+        raise ChatRouteError("installed_plugins must be a bounded list")
+    for name in installed:
+        _text(name)
+    if len(set(installed)) != len(installed):
+        raise ChatRouteError("duplicate installed plugin")
+    if not isinstance(caps, list) or len(caps) > 256:
+        raise ChatRouteError("capabilities must be a bounded list")
+    seen = set()
+    for cap in caps:
+        _keys(cap, {"tool", "operation", "scope", "state", "observed_at", "expires_at"}, "capability")
+        key = tuple(_text(cap[k]) for k in ("tool", "operation", "scope"))
+        if key in seen:
+            raise ChatRouteError("duplicate capability scope")
+        seen.add(key)
+        if cap["tool"] not in installed or not isinstance(cap["state"], str) or cap["state"] not in STATES:
+            raise ChatRouteError("unknown tool or capability state")
+        observed, expires = _time(cap["observed_at"]), _time(cap["expires_at"])
+        if not timedelta(0) < expires - observed <= timedelta(days=7):
+            raise ChatRouteError("capability lifetime must be positive and at most seven days")
+    return value
+
+
+def plan(facts, inventory=None, *, now=None):
+    _keys(facts, FLAGS | {"needs"}, "facts")
+    if any(type(facts[k]) is not bool for k in FLAGS):
+        raise ChatRouteError("suitability flags must be JSON booleans")
+    needs = facts["needs"]
+    if not isinstance(needs, list) or len(needs) > 16:
+        raise ChatRouteError("needs must be a list of at most sixteen capabilities")
+    for need in needs:
+        _keys(need, {"tool", "operation", "scope"}, "need")
+        for val in need.values():
+            _text(val)
+    if inventory is not None:
+        validate_inventory(inventory)
+    # The caller supplies semantic judgments once. No classifier model is run.
+    if not facts["materials_approved"]:
+        return {"route": "not_ready", "reason": "materials_approval_not_established"}
+    for flag in ("substantial", "parent_has_independent_work", "handoff_proportionate"):
+        if not facts[flag]:
+            return {"route": "codex", "reason": flag + "_false"}
+    if facts["requires_repeated_local_access"]:
+        return {"route": "codex", "reason": "repeated_local_access"}
+    instant = now or datetime.now(timezone.utc)
+    checks = []
+    caps = inventory["capabilities"] if inventory else []
+    for need in needs:
+        cap = next((c for c in caps if all(c[k] == v for k, v in need.items())), None)
+        state = "unknown"
+        if cap:
+            state = cap["state"] if _time(cap["observed_at"]) <= instant < _time(cap["expires_at"]) else "stale"
+        checks.append({**need, "state": state, "installed": bool(inventory and need["tool"] in inventory["installed_plugins"])})
+    if any(c["state"] == "blocked" for c in checks):
+        return {"route": "not_ready", "reason": "required_capability_blocked", "checks": checks}
+    if any(c["state"] != "verified" for c in checks):
+        return {"route": "preflight_needed", "reason": "verify_only_required_capabilities", "checks": checks}
+    return {"route": "chat_candidate", "reason": "verify_live_surface_model_and_permissions", "checks": checks}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--facts", type=Path, required=True)
+    parser.add_argument("--inventory", type=Path)
+    args = parser.parse_args(argv)
+    try:
+        facts = _load_json_object(args.facts, purpose="facts")
+        inventory = _load_json_object(args.inventory, purpose="inventory") if args.inventory else None
+        print(json.dumps(plan(facts, inventory), ensure_ascii=False, separators=(",", ":")))
+        return 0
+    except (ChatRouteError, TypeError) as exc:
+        print(json.dumps({"error": str(exc)}, ensure_ascii=False), file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/plugins/codex-task-routing/scripts/chat_transfer.py b/plugins/codex-task-routing/scripts/chat_transfer.py
new file mode 100644
index 0000000..753cafc
--- /dev/null
+++ b/plugins/codex-task-routing/scripts/chat_transfer.py
@@ -0,0 +1,594 @@
+#!/usr/bin/env python3
+"""Serve one local-only Browser handoff for an approved Chat route bundle.
+
+The server deliberately exposes a prepared prompt only through a loopback
+HTML document.  It never opens ChatGPT, calls a model or network API, reads
+authentication or conversation data, or evaluates browser-side script.  A
+supported Browser surface can read the readonly prompt field and paste it
+into an already-approved temporary Chat.  It can later put that Chat's final
+JSON response into this form for local schema/correlation validation.
+"""
+
+from __future__ import annotations
+
+import argparse
+import html
+import json
+import os
+import secrets
+import socket
+import stat
+import sys
+import tempfile
+import time
+import threading
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from typing import Any, Mapping
+from urllib.parse import parse_qs, urlsplit
+
+
+# The helper is intentionally loaded from the same installed plugin directory.
+# It is the protocol authority for prepared-bundle and response validation.
+SCRIPT_DIRECTORY = Path(__file__).resolve().parent
+if str(SCRIPT_DIRECTORY) not in sys.path:
+    sys.path.insert(0, str(SCRIPT_DIRECTORY))
+import chatgpt_route  # noqa: E402
+
+
+DEFAULT_TIMEOUT_SECONDS = 10 * 60
+MAX_TIMEOUT_SECONDS = 60 * 60
+# A URL-encoded form can expand a valid 256 KiB reply substantially.  The
+# decoded canonical reply is separately kept within chatgpt_route.MAX_JSON_BYTES.
+MAX_POST_BYTES = chatgpt_route.MAX_JSON_BYTES * 3
+CONTENT_SECURITY_POLICY = (
+    "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; "
+    "base-uri 'none'; frame-ancestors 'none'"
+)
+
+
+class ChatTransferError(ValueError):
+    """A local transfer error that does not include submitted material."""
+
+
+def _origin_for(port: int) -> str:
+    return f"http://127.0.0.1:{port}"
+
+
+def _host_is_valid(host: str | None, port: int) -> bool:
+    return host == f"127.0.0.1:{port}"
+
+
+def _origin_is_valid(origin: str | None, port: int) -> bool:
+    return origin == _origin_for(port)
+
+
+def _path_is_valid(path: str, token: str) -> bool:
+    parsed = urlsplit(path)
+    return not parsed.query and parsed.path == f"/transfer/{token}"
+
+
+def _minimal_error_page() -> bytes:
+    """Return an unauthenticated error document with no transfer material."""
+
+    return b"<!doctype html><meta charset=utf-8><title>Unavailable</title><p>Local transfer unavailable.</p>"
+
+
+def _parse_post_length(value: str | None) -> int:
+    try:
+        length = int(value or "")
+    except ValueError as exc:
+        raise ChatTransferError("invalid response length") from exc
+    if length < 0 or length > MAX_POST_BYTES:
+        raise ChatTransferError("response exceeds the size limit")
+    return length
+
+
+def _response_text_from_form(body: bytes) -> str:
+    try:
+        form = parse_qs(
+            body.decode("utf-8"), strict_parsing=True, keep_blank_values=True, max_num_fields=2, errors="strict"
+        )
+    except (UnicodeDecodeError, ValueError) as exc:
+        raise ChatTransferError("invalid form data") from exc
+    values = form.get("response_json")
+    if set(form) != {"response_json"} or values is None or len(values) != 1:
+        raise ChatTransferError("exactly one response JSON field is required")
+    response_text = values[0]
+    if len(response_text.encode("utf-8")) > chatgpt_route.MAX_JSON_BYTES:
+        raise ChatTransferError("response exceeds the size limit")
+    return response_text
+
+
+def _response_parse_status(response_text: str, error: chatgpt_route.ChatRouteError) -> str:
+    """Describe a JSON rejection without exposing any response characters."""
+
+    cause = error.__cause__
+    if isinstance(cause, json.JSONDecodeError):
+        if cause.pos < len(response_text):
+            character = f"U+{ord(response_text[cause.pos]):04X}"
+        else:
+            character = "EOF"
+        return (
+            "Response JSON parse rejected "
+            f"(syntax line={cause.lineno}, column={cause.colno}, character={character}, "
+            f"decoded_chars={len(response_text)})."
+        )
+    if isinstance(cause, chatgpt_route.DuplicateKeyError):
+        return f"Response JSON parse rejected (duplicate object key, decoded_chars={len(response_text)})."
+    if isinstance(cause, ValueError):
+        return f"Response JSON parse rejected (unsupported JSON value, decoded_chars={len(response_text)})."
+    if "must be a JSON object" in str(error):
+        return f"Response JSON parse rejected (top-level object required, decoded_chars={len(response_text)})."
+    return f"Response JSON parse rejected (unclassified parser failure, decoded_chars={len(response_text)})."
+
+
+def _safe_reply_path(path: Path | str) -> Path:
+    """Return one explicit reply target after link/reparse and parent checks."""
+
+    safe_path = chatgpt_route._assert_safe_ancestors(path)
+    parent = safe_path.parent
+    try:
+        metadata = os.lstat(parent)
+    except OSError as exc:
+        raise ChatTransferError("reply directory is unavailable") from exc
+    if not stat.S_ISDIR(metadata.st_mode) or chatgpt_route._is_link_or_reparse(parent):
+        raise ChatTransferError("reply directory is unavailable")
+    if os.path.lexists(safe_path) and chatgpt_route._is_link_or_reparse(safe_path):
+        raise ChatTransferError("refusing a symlink or reparse-point reply path")
+    if os.path.lexists(safe_path):
+        try:
+            target_metadata = os.lstat(safe_path)
+        except OSError as exc:
+            raise ChatTransferError("reply path is unavailable") from exc
+        if not stat.S_ISREG(target_metadata.st_mode):
+            raise ChatTransferError("reply path is not a regular file")
+    return safe_path
+
+
+def _canonical_response_bytes(response: Mapping[str, Any]) -> bytes:
+    encoded = (
+        json.dumps(response, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+        + "\n"
+    ).encode("utf-8")
+    if len(encoded) > chatgpt_route.MAX_JSON_BYTES:
+        raise ChatTransferError("response exceeds the supported reply size limit")
+    return encoded
+
+
+def _load_existing_reply(path: Path) -> dict[str, Any] | None:
+    if not os.path.lexists(path):
+        return None
+    try:
+        return chatgpt_route._load_json_object(path, purpose="existing response")
+    except chatgpt_route.ChatRouteError as exc:
+        raise ChatTransferError("existing reply is unavailable") from exc
+
+
+def _atomic_create_reply(path: Path, content: bytes) -> bool:
+    """Create *path* atomically without replacing an independently made file.
+
+    ``os.replace`` would make the write atomic but could overwrite a different
+    reply created after the initial check.  Linking a fully-fsynced temporary
+    file creates the final name only when it did not already exist.
+    """
+
+    temporary_name: str | None = None
+    try:
+        descriptor, temporary_name = tempfile.mkstemp(
+            prefix=".chat-transfer-", dir=path.parent
+        )
+        with os.fdopen(descriptor, "wb") as handle:
+            handle.write(content)
+            handle.flush()
+            os.fsync(handle.fileno())
+        temporary = Path(temporary_name)
+        if chatgpt_route._is_link_or_reparse(temporary):
+            raise ChatTransferError("temporary reply path is unsafe")
+        try:
+            os.link(temporary, path)
+        except FileExistsError:
+            return False
+        return True
+    except ChatTransferError:
+        raise
+    except OSError as exc:
+        raise ChatTransferError("cannot save the reply") from exc
+    finally:
+        if temporary_name and os.path.lexists(temporary_name):
+            try:
+                os.unlink(temporary_name)
+            except OSError:
+                pass
+
+
+class TransferState:
+    """The bounded, in-memory state for exactly one bundle and reply target."""
+
+    def __init__(self, *, bundle: Mapping[str, Any], reply_path: Path) -> None:
+        request_id, input_sha256 = chatgpt_route._expected_request(bundle)
+        prompt = bundle.get("prompt")
+        if not isinstance(prompt, str):
+            raise ChatTransferError("prepared bundle prompt is unavailable")
+        self.bundle = dict(bundle)
+        self.prompt = prompt
+        self.request_id = request_id
+        self.input_sha256 = input_sha256
+        self.reply_path = _safe_reply_path(reply_path)
+        self.response_status: str | None = None
+        # Exposed only on the authorized local page.  These fixed strings
+        # identify transport progress without retaining response material.
+        self.transfer_status = "Waiting for one response JSON submission."
+
+    @classmethod
+    def from_paths(cls, bundle_path: Path | str, reply_path: Path | str) -> "TransferState":
+        bundle = chatgpt_route._load_json_object(Path(bundle_path), purpose="prepared bundle")
+        return cls(bundle=bundle, reply_path=Path(reply_path))
+
+    def save_response_text(self, response_text: str) -> tuple[dict[str, Any], bool]:
+        """Validate and save one response, accepting only an identical retry."""
+
+        try:
+            response = chatgpt_route._parse_json_object(response_text, purpose="response")
+        except chatgpt_route.ChatRouteError as exc:
+            self.transfer_status = _response_parse_status(response_text, exc)
+            raise ChatTransferError(str(exc)) from exc
+        return self.save_response(response)
+
+    def save_response(self, response: Mapping[str, Any]) -> tuple[dict[str, Any], bool]:
+        """Return validation metadata and whether this call created the reply."""
+
+        try:
+            _safe_reply_path(self.reply_path)
+        except chatgpt_route.ChatRouteError as exc:
+            raise ChatTransferError("reply path is unsafe") from exc
+        try:
+            validated = chatgpt_route.validate_exchange(self.bundle, response)
+        except chatgpt_route.ChatRouteError as exc:
+            raise ChatTransferError(str(exc)) from exc
+        content = _canonical_response_bytes(response)
+        existing = _load_existing_reply(self.reply_path)
+        if existing is not None:
+            try:
+                chatgpt_route.validate_exchange(self.bundle, existing)
+            except chatgpt_route.ChatRouteError as exc:
+                raise ChatTransferError("existing reply is not a valid response for this bundle") from exc
+            if existing != dict(response):
+                raise ChatTransferError("reply already exists with a different response")
+            self.response_status = validated["status"]
+            return validated, False
+
+        created = _atomic_create_reply(self.reply_path, content)
+        if not created:
+            # Another process won the race.  It is idempotent only when its
+            # parsed protocol object is exactly the response just submitted.
+            existing = _load_existing_reply(self.reply_path)
+            if existing is None:
+                raise ChatTransferError("reply could not be saved")
+            try:
+                chatgpt_route.validate_exchange(self.bundle, existing)
+            except chatgpt_route.ChatRouteError as exc:
+                raise ChatTransferError("existing reply is not a valid response for this bundle") from exc
+            if existing != dict(response):
+                raise ChatTransferError("reply already exists with a different response")
+        self.response_status = validated["status"]
+        return validated, created
+
+
+def _response_status_message(status: str | None, *, created: bool | None = None) -> str:
+    if status is None:
+        return "Waiting for one response JSON submission."
+    prefix = "Response saved" if created else "Matching response already saved"
+    if status == "completed":
+        return (
+            f"{prefix} after schema and request-correlation validation. "
+            "Response status is completed. Parent semantic acceptance is still required."
+        )
+    return (
+        f"{prefix} after schema and request-correlation validation. "
+        f"Response status is {status}; this is not a completed outcome and requires parent attention."
+    )
+
+
+def _page(state: TransferState, *, status: str, error: bool = False) -> bytes:
+    status_class = "error" if error else "status"
+    escaped_prompt = html.escape(state.prompt, quote=False)
+    escaped_status = html.escape(status, quote=False)
+    action = html.escape(f"/transfer/{state.token}", quote=True)
+    document = f"""<!doctype html>
+<html lang=\"en\">
+<meta charset=\"utf-8\">
+<!-- Keep the form's same-origin Origin header.  no-referrer makes an HTML
+     form submission opaque (Origin: null) even when the target is local. -->
+<meta name=\"referrer\" content=\"same-origin\">
+<title>Local Chat transfer</title>
+<style>
+body {{ font-family: system-ui, sans-serif; margin: 2rem auto; max-width: 58rem; padding: 0 1rem; }}
+label {{ display: block; font-weight: 650; margin-top: 1rem; }}
+textarea {{ box-sizing: border-box; min-height: 12rem; width: 100%; font: 0.9rem ui-monospace, monospace; }}
+.status {{ border-left: .3rem solid #286; padding-left: .75rem; }}
+.error {{ border-left: .3rem solid #a22; padding-left: .75rem; }}
+</style>
+<h1>Local Chat transfer</h1>
+<p>Use supported Browser DOM controls to move this approved prompt and the final response. This page does not contact Chat or any external service.</p>
+<form method=\"post\" action=\"{action}\" accept-charset=\"utf-8\">
+  <label for=\"request_prompt\">Request prompt</label>
+  <textarea id=\"request_prompt\" readonly aria-readonly=\"true\">{escaped_prompt}</textarea>
+  <label for=\"response_json\">Response JSON</label>
+  <textarea id=\"response_json\" name=\"response_json\" required></textarea>
+  <button type=\"submit\">Save verified response</button>
+</form>
+<h2>Transfer status</h2>
+<p class=\"{status_class}\">{escaped_status}</p>
+</html>
+"""
+    return document.encode("utf-8")
+
+
+class TransferServer(HTTPServer):
+    """A loopback-only HTTP server carrying one TransferState."""
+
+    def __init__(self, state: TransferState) -> None:
+        super().__init__(("127.0.0.1", 0), TransferHandler)
+        self.state = state
+        self.completed = False
+        self.client_timeout_seconds = 1.0
+        self.deadline: float | None = None
+
+    @property
+    def origin(self) -> str:
+        return _origin_for(self.server_port)
+
+    def get_request(self) -> tuple[Any, Any]:
+        request, address = super().get_request()
+        request.settimeout(self.client_timeout_seconds)
+        return request, address
+
+    def finish_request(self, request: Any, client_address: Any) -> None:
+        # Socket timeouts bound inactivity, not a client continuously sending
+        # bytes. Interrupt the active connection at the absolute CLI deadline.
+        def expire() -> None:
+            try:
+                request.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+
+        timer = None
+        if self.deadline is not None:
+            timer = threading.Timer(max(0.0, self.deadline - time.monotonic()), expire)
+            timer.daemon = True
+            timer.start()
+        try:
+            super().finish_request(request, client_address)
+        except OSError:
+            # A disconnected Browser or expired socket carries no useful
+            # traceback; preserve the one-shot server until its deadline.
+            self.state.transfer_status = "Browser connection ended before transfer completion."
+        finally:
+            if timer is not None:
+                timer.cancel()
+
+
+class TransferHandler(BaseHTTPRequestHandler):
+    server: TransferServer
+    # Explicit close framing keeps Browser form responses unambiguous across
+    # the loopback environments exercised by this helper.
+    protocol_version = "HTTP/1.0"
+
+    def log_message(self, _format: str, *args: object) -> None:
+        # Do not emit the capability URL, submitted response, or request paths.
+        return
+
+    def _host_is_valid(self) -> bool:
+        return _host_is_valid(self.headers.get("Host"), self.server.server_port)
+
+    def _path_is_valid(self) -> bool:
+        return _path_is_valid(self.path, self.server.state.token)
+
+    def _set_transfer_status(self, status: str) -> None:
+        self.server.state.transfer_status = status
+
+    def _send(self, code: HTTPStatus, content: bytes, *, content_type: str = "text/html; charset=utf-8") -> None:
+        # The response is complete before this connection is closed.  Each
+        # local capability URL has one request per connection, avoiding a
+        # keep-alive parser retaining form data after submission.
+        self.close_connection = True
+        self.send_response(code)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(content)))
+        self.send_header("Connection", "close")
+        self.send_header("Cache-Control", "no-store, max-age=0")
+        self.send_header("Pragma", "no-cache")
+        self.send_header(
+            "Content-Security-Policy",
+            CONTENT_SECURITY_POLICY,
+        )
+        # Match the document policy so same-origin Browser form submissions
+        # retain their Origin header while cross-origin navigation leaks none.
+        self.send_header("Referrer-Policy", "same-origin")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.end_headers()
+        self.wfile.write(content)
+
+    def _send_error_page(self, code: HTTPStatus, status: str) -> None:
+        self._send(code, _page(self.server.state, status=status, error=True))
+
+    def _send_unavailable(self, code: HTTPStatus) -> None:
+        self._send(code, _minimal_error_page())
+
+    def do_GET(self) -> None:
+        if not self._host_is_valid():
+            self._send_unavailable(HTTPStatus.BAD_REQUEST)
+            return
+        if not self._path_is_valid():
+            self._send_unavailable(HTTPStatus.NOT_FOUND)
+            return
+        self._send(
+            HTTPStatus.OK,
+            _page(
+                self.server.state,
+                status=(
+                    _response_status_message(self.server.state.response_status)
+                    if self.server.state.response_status is not None
+                    else self.server.state.transfer_status
+                ),
+            ),
+        )
+
+    def do_POST(self) -> None:
+        if not self._host_is_valid():
+            self._send_unavailable(HTTPStatus.BAD_REQUEST)
+            return
+        if not self._path_is_valid():
+            self._send_unavailable(HTTPStatus.NOT_FOUND)
+            return
+        origin_valid = _origin_is_valid(self.headers.get("Origin"), self.server.server_port)
+        self._set_transfer_status(
+            f"POST received (host_valid=true, path_valid=true, origin_valid={'true' if origin_valid else 'false'})."
+        )
+        if not origin_valid:
+            self._send_unavailable(HTTPStatus.FORBIDDEN)
+            return
+        self._set_transfer_status("POST origin accepted; checking form encoding.")
+        content_type = self.headers.get("Content-Type", "")
+        if content_type.split(";", 1)[0].strip().lower() != "application/x-www-form-urlencoded":
+            self._set_transfer_status("Response form rejected before body processing.")
+            self._send_error_page(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "Transfer status: response must use the local form.")
+            return
+        try:
+            length = _parse_post_length(self.headers.get("Content-Length"))
+        except ChatTransferError as exc:
+            code = HTTPStatus.REQUEST_ENTITY_TOO_LARGE if "size limit" in str(exc) else HTTPStatus.BAD_REQUEST
+            self._set_transfer_status("Response form rejected before body processing.")
+            self._send_error_page(code, f"Transfer status: {exc}.")
+            return
+        self._set_transfer_status(f"POST body reading (content_length={length} bytes).")
+        try:
+            body = self.rfile.read(length)
+        except (OSError, TimeoutError):
+            # The client socket is bounded by serve_transfer's remaining
+            # deadline.  A partial upload must not keep the CLI alive.
+            self._set_transfer_status("Response body did not arrive before the transfer deadline.")
+            self.close_connection = True
+            return
+        if len(body) != length:
+            self._set_transfer_status("Response body was incomplete.")
+            self._send_error_page(HTTPStatus.BAD_REQUEST, "Transfer status: incomplete response body.")
+            return
+        self._set_transfer_status("Checking response JSON format.")
+        try:
+            response_text = _response_text_from_form(body)
+        except ChatTransferError as exc:
+            code = HTTPStatus.REQUEST_ENTITY_TOO_LARGE if "size limit" in str(exc) else HTTPStatus.BAD_REQUEST
+            self._set_transfer_status("Response JSON was rejected before correlation validation.")
+            self._send_error_page(code, f"Transfer status: {exc}.")
+            return
+        self._set_transfer_status("Validating response correlation.")
+        try:
+            validated, created = self.server.state.save_response_text(response_text)
+        except ChatTransferError as exc:
+            if not self.server.state.transfer_status.startswith("Response JSON parse rejected"):
+                self._set_transfer_status("Response JSON was rejected by correlation or save checks.")
+            self._send_error_page(HTTPStatus.UNPROCESSABLE_ENTITY, f"Transfer status: response rejected ({exc}).")
+            return
+        self.server.completed = True
+        self._send(
+            HTTPStatus.OK,
+            _page(
+                self.server.state,
+                status=_response_status_message(validated["status"], created=created),
+            ),
+        )
+
+
+def create_server(state: TransferState) -> TransferServer:
+    """Create a random-port, random-capability loopback server for tests or CLI."""
+
+    state.token = secrets.token_urlsafe(32)
+    return TransferServer(state)
+
+
+def serve_transfer(state: TransferState, *, timeout_seconds: int) -> int:
+    server = create_server(state)
+    deadline = time.monotonic() + timeout_seconds
+    server.deadline = deadline
+    metadata = {
+        "event": "listening",
+        "input_sha256": state.input_sha256,
+        "request_id": state.request_id,
+        "timeout_seconds": timeout_seconds,
+        "url": f"{server.origin}/transfer/{state.token}",
+    }
+    print(json.dumps(metadata, ensure_ascii=True, sort_keys=True), flush=True)
+    try:
+        while not server.completed:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                print(
+                    json.dumps(
+                        {"event": "timeout", "request_id": state.request_id},
+                        ensure_ascii=True,
+                        sort_keys=True,
+                    ),
+                    flush=True,
+                )
+                return 3
+            server.timeout = min(remaining, 1.0)
+            # A submitted form may take more than a second to arrive from the
+            # Browser.  Bound the read by the transfer deadline itself, rather
+            # than closing a valid in-flight response after one second.
+            server.client_timeout_seconds = max(0.01, remaining)
+            server.handle_request()
+    except KeyboardInterrupt:
+        print(
+            json.dumps(
+                {"event": "cancelled", "request_id": state.request_id},
+                ensure_ascii=True,
+                sort_keys=True,
+            ),
+            flush=True,
+        )
+        return 130
+    finally:
+        server.server_close()
+    final_status = state.response_status
+    print(
+        json.dumps(
+            {"event": "saved", "request_id": state.request_id, "status": final_status},
+            ensure_ascii=True,
+            sort_keys=True,
+        ),
+        flush=True,
+    )
+    return 0 if final_status == "completed" else 3
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Run one local Browser-to-Chat response transfer")
+    parser.add_argument("--bundle", type=Path, required=True, help="approved bundle.json path")
+    parser.add_argument("--reply", type=Path, required=True, help="the single reply.json path to create")
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=DEFAULT_TIMEOUT_SECONDS,
+        help=f"seconds to wait before stopping (1-{MAX_TIMEOUT_SECONDS}; default {DEFAULT_TIMEOUT_SECONDS})",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    if not 1 <= args.timeout <= MAX_TIMEOUT_SECONDS:
+        print("error: timeout must be within the supported range", file=sys.stderr)
+        return 2
+    try:
+        state = TransferState.from_paths(args.bundle, args.reply)
+    except (ChatTransferError, chatgpt_route.ChatRouteError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    return serve_transfer(state, timeout_seconds=args.timeout)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/plugins/codex-task-routing/scripts/chatgpt_route.py b/plugins/codex-task-routing/scripts/chatgpt_route.py
index c4093a4..7b88910 100644
--- a/plugins/codex-task-routing/scripts/chatgpt_route.py
+++ b/plugins/codex-task-routing/scripts/chatgpt_route.py
@@ -177,6 +177,12 @@ def _require_keys(value: Mapping[str, Any], *, required: set[str], allowed: set[
 def _require_nonempty_string(value: Any, *, purpose: str, maximum: int) -> str:
     if not isinstance(value, str) or not value.strip() or len(value) > maximum:
         raise ChatRouteError(f"{purpose} must be a non-empty string within the size limit")
+    # JSON permits escaped lone surrogates that cannot be saved as UTF-8.
+    # Reject them at the shared protocol boundary, before hashing or saving.
+    try:
+        value.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        raise ChatRouteError(f"{purpose} must contain valid Unicode text") from exc
     return value
 
 
diff --git a/plugins/codex-task-routing/scripts/routing.py b/plugins/codex-task-routing/scripts/routing.py
index eb3f541..08f86c6 100644
--- a/plugins/codex-task-routing/scripts/routing.py
+++ b/plugins/codex-task-routing/scripts/routing.py
@@ -19,6 +19,7 @@
 import secrets
 import stat
 import sys
+import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Iterator, Mapping
@@ -58,6 +59,14 @@
 # an effective policy mid-document.
 MAX_ADDITIONAL_CONTEXT_CHARS = 8000
 
+# Windows can briefly retain a directory handle immediately after another
+# process creates a fresh cache tree.  Only retry the Windows errors that can
+# describe that short sharing/access window, and leave all other filesystem
+# errors visible to the caller.
+CACHE_RENAME_MAX_ATTEMPTS = 3
+CACHE_RENAME_RETRY_SECONDS = 0.05
+WINDOWS_TRANSIENT_RENAME_WINERRORS = frozenset({5, 32, 33})
+
 EFFORT_ORDER = {"medium": 0, "high": 1, "xhigh": 2, "max": 3}
 MODEL_FIELDS = ("id", "min_effort", "default_effort", "max_effort")
 REQUIRED_MODEL_ROLES = ("luna", "terra", "sol", "astra")
@@ -532,16 +541,30 @@ def _cache_matches_policy(directory: Path, policy: Policy) -> bool:
     return True
 
 
+def _verified_existing_cache(destination: Path, policy: Policy) -> bool:
+    """Return whether a published cache is exact, rejecting unsafe entries."""
+
+    if not os.path.lexists(destination):
+        return False
+    if _is_link_or_reparse(destination) or not destination.is_dir():
+        raise RoutingError("policy cache entry is unsafe")
+    if not _cache_matches_policy(destination, policy):
+        raise RoutingError("policy cache entry does not match the effective policy")
+    return True
+
+
+def _is_transient_windows_rename_error(exc: OSError) -> bool:
+    """Recognize only bounded-retry candidates from Windows directory sharing."""
+
+    return os.name == "nt" and getattr(exc, "winerror", None) in WINDOWS_TRANSIENT_RENAME_WINERRORS
+
+
 def _cache_directory(codex_home: Path, policy: Policy) -> Path:
     """Create a fresh content-addressed cache directory without touching an old one."""
 
     cache_root = _safe_mkdir(_absolute(codex_home) / CACHE_RELATIVE)
     destination = cache_root / policy.content_hash
-    if os.path.lexists(destination):
-        if _is_link_or_reparse(destination) or not destination.is_dir():
-            raise RoutingError("policy cache entry is unsafe")
-        if not _cache_matches_policy(destination, policy):
-            raise RoutingError("policy cache entry does not match the effective policy")
+    if _verified_existing_cache(destination, policy):
         return destination
     staging = cache_root / f".routing-staging-{policy.content_hash}-{secrets.token_hex(8)}"
     try:
@@ -549,21 +572,34 @@ def _cache_directory(codex_home: Path, policy: Policy) -> Path:
         staging.mkdir()
         render_policy(policy, staging)
         # os.rename does not overwrite an existing destination on Windows.
-        os.rename(staging, destination)
-    except OSError as exc:
-        # Windows uses EEXIST and Linux may use ENOTEMPTY when a concurrent
-        # hook has already atomically published this non-empty directory.
-        if exc.errno in {errno.EEXIST, errno.ENOTEMPTY}:
-            if (
-                os.path.lexists(destination)
-                and not _is_link_or_reparse(destination)
-                and destination.is_dir()
-                and _cache_matches_policy(destination, policy)
-            ):
+        for attempt in range(CACHE_RENAME_MAX_ATTEMPTS):
+            try:
+                os.rename(staging, destination)
                 return destination
-            raise RoutingError("policy cache entry appeared during rendering") from exc
+            except OSError as exc:
+                # Windows uses EEXIST and Linux may use ENOTEMPTY when a
+                # concurrent hook has already atomically published this
+                # non-empty directory.  Never accept a link or mismatched
+                # directory as a concurrent winner.
+                if exc.errno in {errno.EEXIST, errno.ENOTEMPTY}:
+                    if _verified_existing_cache(destination, policy):
+                        return destination
+                    raise RoutingError("policy cache entry appeared during rendering") from exc
+                # A fresh directory can be temporarily held by Windows file
+                # indexing or another local process.  WinError 5 is also used
+                # for permanent denial, so this remains short and bounded; a
+                # persistent failure still escapes as a RoutingError.
+                if (
+                    _is_transient_windows_rename_error(exc)
+                    and attempt + 1 < CACHE_RENAME_MAX_ATTEMPTS
+                ):
+                    if _verified_existing_cache(destination, policy):
+                        return destination
+                    time.sleep(CACHE_RENAME_RETRY_SECONDS * (attempt + 1))
+                    continue
+                raise RoutingError("cannot create the policy cache entry") from exc
+    except OSError as exc:
         raise RoutingError("cannot create the policy cache entry") from exc
-    return destination
 
 
 def _guidance_directories(cwd: Path, codex_home: Path | None) -> list[Path]:
diff --git a/plugins/codex-task-routing/skills/task-routing/SKILL.md b/plugins/codex-task-routing/skills/task-routing/SKILL.md
index b559f41..377128b 100644
--- a/plugins/codex-task-routing/skills/task-routing/SKILL.md
+++ b/plugins/codex-task-routing/skills/task-routing/SKILL.md
@@ -5,6 +5,8 @@ description: Apply Codex Task Routing for substantial independent tasks, includi
 
 # Task routing
 
+Avoid a classifier-only model call or repeated full tool inventories. For a substantial Chat candidate, reuse the scoped, expiring capability inventory and deterministic `chat_plan.py` checks in [capability reuse](references/chat-capabilities.md); short work needs no extra planning files. Use [local transfer](references/chat-transfer.md) when supported Browser DOM operations can move the prepared prompt and final response without echoing their full text into the parent's context. These helpers do not authorize dispatch, attest a model, or replace semantic acceptance.
+
 For an enabled normal Chat route, evaluate Chat suitability before choosing a native child for each substantial independent work unit; users need not mention this plugin again. Prefer 6 Pro for work that can finish with approved materials and verified tools when preparation, waiting, acceptance and rework are proportionate. Split analysis or design from local execution when useful. Preserve the parent's useful independent work requirement. A small remainder or repeated local inspection stays in Codex. Use the route reference's selection and handoff checks; enabling Chat does not grant data sharing or tool permissions.
 
 Use the effective policy already supplied by this plugin's start hook. Keep the user's chosen parent model and effort. It authorizes choosing suitable standard subagents only within the user's task, available tools, and higher-priority instructions. A child requires a bounded independent task and useful independent work for its direct parent. Short work stays with the current capable agent.
diff --git a/plugins/codex-task-routing/skills/task-routing/references/chat-capabilities.md b/plugins/codex-task-routing/skills/task-routing/references/chat-capabilities.md
new file mode 100644
index 0000000..f5e74b8
--- /dev/null
+++ b/plugins/codex-task-routing/skills/task-routing/references/chat-capabilities.md
@@ -0,0 +1,72 @@
+# 接続の再利用と機械的な適合確認
+
+自然文の意味判断、機密の範囲、引継ぎが見合うかはrootが一度判断します。分類専用の子やChatを起動しません。短い仕事ではCLI入力を作る負担も避け、その場で判断します。まとまった工程では下の事実JSONと確認済み接続を `chat_plan.py` で照合できます。これは送信許可や実行器ではなく、モデルを使わない補助です。
+
+## 接続一覧を毎回読まない
+
+同じ工程は同じfactsファイルと一行の選定理由を再利用し、材料・範囲・必要操作・親の独立作業・準備や検収の負担が変わった時だけ意味判断をやり直します。期限の照合はCLIで再実行でき、分類のためにモデルへ材料を再送しません。新しいChatのUI・6 Pro・必要プラグインの提供状態は現在の画面で確認します。
+
+初回は依頼に必要な接続だけ、利用者が一覧調査を依頼した時は既存接続のメタデータだけを確認します。サポートされたChatの設定・一時Chatのプラスメニューで確認し、Codex側のツール一覧からChatの能力を推測しません。接続の確認だけを目的にメール本文や顧客データを取得しません。
+
+結果は利用者が指定したローカル領域、通常は `<CODEX_HOME>/codex-task-routing/capabilities.json` に保存します。配布キャッシュや公開repoには個人の接続一覧を入れません。秘密、アカウント識別子、会話URL、応答全文、認証情報は保存しません。フックは一覧を読み込まず、会話コンテキストにも注入しません。
+
+- `installed_plugins` はインストール済みの名前です。接続・権限・Chatでの取得成功を保証しません。
+- `capabilities` は実際に必要となった操作と対象範囲だけ記録します。`advertised` はそのChat側で操作の提供を確認、`verified` は同じ操作と対象範囲で実取得を確認、`blocked` は取得不能です。設定の「すべて許可」は新しい仕事の承認ではありません。
+- `scope` は完全一致です。例のrepoでの成功を全repo、書込み、CI実行へ拡張しません。UIだけで見た操作はverifiedにしません。Chatの自己申告だけで更新せず、ツール実行の表示、出典、取得範囲を確認します。
+- 各記録に確認日時と有効期限を付けます。最大7日で、短い期限を選べます。未来日時・期限切れはstaleとして必要分だけ再確認します。接続エラー、再ログイン、権限やアカウント変更に気づいたら期限内でも直ちにblockedまたはadvertisedへ戻します。
+- 検索は `needs` に合う操作だけ。未登録・期限切れは不存在と断定せず、必要な接続だけ確認します。本文が十分ならツールなしの依頼にできます。
+
+GitHubでは承認されたrepoの原文・差分・Issue・CIの読取りを使って調査やレビューをまとめて渡せます。Gmailは指定範囲の検索・要約、Canva/Figmaは対象デザインの読取り・提案、Stripeは承認済み範囲の調査、公式Docs MCPは仕様確認の候補です。いずれも実際のChat側の操作と対象権限を確認します。プラグインのスキルだけの導入からローカル編集・ビルド・本番実行がChatで可能とは扱いません。書込みが必要な時はその操作・対象・承認を別途確定し、読取実績を流用しません。
+
+## 入力と結果
+
+接続台帳の最小例（架空）:
+
+```json
+{
+  "schema_version": 1,
+  "surface": "chat-temporary",
+  "model": "6 Pro",
+  "installed_plugins": ["GitHub"],
+  "capabilities": [{
+    "tool": "GitHub", "operation": "fetch_file", "scope": "example/project",
+    "state": "verified",
+    "observed_at": "2026-09-14T00:00:00Z",
+    "expires_at": "2026-09-15T00:00:00Z"
+  }]
+}
+```
+
+一度判断する事実を `facts.json` に指定します。目的・材料の全文は含めません。
+
+```json
+{
+  "substantial": true,
+  "parent_has_independent_work": true,
+  "materials_approved": true,
+  "handoff_proportionate": true,
+  "requires_repeated_local_access": false,
+  "needs": [{"tool": "GitHub", "operation": "fetch_file", "scope": "example/project"}]
+}
+```
+
+```text
+python <plugin-root>/scripts/chat_plan.py --facts facts.json --inventory <CODEX_HOME>/codex-task-routing/capabilities.json
+```
+
+本文だけで完結する場合は `needs: []` とし、`--inventory` を省略できます。CLIはネットワークやユーザー設定を探索せず、指定ファイルだけ読みます。重複キー・未知キー・不正な型は拒否します。出力には該当操作の状態と短い理由だけを含めます。
+
+| route | rootの次の動作 |
+| --- | --- |
+| chat_candidate | 有効なopt-in設定、現在のChat／一時Chat／6 Pro、対象と操作の承認を確認して実依頼を準備 |
+| preflight_needed | 必須の未確認・期限切れ操作だけ確認。既に材料が十分ならツール不要の依頼へ縮小 |
+| codex | 現担当で進めるか、有効方針の通常子のmodel・effort表を使う |
+| not_ready | 不足・取得不能を報告し、既存の承認内で解決。暗黙にWork/APIへ切り替えない |
+
+CLIの終了コード0は分類処理の成功です。`chat_candidate` も送信の許可・モデル・利用枠・内容の十分さを証明しません。Chatの有効化や親のモデル設定は変更しません。接続確認のためだけの追加Chatは作らず、必要な実取得を本来の仕事と一緒に行って証拠を返します。
+
+## 引継ぎ本文を再生成しない
+
+依頼JSONは一度だけ作り、`chatgpt_route.py prepare` の出力をファイルへリダイレクトします。大きなpromptをツール出力へ流して親が再転記したり、同じ材料を別の分類依頼へ送ったりしません。対応するBrowser機能がある時は [ローカル転送補助](chat-transfer.md) でDOMの値を変数のまま受け渡します。返却もID/hash/形式を機械照合し、親は要点と合格条件・重要根拠だけ読みます。承認確認や品質受入は省略しません。
+
+この仕組みが削るのは分類専用のモデル呼出しと本文の再転記です。rootの判断・Browserツール呼出し・検収にはCodex側の利用負担が残ります。ファイル文字数はトークン実測ではなく、削減率は比較測定なしに提示しません。
diff --git a/plugins/codex-task-routing/skills/task-routing/references/chat-transfer.md b/plugins/codex-task-routing/skills/task-routing/references/chat-transfer.md
new file mode 100644
index 0000000..1701ad6
--- /dev/null
+++ b/plugins/codex-task-routing/skills/task-routing/references/chat-transfer.md
@@ -0,0 +1,65 @@
+# Browser local transfer
+
+`chat_transfer.py` is a local loopback helper for an already approved
+`bundle.json`. It only renders that bundle's prepared `prompt` in a readonly
+Browser form and accepts one JSON response for one explicitly named
+`reply.json`. It does not open or control Chat, use an API, access an account,
+authentication, conversations, memories, MCP, or any external network.
+
+```text
+python <plugin-root>/scripts/chat_transfer.py --bundle bundle.json --reply reply.json --timeout 600
+```
+
+Its compact stdout metadata contains the random loopback URL, request ID,
+input hash, and lifecycle event. It never prints the prompt or response. The
+server binds only `127.0.0.1` on a random port; its unguessable URL token,
+strict Host and Origin checks, no-store headers, CSP, form-size limit, and
+link/reparse-safe input/output checks constrain the transfer. It stops after a
+valid submission, when its timeout expires, or with Ctrl+C.
+
+Open the printed URL using the supported Browser. The stable `Request prompt`
+readonly field and `Response JSON` field are plain HTML DOM controls, so a
+Browser automation surface can read and fill them without page-script
+evaluation. Move the prompt to a new, approved temporary normal Chat only
+after the parent has confirmed the content is authorized for that destination,
+the selected UI model is `6 Pro`, and the temporary-Chat conditions hold. The
+helper does not make those checks and does not send anything itself.
+
+Choose supported Browser operations that do not automatically emit the full
+page on navigation. For example, once the documented Browser runtime is
+initialized, create a tab with its `tabs.new()` method and navigate with
+`goto()`, then read only the relevant DOM fields. A wrapper that automatically
+prints its initial AX tree can echo the entire prompt and defeat this saving.
+Keep the prompt and final answer in Browser-runtime variables; report only
+length, equality, selected model, and the save result to the parent. Never
+read browser storage, session state or private application APIs to move data.
+
+If the Chat UI adds a connector chip while selecting an approved source, keep
+the prompt-text equality check separate from the chip's UI verification. UI
+sentinels or spacing added by that chip are not part of the prepared prompt
+and must not be copied into it.
+
+Some Chat composer versions convert source URLs to chips and add display-only
+newlines to `innerText`. Do not blindly trim or remove newlines to force a
+match. Inspect the current editable DOM once; when it uses observed paragraph,
+text and `BR` nodes, serialize text nodes, preserve each `BR` as a newline,
+and preserve paragraph boundaries. Compare the exact prepared prompt plus only
+the separately observed initial connector-chip prefix. Unknown structure or
+an actual mismatch blocks sending. No element IDs or DOM schema are guaranteed
+across Chat UI updates; reuse the supported controls observed in that session.
+
+Read the completed assistant response from its observed DOM container into a
+variable. If it has exactly one Markdown JSON fence, remove only that fence
+mechanically; do not rewrite fields. Fill `Response JSON` from the variable
+and submit through the normal form. Do not print the full response as an
+intermediate step, send it to another model for formatting, or use a network
+request to Chat as a shortcut.
+
+Paste the final Chat answer as the required response JSON and choose `Save
+verified response`. The helper calls `chatgpt_route.validate_exchange` before
+writing. An identical retry is idempotent; a different value can never replace
+an existing reply. `Transfer status` distinguishes a saved server submission,
+the response format/correlation check, and `status: completed`. Even a
+completed status is not semantic acceptance: the parent must still assess the
+answer against the task's acceptance criteria, evidence, scope, and UI model
+confirmation.
diff --git a/plugins/codex-task-routing/skills/task-routing/references/chatgpt.md b/plugins/codex-task-routing/skills/task-routing/references/chatgpt.md
index 63de328..60d2689 100644
--- a/plugins/codex-task-routing/skills/task-routing/references/chatgpt.md
+++ b/plugins/codex-task-routing/skills/task-routing/references/chatgpt.md
@@ -22,6 +22,8 @@
 
 ## 送信前の材料と連携確認
 
+0.7.0では [接続の再利用と機械的な適合確認](chat-capabilities.md) を使い、接続の全件調査や分類専用のモデル呼出しを繰り返しません。必要な操作・対象範囲・有効期限だけ照合します。大きな依頼と回答は [ローカル転送補助](chat-transfer.md) で本文を変数のまま移し、ID/hashを機械照合できます。いずれもopt-in、送信対象の承認、UIモデル確認、品質受入を代替しません。
+
 AIが新規依頼を組み立てる時は `handoff` の6項目をそろえます。これはAIの運用手順です。CLIのスキーマは既存のprepare呼出しとの互換性を保つため省略も受理し、入力欄の充足や情報の十分さを保証しません。既存の `task` は目的と許可範囲、`materials` は必要な材料、`acceptance_criteria` は受入条件です。専用欄は指示の抜けを防ぐもので、資料や実行権限の存在を証明しません。既存の承認を確認し、不足がある場合だけ必要なデータと送信先を特定して確認します。
 
 | handoff項目 | 指定する内容 |
@@ -33,7 +35,7 @@ AIが新規依頼を組み立てる時は `handoff` の6項目をそろえます
 | return_format | 回答JSONのresult/evidence内に求める構成、比較表・根拠・反例など |
 | stop_conditions | 情報不足・接続不能・範囲拡大など親へ戻す条件 |
 
-MCPが必要な仕事では、この依頼に必要な対応だけを整理します。全接続先の一覧収集や、確認のためだけの無関係な取得はしません。
+MCPが必要な仕事では、この依頼に必要な対応だけを整理します。利用者が接続一覧の調査を依頼した場合はメタデータを確認して再利用用に記録できます。通常の仕事では全接続先の一覧収集を繰り返さず、確認のためだけの無関係な取得はしません。
 
 | 必要情報・操作 | Chat側の承認済みツール | 確認状態 | 回収する証拠 |
 | --- | --- | --- | --- |
@@ -47,7 +49,7 @@ MCPが必要な仕事では、この依頼に必要な対応だけを整理し
 2. サポートされたBrowser操作で `https://chatgpt.com/` を開き、通常Chatから新しい一時Chatを作成します。既存の下書きがある入力欄を使わず、ログイン入力は利用者に任せます。送信直前に、Chatが選択されWorkが未選択であること、一時Chatであること、UIのモデル表示が `6 Pro` であることを確認します。選択・確認できなければ送信しません。
 3. 必要なら一時Chatのプラスメニューで、既存の承認済みプラグインまたはMCPを選択します。新しい接続・権限・独自MCPサーバーの追加はこの設定では承認されません。接続は材料の受け渡し方法の選択肢であり、一時Chatの起動やモデル指定の手段ではありません。
 4. `python <plugin-root>/scripts/chatgpt_route.py prepare --input request.json` で依頼を1回だけ準備します。出力された `prompt` を初回の実依頼として送ります。接続確認だけのハンドシェイクは送りません。
-5. `prompt` が複数行なら、`typeText` で打ち込まず、サポートされた `paste({format:'text'})` または対応した `fill` で入力します。送信前に入力欄の全文が `prompt` と一致し、既存の文面が混ざっていないことを確認してから、送信ボタンを1回だけ使います。確認できない場合は送信しません。
+5. `prompt` が複数行なら、`typeText` で打ち込まず、サポートされた `paste({format:'text'})` または対応した `fill` で入力します。送信前に入力欄の本文が `prompt` と一致し、既存の文面が混ざっていないことを確認してから、送信ボタンを1回だけ使います。選択したプラグインのチップが入力欄に含まれるUIでは、観測済みのチップだけを本文とは別に確認し、本文を任意に削除・trimして照合しません。確認できない場合は送信しません。
 6. サポートされたBrowser操作で同じ一時Chatの最終回答を取得します。`send_message_to_thread` と `read_thread` は一時Chatに使いません。前者は内部で `isTemporaryChat: false` を固定するためです。30秒程度から間隔を広げて完了を待ち、経過を利用者へ伝えます。10分で回答がなければ未完了とし、必要ならサポートされたhandoff機能でタブを保持します。回答待ちや取得失敗で同じ依頼を再送したり、処理中のタブを勝手に閉じたりしません。
 7. 最終回答だけを明示したローカル作業場所へ `reply.json` として保存し、`python <plugin-root>/scripts/chatgpt_route.py validate --request bundle.json --response reply.json` で `request_id` と `input_sha256` を照合します。`blocked` と `missing_input` は成功ではありません。親は合格条件、出典と対象範囲、重要な反例を確認し、有効な根拠を再利用して必要な成果物だけを保存します。同じ分析を全面的にやり直す検収を常態化させません。
 8. UIの `6 Pro` 表示を回答取得後にも確認します。モデル表示が変わった場合は受入を止め、別モデルの結果として扱います。バックエンドのモデルIDは不明のままにし、自己申告で補完しません。
@@ -96,4 +98,6 @@ python <plugin-root>/scripts/chatgpt_route.py validate --request bundle.json --r
 
 ## 確認済みの範囲と制約
 
-この版の自動テストは設定の既定値・legacy互換・依頼IDと材料hashの照合を対象にします。実行ごとに、Browser上で通常Chat、一時Chat、`6 Pro` の表示を送信前後で確認します。Browserを使わずに通常Chatの一時Chatを新規作成し、モデルを指定する正規の外部入口は、この版では扱いません。今後の課題は、必要な正規外部入口の提供と、認証付きMCP常用接続の整備です。
+通常ChatのGPT-6 ProとWork/Codexは別の利用枠です。WorkはCodexと共有します。通常Chatへの振分けでも、rootの準備・Browser操作・検収はCodex側の負担として残ります。GPT-6 Proは無制限ではなく、上限時の自動モデル変更もあるため、送信前後の6 Pro確認を省略しません。第三者サービスの料金・ツール固有制限まで無料になるとは扱いません。[公式モデルと利用枠](https://help.openai.com/en/articles/20001354)、[Work/Codex料金](https://learn.chatgpt.com/docs/pricing)。2026-09-14確認。上限値をコードへ固定せず、利用時のUIを優先します。
+
+この版の自動テストは設定の既定値・legacy互換・依頼IDと材料hashの照合、接続台帳の範囲と期限、ローカル転送のHTTP境界と保存を対象にします。実行ごとに、Browser上で通常Chat、一時Chat、`6 Pro` の表示を送信前後で確認します。Browserを使わずに通常Chatの一時Chatを新規作成し、モデルを指定する正規の外部入口は、この版では扱いません。MCPの実利用可否は各接続・操作・対象範囲で確認し、全プラグインで同じ能力が使えるとは保証しません。
diff --git a/scripts/check_package.py b/scripts/check_package.py
index fea89b5..fd6200b 100644
--- a/scripts/check_package.py
+++ b/scripts/check_package.py
@@ -20,6 +20,8 @@ def check():
     for name in ['LICENSE', 'scripts/routing.py', 'scripts/chatgpt_route.py', 'scripts/updater.py', 'scripts/updater_entry.py', 'scripts/install_updater.py', 'hooks/hooks.json', 'defaults/config.json', 'skills/task-routing/SKILL.md', 'skills/task-routing/references/configuration.md', 'skills/task-routing/references/chatgpt.md', 'skills/task-routing/references/observation.md', 'skills/task-routing/references/observation-example.json']:
         assert (PLUGIN / name).is_file(), name
     hooks = json.loads((PLUGIN / 'hooks/hooks.json').read_text(encoding='utf-8'))['hooks']
+    for name in ['scripts/chat_plan.py', 'scripts/chat_transfer.py', 'skills/task-routing/references/chat-capabilities.md', 'skills/task-routing/references/chat-transfer.md']:
+        assert (PLUGIN / name).is_file(), name
     assert set(hooks) == {'SessionStart', 'SubagentStart'}
     for groups in hooks.values():
         for group in groups:
diff --git a/tests/test_chat_boundary_regressions.py b/tests/test_chat_boundary_regressions.py
new file mode 100644
index 0000000..b868cf0
--- /dev/null
+++ b/tests/test_chat_boundary_regressions.py
@@ -0,0 +1,69 @@
+"""Regressions for untrusted JSON and capability timestamp boundaries."""
+import copy
+from datetime import datetime, timezone
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+SCRIPTS = Path(__file__).resolve().parents[1] / 'plugins/codex-task-routing/scripts'
+sys.path.insert(0, str(SCRIPTS))
+import chat_plan
+import chat_transfer
+import chatgpt_route
+
+
+class ChatBoundaryRegressions(unittest.TestCase):
+    def setUp(self):
+        self.request = dict(task='Review supplied text.', materials='Fixture only.',
+                            acceptance_criteria=['Return evidence.'])
+        self.bundle = chatgpt_route.prepare_payload(self.request)
+        self.response = dict(request_id=self.bundle['request_id'],
+                             input_sha256=self.bundle['input_sha256'],
+                             status='completed', result='Valid result.', evidence=[])
+
+    def test_lone_surrogates_rejected_before_preparing_or_saving(self):
+        for surrogate in ('\ud800', '\udfff'):
+            for field in ('task', 'materials', 'acceptance_criteria'):
+                invalid = copy.deepcopy(self.request)
+                invalid[field] = [surrogate] if field == 'acceptance_criteria' else surrogate
+                with self.subTest(surrogate=repr(surrogate), field=field):
+                    with self.assertRaisesRegex(chatgpt_route.ChatRouteError, 'valid Unicode'):
+                        chatgpt_route.prepare_payload(invalid)
+            for field in ('result', 'evidence'):
+                invalid = dict(self.response)
+                invalid[field] = [surrogate] if field == 'evidence' else surrogate
+                with self.subTest(surrogate=repr(surrogate), field=field):
+                    root = Path(__file__).resolve().parent / '.tmp-runtime-tests'
+                    root.mkdir(exist_ok=True)
+                    with tempfile.TemporaryDirectory(dir=root) as directory:
+                        reply = Path(directory) / 'reply.json'
+                        state = chat_transfer.TransferState(bundle=self.bundle, reply_path=reply)
+                        with self.assertRaisesRegex(chat_transfer.ChatTransferError, 'valid Unicode'):
+                            state.save_response_text(json.dumps(invalid))
+                        self.assertFalse(reply.exists())
+                        self.assertIsNone(state.response_status)
+                        # The same request can still accept a corrected response.
+                        self.assertTrue(state.save_response(self.response)[1])
+
+    def test_valid_supplementary_unicode_roundtrips(self):
+        response = dict(self.response, result='Supplementary: \U0001f680')
+        parsed = chatgpt_route._parse_json_object(json.dumps(response), purpose='response')
+        self.assertEqual(chatgpt_route.validate_exchange(self.bundle, parsed)['status'], 'completed')
+        self.assertEqual(json.loads(chat_transfer._canonical_response_bytes(parsed)), response)
+
+    def test_inventory_upper_date_boundary_has_no_overflow(self):
+        inventory = dict(schema_version=1, surface='chat-temporary', model='6 Pro',
+                         installed_plugins=['Example'], capabilities=[dict(
+                             tool='Example', operation='read', scope='fixture', state='verified',
+                             observed_at='9999-12-30T00:00:00Z', expires_at='9999-12-31T00:00:00Z')])
+        self.assertEqual(chat_plan.validate_inventory(inventory), inventory)
+        for timestamp in ('0001-01-01T00:00:00+01:00', '9999-12-31T23:59:59-01:00'):
+            with self.subTest(timestamp=timestamp):
+                with self.assertRaises(chatgpt_route.ChatRouteError):
+                    chat_plan._time(timestamp)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_chat_plan.py b/tests/test_chat_plan.py
new file mode 100644
index 0000000..af2237d
--- /dev/null
+++ b/tests/test_chat_plan.py
@@ -0,0 +1,83 @@
+import copy
+from datetime import datetime, timezone
+from pathlib import Path
+import sys
+import unittest
+
+SCRIPTS = Path(__file__).resolve().parents[1] / 'plugins/codex-task-routing/scripts'
+sys.path.insert(0, str(SCRIPTS))
+import chat_plan as cp
+
+
+class ChatPlanTests(unittest.TestCase):
+    def setUp(self):
+        self.now = datetime(2026, 9, 14, tzinfo=timezone.utc)
+        self.need = {'tool': 'GitHub', 'operation': 'fetch_file', 'scope': 'example/project'}
+        self.facts = dict(substantial=True, parent_has_independent_work=True,
+                          materials_approved=True, handoff_proportionate=True,
+                          requires_repeated_local_access=False, needs=[self.need])
+        self.inventory = dict(schema_version=1, surface='chat-temporary', model='6 Pro',
+                              installed_plugins=['GitHub'], capabilities=[dict(
+                                  **self.need, state='verified', observed_at='2026-09-13T00:00:00Z',
+                                  expires_at='2026-09-15T00:00:00Z')])
+
+    def run_plan(self):
+        return cp.plan(self.facts, self.inventory, now=self.now)
+
+    def test_matching_verified_capability_is_only_a_candidate(self):
+        self.assertEqual(self.run_plan()['route'], 'chat_candidate')
+
+    def test_unknown_advertised_and_expired_require_preflight(self):
+        for state in ('advertised', 'verified'):
+            self.inventory['capabilities'][0]['state'] = state
+            if state == 'verified':
+                self.inventory['capabilities'][0]['expires_at'] = self.now.isoformat()
+            self.assertEqual(self.run_plan()['route'], 'preflight_needed')
+        self.inventory['capabilities'] = []
+        self.assertEqual(self.run_plan()['checks'][0]['state'], 'unknown')
+
+    def test_scope_and_operation_not_generalized(self):
+        for key in ('scope', 'operation'):
+            changed = copy.deepcopy(self.facts)
+            changed['needs'][0][key] = 'different'
+            self.assertEqual(cp.plan(changed, self.inventory, now=self.now)['route'], 'preflight_needed')
+
+    def test_no_required_tools_needs_no_inventory(self):
+        self.facts['needs'] = []
+        self.assertEqual(cp.plan(self.facts)['route'], 'chat_candidate')
+
+    def test_gates_and_blocked(self):
+        for key in ('substantial', 'parent_has_independent_work', 'handoff_proportionate'):
+            changed = {**self.facts, key: False}
+            self.assertEqual(cp.plan(changed)['route'], 'codex')
+        self.facts['requires_repeated_local_access'] = True
+        self.assertEqual(self.run_plan()['route'], 'codex')
+        self.facts['requires_repeated_local_access'] = False
+        self.facts['materials_approved'] = False
+        self.assertEqual(self.run_plan()['route'], 'not_ready')
+        self.facts['materials_approved'] = True
+        self.inventory['capabilities'][0]['state'] = 'blocked'
+        self.assertEqual(self.run_plan()['route'], 'not_ready')
+
+    def test_malformed_inputs_fail_closed(self):
+        for update in ({'model': 'Work'}, {'schema_version': True}, {'unknown': 1}):
+            with self.assertRaises(cp.ChatRouteError):
+                cp.validate_inventory({**self.inventory, **update})
+        for key in ('state', 'observed_at', 'expires_at'):
+            broken = copy.deepcopy(self.inventory)
+            broken['capabilities'][0][key] = 'invalid'
+            with self.assertRaises(cp.ChatRouteError):
+                cp.validate_inventory(broken)
+        with self.assertRaises(cp.ChatRouteError):
+            cp.plan({**self.facts, 'substantial': 'true'})
+
+    def test_future_observation_and_long_lifetime(self):
+        self.inventory['capabilities'][0].update(observed_at='2026-09-15T00:00:00Z', expires_at='2026-09-16T00:00:00Z')
+        self.assertEqual(self.run_plan()['checks'][0]['state'], 'stale')
+        self.inventory['capabilities'][0]['expires_at'] = '2026-10-01T00:00:00Z'
+        with self.assertRaises(cp.ChatRouteError):
+            self.run_plan()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_chat_transfer.py b/tests/test_chat_transfer.py
new file mode 100644
index 0000000..51934d3
--- /dev/null
+++ b/tests/test_chat_transfer.py
@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+import shutil
+import socket
+import sys
+import unittest
+import uuid
+import threading
+import time
+from contextlib import redirect_stdout
+from pathlib import Path
+from types import SimpleNamespace
+from urllib.parse import urlencode
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ROUTE_SCRIPT = ROOT / "plugins" / "codex-task-routing" / "scripts" / "chatgpt_route.py"
+TRANSFER_SCRIPT = ROOT / "plugins" / "codex-task-routing" / "scripts" / "chat_transfer.py"
+
+
+def load_module(name: str, path: Path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+chatgpt_route = load_module("chatgpt_route_for_transfer_test", ROUTE_SCRIPT)
+chat_transfer = load_module("chat_transfer_under_test", TRANSFER_SCRIPT)
+
+
+class ChatTransferTestCase(unittest.TestCase):
+    def setUp(self) -> None:
+        fixture_root = Path(__file__).resolve().parent / ".tmp-runtime-tests"
+        fixture_root.mkdir(exist_ok=True)
+        self.base = fixture_root / f"chat-transfer-{uuid.uuid4().hex}"
+        self.base.mkdir()
+        self.bundle_path = self.base / "bundle.json"
+        self.reply_path = self.base / "reply.json"
+        request = {
+            "request_id": str(uuid.uuid4()),
+            "task": "Assess the supplied bounded evidence.",
+            "materials": "Supplied content only.",
+            "acceptance_criteria": ["Return a concise finding."],
+        }
+        self.bundle = chatgpt_route.prepare_payload(request)
+        self.bundle_path.write_text(json.dumps(self.bundle), encoding="utf-8")
+        self.state = chat_transfer.TransferState.from_paths(self.bundle_path, self.reply_path)
+        self.state.token = "test-capability-token"
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.base, ignore_errors=True)
+
+    def response(self, *, result: str = "bounded finding", request_id: str | None = None) -> dict[str, object]:
+        return {
+            "request_id": request_id or self.bundle["request_id"],
+            "input_sha256": self.bundle["input_sha256"],
+            "status": "completed",
+            "result": result,
+            "evidence": ["supplied evidence"],
+        }
+
+    def test_page_has_stable_dom_fields_without_submitting_the_prompt(self) -> None:
+        document = chat_transfer._page(
+            self.state, status=chat_transfer._response_status_message(None)
+        ).decode("utf-8")
+        self.assertIn('<label for="request_prompt">Request prompt</label>', document)
+        self.assertIn('id="request_prompt" readonly', document)
+        self.assertNotIn('name="request_prompt"', document)
+        self.assertIn('<label for="response_json">Response JSON</label>', document)
+        self.assertIn('Save verified response', document)
+        self.assertIn('Transfer status', document)
+        self.assertIn('name="referrer" content="same-origin"', document)
+        self.assertIn('&lt;canonical UUID&gt;', document)
+        self.assertIn("default-src 'none'", chat_transfer.CONTENT_SECURITY_POLICY)
+        self.assertIn("form-action 'self'", chat_transfer.CONTENT_SECURITY_POLICY)
+
+    def test_unauthenticated_paths_do_not_receive_prompt_or_token(self) -> None:
+        error_document = chat_transfer._minimal_error_page().decode("utf-8")
+        self.assertNotIn(self.bundle["prompt"], error_document)
+        self.assertNotIn(self.state.token, error_document)
+        self.assertNotIn("Response JSON", error_document)
+        self.assertFalse(chat_transfer._path_is_valid("/transfer/not-the-token", self.state.token))
+        self.assertFalse(chat_transfer._path_is_valid(f"/transfer/{self.state.token}?x=1", self.state.token))
+
+    def test_wrong_host_and_origin_are_rejected(self) -> None:
+        port = 32123
+        self.assertFalse(chat_transfer._host_is_valid("localhost:32123", port))
+        self.assertFalse(chat_transfer._origin_is_valid("http://127.0.0.1:1", port))
+        self.assertTrue(chat_transfer._host_is_valid("127.0.0.1:32123", port))
+        self.assertTrue(chat_transfer._origin_is_valid("http://127.0.0.1:32123", port))
+
+    def test_oversize_post_is_rejected_without_writing_a_reply(self) -> None:
+        self.assertGreater(chat_transfer.MAX_POST_BYTES, chatgpt_route.MAX_JSON_BYTES)
+        with self.assertRaisesRegex(chat_transfer.ChatTransferError, "size limit"):
+            chat_transfer._parse_post_length(str(chat_transfer.MAX_POST_BYTES + 1))
+        body = urlencode({"response_json": "x" * (chatgpt_route.MAX_JSON_BYTES + 1)}).encode("utf-8")
+        # The request length is checked before this parser in the handler; this
+        # checks the decoded response size independently.
+        with self.assertRaisesRegex(chat_transfer.ChatTransferError, "size limit"):
+            chat_transfer._response_text_from_form(body)
+        self.assertFalse(self.reply_path.exists())
+
+    def test_wrong_request_id_is_not_saved(self) -> None:
+        with self.assertRaisesRegex(chat_transfer.ChatTransferError, "request_id"):
+            self.state.save_response(self.response(request_id=str(uuid.uuid4())))
+        self.assertFalse(self.reply_path.exists())
+
+    def test_json_parse_status_reports_only_location_metadata(self) -> None:
+        invalid = '{"request_id": }'
+        with self.assertRaises(chat_transfer.ChatTransferError):
+            self.state.save_response_text(invalid)
+        status = self.state.transfer_status
+        self.assertIn("syntax line=1", status)
+        self.assertIn("column=16", status)
+        self.assertIn("character=U+007D", status)
+        self.assertIn("decoded_chars=16", status)
+        self.assertNotIn(invalid, status)
+
+    def test_http_boundary_and_verified_save(self) -> None:
+        # Feed wire bytes to the real HTTP parser and handler in memory.
+        # This keeps local HTTP filters and OS socket EOF behavior out of the
+        # protocol assertions; Browser transport is verified separately.
+        server = SimpleNamespace(state=self.state, server_port=32123, completed=False)
+        path = f'/transfer/{self.state.token}'
+
+        def request(method, target, body=b'', headers=None, *, close=True):
+            request_headers = {'Host': f'127.0.0.1:{server.server_port}'}
+            request_headers.update(headers or {})
+            if close:
+                request_headers['Connection'] = 'close'
+            if method == 'POST':
+                request_headers.setdefault('Content-Length', str(len(body)))
+            frame = [f'{method} {target} HTTP/1.1']
+            frame.extend(f'{name}: {value}' for name, value in request_headers.items())
+            raw_request = ('\r\n'.join(frame) + '\r\n\r\n').encode('ascii') + body
+            received = bytearray()
+            connection = SimpleNamespace(
+                makefile=lambda *_args: io.BytesIO(raw_request),
+                sendall=received.extend,
+            )
+            chat_transfer.TransferHandler(connection, ('127.0.0.1', 0), server)
+            head, separator, body_bytes = bytes(received).partition(b'\r\n\r\n')
+            self.assertTrue(separator, 'server must return a complete HTTP response')
+            lengths = [line.split(b':', 1)[1].strip() for line in head.split(b'\r\n')
+                       if line.lower().startswith(b'content-length:')]
+            self.assertEqual(len(lengths), 1)
+            self.assertEqual(len(body_bytes), int(lengths[0]))
+            status_line = head.split(b'\r\n', 1)[0].decode('ascii')
+            self.assertRegex(status_line, r'^HTTP/1\.0 [1-5]\d\d ')
+            self.assertIn(b'\r\nConnection: close\r\n', b'\r\n' + head + b'\r\n')
+            self.assertIn(b'\r\nReferrer-Policy: same-origin\r\n', b'\r\n' + head + b'\r\n')
+            return int(status_line.split()[1]), body_bytes.decode('utf-8')
+
+        for method, target, headers in [
+            ('GET', '/wrong-token', {}),
+            ('GET', path, {'Host': 'attacker.invalid'}),
+            ('POST', path, {'Origin': 'https://attacker.invalid'}),
+        ]:
+            with self.subTest(method=method, headers=headers):
+                status, body = request(method, target, headers=headers)
+                self.assertGreaterEqual(status, 400)
+                self.assertNotIn('Supplied content only.', body)
+                self.assertNotIn(self.state.token, body)
+        headers = {'Origin': 'http://127.0.0.1:32123', 'Content-Type': 'application/x-www-form-urlencoded'}
+        invalid = self.response(request_id=str(uuid.uuid4()))
+        status, _ = request('POST', path, urlencode({'response_json': json.dumps(invalid)}).encode('utf-8'), headers)
+        self.assertEqual(status, 422)
+        self.assertFalse(self.reply_path.exists())
+        # The Browser form percent-encodes UTF-8; keep this near the live
+        # response size while exercising a request without Connection: close.
+        valid = self.response(result="確認" * 1700)
+        status, body = request(
+            'POST',
+            path,
+            urlencode({'response_json': json.dumps(valid, ensure_ascii=False)}).encode('utf-8'),
+            headers,
+            close=False,
+        )
+        self.assertEqual(status, 200)
+        self.assertIn('Parent semantic acceptance is still required', body)
+        self.assertEqual(json.loads(self.reply_path.read_text(encoding='utf-8')), valid)
+
+    def test_form_rejects_invalid_utf8_and_duplicate_field(self) -> None:
+        for body in (b'response_json=%FF', b'response_json=a&response_json=b'):
+            with self.assertRaises(chat_transfer.ChatTransferError):
+                chat_transfer._response_text_from_form(body)
+
+    def test_same_response_retry_is_idempotent(self) -> None:
+        response = self.response()
+        validated, created = self.state.save_response(response)
+        self.assertTrue(created)
+        self.assertEqual(validated["status"], "completed")
+        first = self.reply_path.read_bytes()
+        validated, created = self.state.save_response(response)
+        self.assertFalse(created)
+        self.assertEqual(validated["status"], "completed")
+        self.assertEqual(self.reply_path.read_bytes(), first)
+        self.assertEqual(
+            chatgpt_route.validate_exchange(
+                self.bundle, json.loads(self.reply_path.read_text(encoding="utf-8"))
+            )["status"],
+            "completed",
+        )
+
+    def test_existing_different_response_is_never_overwritten(self) -> None:
+        original = self.response(result="first result")
+        self.reply_path.write_text(json.dumps(original), encoding="utf-8")
+        before = self.reply_path.read_bytes()
+        with self.assertRaisesRegex(chat_transfer.ChatTransferError, "already exists with a different response"):
+            self.state.save_response(self.response(result="second result"))
+        self.assertEqual(self.reply_path.read_bytes(), before)
+
+    def test_completed_status_is_not_semantic_acceptance(self) -> None:
+        status = chat_transfer._response_status_message("completed", created=True)
+        self.assertIn("Response status is completed", status)
+        self.assertIn("Parent semantic acceptance is still required", status)
+
+    def test_timeout_stops_without_echoing_prompt_or_response(self) -> None:
+        with redirect_stdout(io.StringIO()) as stdout:
+            self.assertEqual(chat_transfer.serve_transfer(self.state, timeout_seconds=1), 3)
+        metadata = stdout.getvalue()
+        self.assertIn('"event": "listening"', metadata)
+        self.assertIn('"event": "timeout"', metadata)
+        self.assertNotIn(self.bundle["prompt"], metadata)
+        self.assertNotIn("bounded finding", metadata)
+
+    def test_absolute_deadline_interrupts_continuous_partial_request(self) -> None:
+        server = chat_transfer.create_server(self.state)
+        application, client = socket.socketpair()
+        application.settimeout(5)
+        server.deadline = time.monotonic() + 0.15
+        worker = threading.Thread(target=server.finish_request,
+                                  args=(application, ('127.0.0.1', 0)), daemon=True)
+        worker.start()
+        try:
+            client.sendall(b'GET /transfer/')
+            until = time.monotonic() + 1
+            while worker.is_alive() and time.monotonic() < until:
+                try:
+                    client.sendall(b'x')
+                except OSError:
+                    break
+                time.sleep(0.01)
+            worker.join(timeout=0.3)
+            self.assertFalse(worker.is_alive(), 'traffic must not extend the absolute deadline')
+            self.assertFalse(self.reply_path.exists())
+        finally:
+            client.close()
+            application.close()
+            server.server_close()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_routing.py b/tests/test_routing.py
index 47ff17e..23b5e8b 100644
--- a/tests/test_routing.py
+++ b/tests/test_routing.py
@@ -11,7 +11,7 @@
 import uuid
 from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 
 SCRIPT = Path(__file__).resolve().parents[1] / "plugins" / "codex-task-routing" / "scripts" / "routing.py"
@@ -204,9 +204,86 @@ def winner_publish_then_signal_collision(source, destination):
                 self.assertEqual(cache.name, policy.content_hash)
                 self.assertTrue(routing._is_managed_directory(cache))
 
-        with patch.object(routing.os, "rename", side_effect=OSError(errno.EACCES, "denied")):
+        denied_calls = 0
+
+        def denied(source, destination):
+            nonlocal denied_calls
+            denied_calls += 1
+            raise OSError(errno.EACCES, "denied")
+
+        with patch.object(routing.os, "rename", side_effect=denied):
             with self.assertRaises(routing.RoutingError):
                 routing._cache_directory(self.base / "home-denied", policy)
+        self.assertEqual(denied_calls, 1)
+
+    @unittest.skipUnless(os.name == "nt", "Windows rename error mapping")
+    def test_cache_rename_retries_only_bounded_windows_access_candidates(self) -> None:
+        policy = self.policy()
+        original_rename = routing.os.rename
+        transient_home = self.base / "home-transient-access"
+        calls = 0
+
+        def fail_twice_then_publish(source, destination):
+            nonlocal calls
+            calls += 1
+            if calls < routing.CACHE_RENAME_MAX_ATTEMPTS:
+                raise OSError(errno.EACCES, "temporary access", None, 5)
+            original_rename(source, destination)
+
+        with (
+            patch.object(routing.os, "rename", side_effect=fail_twice_then_publish),
+            patch.object(routing.time, "sleep") as sleep,
+        ):
+            cache = routing._cache_directory(transient_home, policy)
+        self.assertEqual(cache.name, policy.content_hash)
+        self.assertEqual(calls, routing.CACHE_RENAME_MAX_ATTEMPTS)
+        self.assertEqual(
+            sleep.call_args_list,
+            [
+                call(routing.CACHE_RENAME_RETRY_SECONDS),
+                call(routing.CACHE_RENAME_RETRY_SECONDS * 2),
+            ],
+        )
+
+        persistent_home = self.base / "home-persistent-access"
+        persistent_calls = 0
+
+        def permanently_denied(source, destination):
+            nonlocal persistent_calls
+            persistent_calls += 1
+            raise OSError(errno.EACCES, "persistent access", None, 5)
+
+        with (
+            patch.object(routing.os, "rename", side_effect=permanently_denied),
+            patch.object(routing.time, "sleep") as sleep,
+        ):
+            with self.assertRaises(routing.RoutingError):
+                routing._cache_directory(persistent_home, policy)
+        self.assertEqual(persistent_calls, routing.CACHE_RENAME_MAX_ATTEMPTS)
+        self.assertEqual(sleep.call_count, routing.CACHE_RENAME_MAX_ATTEMPTS - 1)
+
+        mismatch_home = self.base / "home-mismatched-winner"
+        mismatch_calls = 0
+
+        def publish_mismatch_then_signal_access(source, destination):
+            nonlocal mismatch_calls
+            mismatch_calls += 1
+            destination.mkdir()
+            (destination / "foreign.txt").write_text("do not overwrite", encoding="utf-8")
+            raise OSError(errno.EACCES, "temporary access", None, 5)
+
+        with (
+            patch.object(routing.os, "rename", side_effect=publish_mismatch_then_signal_access),
+            patch.object(routing.time, "sleep") as sleep,
+        ):
+            with self.assertRaises(routing.RoutingError):
+                routing._cache_directory(mismatch_home, policy)
+        self.assertEqual(mismatch_calls, 1)
+        self.assertEqual(sleep.call_count, 0)
+        self.assertEqual(
+            (mismatch_home / routing.CACHE_RELATIVE / policy.content_hash / "foreign.txt").read_text(encoding="utf-8"),
+            "do not overwrite",
+        )
 
     def test_mutated_cache_is_not_reused_as_effective_policy(self) -> None:
         policy = self.policy()