fix(codex): restore hardened WeChat send backend lineage

.agent/dispatch/V2R1-PERSONAL-WEIXIN-IMMEDIATE-WORK-UNITS.json

@@ -0,0 +1,32 @@
+{
+  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
+  "created_at": "2026-07-17T14:13:38+00:00",
+  "schema_version": 1,
+  "state_writer": "openclaw",
+  "work_units": [
+    {
+      "fallback": "codex",
+      "gate": "R0_RUNNER_LIVE_RECOVERY",
+      "id": "R0-DISCOVER-SECRET-PROVIDER",
+      "primary": "openclaw",
+      "status": "READY",
+      "track": "R_RUNNER"
+    },
+    {
+      "fallback": "codex",
+      "gate": "P2_REAL_B3_EVIDENCE",
+      "id": "P2-INVENTORY-REAL-EVIDENCE",
+      "primary": "openclaw",
+      "status": "READY",
+      "track": "P_PERSONAL_WEIXIN"
+    },
+    {
+      "fallback": "openclaw",
+      "gate": "P1_REPAIR3_CODE",
+      "id": "P1-POST-CONTRACT-REACK",
+      "primary": "codex",
+      "status": "READY",
+      "track": "P_PERSONAL_WEIXIN"
+    }
+  ]
+}

.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml

@@ -0,0 +1,157 @@
+{
+  "adaptive_dispatch": {
+    "codex_first": [
+      "complex_code",
+      "security_boundary",
+      "architecture",
+      "high_risk_review",
+      "strong_model_materially_better"
+    ],
+    "default_orchestrator": "openclaw",
+    "dual_required": [
+      "release_gate",
+      "crypto_or_identity_change",
+      "production_migration_or_rollback"
+    ],
+    "max_attempts_per_work_unit": 2,
+    "on_timeout": [
+      "mark_TIMEOUT",
+      "preserve_logs",
+      "switch_agent"
+    ],
+    "openclaw_first": [
+      "quality_close",
+      "large_or_repetitive",
+      "long_running",
+      "infra_or_git_inventory",
+      "cost_sensitive"
+    ],
+    "single_state_writer_lease_required": true,
+    "strict_role_partition": false,
+    "timeout_can_pass": false
+  },
+  "authoritative_state": {
+    "b3": "BLOCKED_EVIDENCE_REQUIRED",
+    "b4": "NOT_STARTED",
+    "baseline": "V1R4C6",
+    "production_release": "NOT_STARTED",
+    "release": "paused",
+    "repair3": {
+      "base": "8d9c6da90fc8caa69b691d243c6fefb40b6b0575",
+      "candidate": "PASS_CANDIDATE_PREPARED",
+      "commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4",
+      "exact4": "PASS",
+      "result": "49a9c2e0cc898ecf2bbd45f454e007644d7f3045"
+    },
+    "runner": {
+      "artifact_sha256": "85c472c05a1e55a5899b215d3cd41df53912787de529ac9cb3d5308da4158130",
+      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
+      "live": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
+      "offline": "PASS",
+      "result": "e789632e359e9232d8ae081e7c4f23ed7b615a9d"
+    }
+  },
+  "contract_id": "V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT",
+  "facts_policy": {
+    "accept": [
+      "committed_git_tree",
+      "full_oid",
+      "worktree_bare_gitea_oid_match",
+      "test_log_with_rc",
+      "immutable_digest",
+      "result_only_commit",
+      "live_evidence"
+    ],
+    "reject": [
+      "verbal_claim",
+      "readme_only",
+      "local_only_commit",
+      "environment_variable_as_secret_ref",
+      "compile_only",
+      "timeout_as_pass",
+      "unrelated_repo",
+      "fabricated_evidence"
+    ]
+  },
+  "forbidden_git": [
+    "reset",
+    "amend",
+    "rebase",
+    "force_push",
+    "direct_main_push",
+    "direct_bare_ref_mutation",
+    "unapproved_gitea_db_mutation",
+    "reuse_accident_worktree",
+    "delete_live_worktree_to_clear_collision"
+  ],
+  "generated_at": "2026-07-17T14:13:38+00:00",
+  "handoff": {
+    "ack": "CONTRACT_ACK=<full_contract_commit_oid>",
+    "advance": "independent_verification_required",
+    "all_agents_match": true
+  },
+  "join_gate": {
+    "requires": [
+      "RUNNER_RECOVERED",
+      "B3_PASS",
+      "B4_PASS",
+      "INTEGRATION_PASS"
+    ]
+  },
+  "model_policy": {
+    "codex": "strongest_model_actually_available_in_runtime",
+    "gpt_5_6": "use_only_if_exposed",
+    "no_unverified_model_claims": true,
+    "openclaw_default": "DeepSeek V4 Flash"
+  },
+  "owner_authorization": {
+    "adaptive_agent_assignment_authorized": true,
+    "agent_substitution_authorized": true,
+    "autonomous_execution": true,
+    "may_bypass_release_controls": false,
+    "may_bypass_secret_controls": false,
+    "may_fabricate_evidence": false,
+    "user_step_by_step_confirmation_required": false
+  },
+  "release_gates": [
+    "MAIN_MERGE",
+    "STAGING_AND_CANARY",
+    "PRODUCTION"
+  ],
+  "schema_version": 3,
+  "state_rules": [
+    "tracks_run_parallel",
+    "runner_block_does_not_block_offline_work_or_evidence_inventory",
+    "join_requires_both_tracks",
+    "failure_stops_dependents",
+    "production_remains_V1R4C6_paused_until_production_pass"
+  ],
+  "status": "ACTIVE",
+  "tracks": {
+    "P_PERSONAL_WEIXIN": {
+      "b3": "BLOCKED_EVIDENCE_REQUIRED",
+      "b4": "NOT_STARTED",
+      "parallel": true,
+      "repair3": "PASS_CANDIDATE_PREPARED",
+      "repair3_commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4"
+    },
+    "R_RUNNER": {
+      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
+      "current": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
+      "fallback": "codex",
+      "parallel": true,
+      "primary": "openclaw",
+      "required": [
+        "artifact_verified",
+        "approved_secret_provider_contract",
+        "owner_probe_verified",
+        "service_healthy",
+        "heartbeat_healthy",
+        "gateway_auth",
+        "durable_task_create_read_cancel",
+        "execution_channel",
+        "live_rollback"
+      ]
+    }
+  }
+}

.agent/state/V2R1-PERSONAL-WEIXIN-PRODUCTION-STATE.json

@@ -0,0 +1,15 @@
+{
+  "b3": "BLOCKED_EVIDENCE_REQUIRED",
+  "b4": "NOT_STARTED",
+  "baseline": "V1R4C6",
+  "contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
+  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
+  "dispatch_commit": null,
+  "production_release": "NOT_STARTED",
+  "release": "paused",
+  "repair3": "PASS_CANDIDATE_PREPARED",
+  "runner": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
+  "schema_version": 1,
+  "single_state_writer": "openclaw",
+  "updated_at": "2026-07-17T14:13:38+00:00"
+}

.agent/tasks/codex/TASK-CODEX-V2R1-ADAPTIVE-PRODUCTION-EXECUTION-20260717.md

@@ -0,0 +1,13 @@
+# Codex V2R1 adaptive execution
+CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
+Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
+Execute P1-POST-CONTRACT-REACK now:
+- read-only verify Repair3 e110cda9968cadc5c7b7179ec5003b05aad11ce4
+- read-only verify RESULT 49a9c2e0cc898ecf2bbd45f454e007644d7f3045
+- verify lineage from 8d9c6da90fc8caa69b691d243c6fefb40b6b0575, exact4, tests, Gitleaks and RESULT-only scope
+- do not modify Repair3 code in this unit
+Act as fallback for OpenClaw only on timeout/unavailability. Use GPT-5.6 only if actually exposed; otherwise strongest visible model. Prefer OpenClaw when quality is comparable or volume is high; prefer Codex for complex/security-critical work.
+Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
+Report:
+CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=codex STATUS=<status> IMPLEMENTATION=<oid-or-NONE> RESULT=<oid-or-NONE> OID_MATCH=<YES-or-NO> BLOCKER=<reason-or-NONE>
+Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.

.agent/tasks/openclaw/TASK-OPENCLAW-V2R1-ADAPTIVE-PRODUCTION-ORCHESTRATION-20260717.md

@@ -0,0 +1,13 @@
+# OpenClaw V2R1 adaptive orchestration
+CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
+Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
+You are default orchestrator and sole authoritative state writer.
+Execute immediately:
+1. R0-DISCOVER-SECRET-PROVIDER: discover approved provider metadata and owner-probe source without reading or outputting secrets. Environment variables alone are not approved secret refs.
+2. P2-INVENTORY-REAL-EVIDENCE: inventory approved restricted B3 evidence locations; validate existing facts; never synthesize evidence.
+3. Independently verify Codex results before advancing.
+Prefer OpenClaw for comparable-quality, large, repetitive, long-running, infrastructure and cost-sensitive work. Prefer Codex strongest actually available model for complex code, security boundaries, architecture and high-risk review. On timeout mark TIMEOUT and switch agent; never infer PASS.
+Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
+Report:
+CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<reason-or-NONE>
+Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.

scripts/wechat_send_backend.py

@@ -0,0 +1,652 @@
+#!/usr/bin/env python3
+"""Hardened WeChat outbound backend for AIOS.
+
+Only the dedicated UCC adapter is a supported send backend.  OpenClaw channel
+discovery is retained for diagnostics, but generic OpenClaw chat or agent
+delivery is never promoted to WeChat delivery.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import selectors
+import signal
+import subprocess
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any, Dict, List, Optional, Tuple
+
+
+OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
+WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
+WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
+SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
+UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
+UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
+    "AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
+)
+UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
+_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
+UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
+    origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
+) or (
+    "http://127.0.0.1:18800",
+    "http://[::1]:18800",
+)
+
+MAX_TARGET_BYTES = 512
+MAX_CONTENT_BYTES = 32 * 1024
+MAX_IDENTIFIER_BYTES = 512
+MAX_METADATA_BYTES = 8 * 1024
+MAX_METADATA_DEPTH = 32
+MAX_METADATA_NODES = 1024
+MAX_RESPONSE_BYTES = 64 * 1024
+MAX_ENDPOINT_BYTES = 256
+_DEFAULT_TIMEOUT_SECONDS = 10
+_MAX_TIMEOUT_SECONDS = 30
+_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
+_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
+_MAX_PATH_DECODE_ROUNDS = 4
+_OPENCLAW_TIMEOUT_RETURN_CODE = 124
+_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
+_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
+
+
+def _bounded_timeout(value: str) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return _DEFAULT_TIMEOUT_SECONDS
+    return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
+
+
+REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
+
+
+class BackendResult:
+    def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
+        self.ok = ok
+        self.backend = backend
+        self.status = status
+        self.data = data
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
+
+
+class _BackendProtocolError(Exception):
+    """Internal marker whose details must never cross the module boundary."""
+
+
+def _openclaw_base_args() -> List[str]:
+    """Return the OpenClaw 2026.6.11-compatible command prefix.
+
+    Gateway/auth global flags are not supported by the installed CLI and would
+    also expose credentials through the process argument vector.
+    """
+
+    return [OPENCLAW_BIN]
+
+
+def _terminate_process_tree(process: Any) -> None:
+    if os.name == "posix":
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        time.sleep(0.05)
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        return
+    if process.poll() is not None:
+        return
+    try:
+        process.terminate()
+        process.wait(timeout=1)
+        return
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+    try:
+        process.kill()
+        process.wait(timeout=1)
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+
+
+def _run_openclaw(
+    args: List[str], timeout: int = 15
+) -> subprocess.CompletedProcess[str]:
+    bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
+    popen_options: Dict[str, Any] = {
+        "stdout": subprocess.PIPE,
+        "stderr": subprocess.PIPE,
+    }
+    if os.name == "posix":
+        popen_options["start_new_session"] = True
+    elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
+        popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+    try:
+        process = subprocess.Popen(args, **popen_options)
+    except (OSError, ValueError):
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    selector = selectors.DefaultSelector()
+    streams = {"stdout": process.stdout, "stderr": process.stderr}
+    buffers = {"stdout": bytearray(), "stderr": bytearray()}
+    try:
+        for name, stream in streams.items():
+            if stream is not None:
+                selector.register(stream, selectors.EVENT_READ, name)
+    except (OSError, ValueError):
+        selector.close()
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None:
+                stream.close()
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    deadline = time.monotonic() + bounded_timeout
+    total_bytes = 0
+    return_code: Optional[int] = None
+    try:
+        while selector.get_map():
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
+                break
+            events = selector.select(min(remaining, 0.1))
+            for key, _mask in events:
+                allowance = MAX_RESPONSE_BYTES - total_bytes
+                try:
+                    chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
+                except OSError:
+                    return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
+                    break
+                if not chunk:
+                    selector.unregister(key.fileobj)
+                    key.fileobj.close()
+                    continue
+                if len(chunk) > allowance:
+                    return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
+                    break
+                buffers[key.data].extend(chunk)
+                total_bytes += len(chunk)
+            if return_code is not None:
+                break
+    finally:
+        selector.close()
+
+    if return_code is not None:
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None and not stream.closed:
+                stream.close()
+        return subprocess.CompletedProcess(args, return_code, "", "")
+    try:
+        completed_return_code = process.wait(timeout=1)
+    except subprocess.TimeoutExpired:
+        _terminate_process_tree(process)
+        return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
+    stdout = buffers["stdout"].decode("utf-8", errors="replace")
+    stderr = buffers["stderr"].decode("utf-8", errors="replace")
+    return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
+
+
+def _json_dict(text: str) -> Dict[str, Any]:
+    try:
+        value = json.loads(text or "{}")
+    except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def discover_openclaw_wechat_channel() -> Optional[str]:
+    """Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
+
+    if WECHAT_CHANNEL:
+        return WECHAT_CHANNEL
+    try:
+        proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
+    except (OSError, subprocess.SubprocessError, ValueError):
+        return None
+    if proc.returncode != 0:
+        return None
+    data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
+    candidates: List[str] = []
+
+    def walk(value: Any) -> None:
+        if isinstance(value, dict):
+            name = value.get("name") or value.get("channel") or value.get("id")
+            kind = value.get("type") or value.get("kind")
+            if isinstance(name, str) and isinstance(kind, str):
+                label = f"{name} {kind}".lower()
+                if "wechat" in label or "weixin" in label:
+                    candidates.append(name)
+            for child in value.values():
+                walk(child)
+        elif isinstance(value, list):
+            for child in value:
+                walk(child)
+
+    walk(data)
+    return candidates[0] if candidates else None
+
+
+def _auth_headers() -> Dict[str, str]:
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    if UCC_ADAPTER_TOKEN:
+        headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
+    return headers
+
+
+def _safe_path(value: Any) -> Optional[str]:
+    if not isinstance(value, str):
+        return None
+    path = value if value.startswith("/") else "/" + value
+    try:
+        encoded_path = path.encode("utf-8")
+    except UnicodeEncodeError:
+        return None
+    if len(encoded_path) > MAX_ENDPOINT_BYTES:
+        return None
+    if not _SAFE_PATH.fullmatch(path):
+        return None
+    decoded = path
+    for _ in range(_MAX_PATH_DECODE_ROUNDS):
+        try:
+            next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
+        except UnicodeDecodeError:
+            return None
+        if next_decoded == decoded:
+            break
+        decoded = next_decoded
+    if "%" in decoded or "\\" in decoded or "\x00" in decoded:
+        return None
+    if not _SAFE_PATH.fullmatch(decoded):
+        return None
+    if any(segment in {".", ".."} for segment in decoded.split("/")):
+        return None
+    return path
+
+
+def _unique_paths(paths: List[str]) -> List[str]:
+    out: List[str] = []
+    for candidate in paths:
+        path = _safe_path(candidate)
+        if path and path not in out:
+            out.append(path)
+        if len(out) == 2:
+            break
+    return out
+
+
+def _ucc_send_paths() -> List[str]:
+    canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
+    return _unique_paths([canonical, _LEGACY_SEND_PATH])
+
+
+def _canonical_origin(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value or len(value) > 2048:
+        return None
+    if "\\" in value or any(ord(character) < 32 for character in value):
+        return None
+    try:
+        parsed = urllib.parse.urlsplit(value)
+        port = parsed.port
+    except (UnicodeError, ValueError):
+        return None
+    scheme = parsed.scheme.lower()
+    if scheme not in {"http", "https"} or not parsed.hostname:
+        return None
+    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
+        return None
+    try:
+        host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
+    except UnicodeError:
+        return None
+    if not host:
+        return None
+    if port is None:
+        port = 80 if scheme == "http" else 443
+    rendered_host = f"[{host}]" if ":" in host else host
+    return f"{scheme}://{rendered_host}:{port}"
+
+
+def _adapter_base_url() -> Optional[str]:
+    origin = _canonical_origin(UCC_ADAPTER_URL)
+    allowed = {
+        canonical
+        for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
+        if canonical
+    }
+    return origin if origin and origin in allowed else None
+
+
+class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    def redirect_request(
+        self,
+        req: Any,
+        fp: Any,
+        code: int,
+        msg: str,
+        headers: Any,
+        newurl: str,
+    ) -> None:
+        del req, fp, code, msg, headers, newurl
+        return None
+
+
+def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
+    opener = urllib.request.build_opener(_NoRedirectHandler())
+    return opener.open(request, timeout=timeout)
+
+
+def ucc_adapter_alive() -> bool:
+    base_url = _adapter_base_url()
+    if not base_url:
+        return False
+    request = urllib.request.Request(
+        base_url + "/health", headers=_auth_headers(), method="GET"
+    )
+    try:
+        with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
+            return 200 <= int(response.status) < 300
+    except (OSError, ValueError, urllib.error.URLError):
+        return False
+
+
+def select_backend() -> Dict[str, Any]:
+    explicit = SEND_BACKEND.strip().lower()
+    if explicit:
+        if explicit == "ucc-adapter":
+            return {"backend": "ucc-adapter", "reason": "explicit"}
+        return {"backend": "unavailable", "reason": "unsupported_backend"}
+    if ucc_adapter_alive():
+        return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
+    return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
+
+
+def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
+    base_url = _adapter_base_url()
+    safe_path = _safe_path(path)
+    if not base_url or not safe_path:
+        raise _BackendProtocolError("invalid adapter endpoint")
+    try:
+        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        raise _BackendProtocolError("invalid request payload") from exc
+    request = urllib.request.Request(
+        base_url + safe_path,
+        data=body,
+        headers=_auth_headers(),
+        method="POST",
+    )
+    with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+        raw = response.read(MAX_RESPONSE_BYTES + 1)
+        if len(raw) > MAX_RESPONSE_BYTES:
+            raise _BackendProtocolError("adapter response too large")
+        try:
+            text = raw.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise _BackendProtocolError("adapter response encoding invalid") from exc
+        data = _json_dict(text)
+        if not data:
+            raise _BackendProtocolError("adapter response invalid")
+        return int(response.status), data
+
+
+def _ucc_response_sent(data: Dict[str, Any]) -> bool:
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
+
+
+def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
+    expected_status = "dry_run" if dry_run else "sent"
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
+
+
+def _utf8_size(value: str) -> Optional[int]:
+    try:
+        return len(value.encode("utf-8"))
+    except UnicodeEncodeError:
+        return None
+
+
+def _valid_identifier(value: Any, required: bool = False) -> bool:
+    if not isinstance(value, str):
+        return False
+    if required and not value.strip():
+        return False
+    if not value:
+        return not required
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return False
+    size = _utf8_size(value)
+    return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
+        ord(character) < 32 or ord(character) == 127 for character in value
+    )
+
+
+def _validate_metadata(metadata: Any) -> Optional[str]:
+    if not isinstance(metadata, dict):
+        return "invalid_input"
+    nodes = 0
+    stack: List[Tuple[Any, int]] = [(metadata, 0)]
+    while stack:
+        value, depth = stack.pop()
+        nodes += 1
+        if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
+            return "input_too_large"
+        if isinstance(value, dict):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            for key, child in value.items():
+                if (
+                    not isinstance(key, str)
+                    or len(key) > MAX_METADATA_BYTES
+                    or _utf8_size(key) is None
+                ):
+                    return "invalid_input"
+                stack.append((child, depth + 1))
+        elif isinstance(value, list):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            stack.extend((child, depth + 1) for child in value)
+        elif isinstance(value, str):
+            if len(value) > MAX_METADATA_BYTES:
+                return "input_too_large"
+            if _utf8_size(value) is None:
+                return "invalid_input"
+        elif value is not None and not isinstance(value, (bool, int, float)):
+            return "invalid_input"
+    try:
+        encoded = json.dumps(
+            metadata,
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
+        return "invalid_input"
+    return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
+
+
+def _validate_payload(
+    target: Any,
+    content: Any,
+    account_id: Any,
+    message_id: Any,
+    metadata: Any,
+    dry_run: Any,
+) -> Optional[str]:
+    if not isinstance(target, str) or not isinstance(content, str):
+        return "invalid_input"
+    if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
+        return "invalid_input"
+    if any(ord(character) < 32 or ord(character) == 127 for character in target):
+        return "invalid_input"
+    if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
+        return "invalid_input"
+    if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    target_size = _utf8_size(target)
+    content_size = _utf8_size(content)
+    if target_size is None or content_size is None:
+        return "invalid_input"
+    if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    return _validate_metadata(metadata)
+
+
+def _safe_provider_message_id(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value:
+        return None
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return None
+    size = _utf8_size(value)
+    if size is None or size > MAX_IDENTIFIER_BYTES:
+        return None
+    if any(ord(character) < 32 or ord(character) == 127 for character in value):
+        return None
+    return value
+
+
+def send_ucc_adapter(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    effective_account = account_id or WECHAT_ACCOUNT_ID
+    effective_metadata: Any = {} if metadata is None else metadata
+    error_code = _validate_payload(
+        target,
+        content,
+        effective_account,
+        message_id,
+        effective_metadata,
+        dry_run,
+    )
+    if error_code:
+        return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
+
+    payload = {
+        "target": target,
+        "content": content,
+        "account_id": effective_account,
+        "message_id": message_id,
+        "metadata": effective_metadata,
+        "dry_run": dry_run,
+    }
+    last_endpoint: Optional[str] = None
+    for path in _ucc_send_paths():
+        last_endpoint = path
+        try:
+            http_status, data = _post_ucc(path, payload)
+        except urllib.error.HTTPError as exc:
+            if exc.code in {404, 405}:
+                continue
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
+            )
+        except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+        except Exception:
+            # Fail closed without returning exception text, type, or chained cause.
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+
+        if not (200 <= http_status < 300):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
+            )
+        if not _ucc_result_is_ok(data, dry_run):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
+            )
+        confirmed_status = "dry_run" if dry_run else "sent"
+        result_data: Dict[str, Any] = {
+            "code": confirmed_status,
+            "http_status": http_status,
+            "endpoint": path,
+        }
+        provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
+        if provider_message_id:
+            result_data["provider_message_id"] = provider_message_id
+        return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
+
+    data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
+    if last_endpoint:
+        data["endpoint"] = last_endpoint
+    return BackendResult(False, "ucc-adapter", "failed", data)
+
+
+def send_openclaw_channel(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    del target, content, account_id, message_id, metadata, dry_run
+    return BackendResult(
+        False,
+        "openclaw-channel",
+        "failed",
+        {"code": "unsupported_backend"},
+    )
+
+
+def send_message(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> Dict[str, Any]:
+    selected = select_backend()
+    if selected.get("backend") == "ucc-adapter":
+        return send_ucc_adapter(
+            target, content, account_id, message_id, metadata, dry_run
+        ).to_dict()
+    code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
+    return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
+
+
+if __name__ == "__main__":
+    print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))

tests/test_wechat_send_backend.py

@@ -1,11 +1,20 @@
 from __future__ import annotations
 
 import importlib.util
+import io
+import os
+import signal
+import subprocess
 import sys
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import SimpleNamespace
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import Mock, call, patch
 
 
 BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -80,6 +89,450 @@ class WechatSendBackendTests(TestCase):
         self.assertEqual(args[1]["content"], "hello")
         self.assertFalse(args[1]["dry_run"])
 
+    def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
+        secret = "ucc-super-secret-token"
+
+        class Response:
+            status = 200
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+            def read(self, _limit=-1):
+                return b'{"ok":true,"status":"sent"}'
+
+        urlopen = Mock(return_value=Response())
+        output = io.StringIO()
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
+            patch.object(self.backend, "_open_no_redirect", urlopen),
+            redirect_stdout(output),
+            redirect_stderr(output),
+        ):
+            status, payload = self.backend._post_ucc(
+                "/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
+            )
+
+        request = urlopen.call_args.args[0]
+        self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
+        self.assertEqual(status, 200)
+        self.assertEqual(payload, {"ok": True, "status": "sent"})
+        self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
+        self.assertNotIn(secret, output.getvalue())
+        self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))
+
+    def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
+        with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
+            paths = self.backend._ucc_send_paths()
+
+        self.assertEqual(paths[0], "/api/internal/wechat/send")
+        self.assertLessEqual(len(paths), 2)
+        self.assertEqual(len(paths), len(set(paths)))
+        self.assertTrue(all(path.startswith("/") for path in paths))
+
+    def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
+        self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
+        for response in (
+            {"ok": 1, "status": "sent"},
+            {"ok": True, "status": "accepted"},
+            {"ok": False, "status": "sent"},
+            {"status": "sent"},
+        ):
+            with self.subTest(response=response):
+                self.assertFalse(self.backend._ucc_response_sent(response))
+
+    def test_ambiguous_canonical_failures_never_fallback(self):
+        secret = "raw-upstream-secret"
+        failures = (
+            TimeoutError(secret),
+            ConnectionError(secret),
+            self.backend._BackendProtocolError(secret),
+            RuntimeError(secret),
+        )
+        for failure in failures:
+            with self.subTest(failure=type(failure).__name__):
+                with (
+                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
+                    patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", message_id="msg-001"
+                    ).to_dict()
+
+                self.assertFalse(result["ok"])
+                self.assertEqual(result["code"], "ucc_adapter_unavailable")
+                self.assertEqual(post.call_count, 1)
+                self.assertNotIn(secret, repr(result))
+
+    def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
+        for http_status in (404, 405):
+            with self.subTest(http_status=http_status):
+                missing = urllib.error.HTTPError(
+                    "http://127.0.0.1/api/internal/wechat/send",
+                    http_status,
+                    "not found",
+                    {},
+                    None,
+                )
+                with (
+                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
+                    patch.object(
+                        self.backend,
+                        "_post_ucc",
+                        side_effect=[missing, (200, {"ok": True, "status": "sent"})],
+                    ) as post,
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", message_id="msg-001"
+                    ).to_dict()
+
+                self.assertTrue(result["ok"])
+                self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
+                self.assertEqual(post.call_count, 2)
+
+    def test_non_not_found_http_error_never_falls_back(self):
+        denied = urllib.error.HTTPError(
+            "http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
+        )
+        with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
+            result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_http_error")
+        post.assert_called_once()
+
+    def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
+        secret = "provider-password=secret-value"
+        with patch.object(
+            self.backend,
+            "_post_ucc",
+            return_value=(200, {"ok": False, "status": "failed", "error": secret}),
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", message_id="msg-001"
+            ).to_dict()
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_rejected")
+        self.assertNotIn(secret, repr(result))
+
+    def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
+        with patch.object(
+            self.backend,
+            "_post_ucc",
+            return_value=(200, {"ok": True, "status": "dry_run"}),
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", dry_run=True
+            ).to_dict()
+
+        self.assertTrue(result["ok"])
+        self.assertEqual(result["status"], "dry_run")
+        self.assertNotEqual(result["status"], "sent")
+
+        for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
+            with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
+                with patch.object(
+                    self.backend,
+                    "_post_ucc",
+                    return_value=(200, {"ok": True, "status": upstream_status}),
+                ):
+                    rejected = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", dry_run=dry_run
+                    ).to_dict()
+                self.assertFalse(rejected["ok"])
+                self.assertEqual(rejected["status"], "failed")
+
+    def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
+        invalid_values = (
+            "\ud800",
+            "contains\x00control",
+            "x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
+            {"not": "a string"},
+        )
+        for provider_message_id in invalid_values:
+            with self.subTest(value=repr(provider_message_id)[:40]):
+                with patch.object(
+                    self.backend,
+                    "_post_ucc",
+                    return_value=(
+                        200,
+                        {
+                            "ok": True,
+                            "status": "sent",
+                            "provider_message_id": provider_message_id,
+                        },
+                    ),
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello"
+                    ).to_dict()
+
+                self.assertTrue(result["ok"])
+                self.assertEqual(result["status"], "sent")
+                self.assertNotIn("provider_message_id", result)
+
+    def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
+        completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
+        with (
+            patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
+            patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
+        ):
+            self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
+
+        command = run.call_args.args[0]
+        self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
+        self.assertFalse({"--gateway", "--token", "--password"} & set(command))
+
+    def test_openclaw_output_overflow_terminates_while_process_is_running(self):
+        writer = (
+            "import os,time;"
+            f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
+            "time.sleep(10)"
+        )
+        started = time.monotonic()
+        completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 2.0)
+        self.assertNotEqual(completed.returncode, 0)
+        self.assertEqual(completed.stdout, "")
+        self.assertEqual(completed.stderr, "")
+
+    def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
+        started = time.monotonic()
+        completed = self.backend._run_openclaw(
+            [sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
+        )
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 3.0)
+        self.assertNotEqual(completed.returncode, 0)
+        self.assertEqual(completed.stdout, "")
+        self.assertEqual(completed.stderr, "")
+
+    def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
+        if os.name != "posix" or not Path("/proc").is_dir():
+            self.skipTest("POSIX process-group and procfs contract")
+
+        def process_is_running(pid):
+            try:
+                state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
+            except (FileNotFoundError, IndexError, OSError):
+                return False
+            return state not in {"X", "Z"}
+
+        with tempfile.TemporaryDirectory() as directory:
+            pid_path = Path(directory) / "descendant.pid"
+            descendant = "import time;time.sleep(30)"
+            launcher = (
+                "import pathlib,subprocess,sys;"
+                f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
+                f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
+            )
+            started = time.monotonic()
+            completed = self.backend._run_openclaw(
+                [sys.executable, "-c", launcher], timeout=1
+            )
+            elapsed = time.monotonic() - started
+            self.assertTrue(pid_path.is_file())
+            descendant_pid = int(pid_path.read_text(encoding="utf-8"))
+            try:
+                deadline = time.monotonic() + 2
+                while process_is_running(descendant_pid) and time.monotonic() < deadline:
+                    time.sleep(0.02)
+                self.assertFalse(process_is_running(descendant_pid))
+            finally:
+                if process_is_running(descendant_pid):
+                    os.kill(descendant_pid, signal.SIGKILL)
+
+        self.assertLess(elapsed, 3.0)
+        self.assertNotEqual(completed.returncode, 0)
+
+    def test_process_tree_termination_uses_posix_process_group(self):
+        if os.name != "posix":
+            self.skipTest("POSIX process-group contract")
+
+        class Process:
+            pid = 43210
+
+            def __init__(self):
+                self.wait_count = 0
+
+            def poll(self):
+                return None
+
+            def wait(self, timeout):
+                self.wait_count += 1
+                if self.wait_count == 1:
+                    raise subprocess.TimeoutExpired(["openclaw"], timeout)
+                return -signal.SIGKILL
+
+        process = Process()
+        with patch.object(self.backend.os, "killpg") as killpg:
+            self.backend._terminate_process_tree(process)
+
+        self.assertEqual(
+            killpg.call_args_list,
+            [
+                call(process.pid, signal.SIGTERM),
+                call(process.pid, signal.SIGKILL),
+            ],
+        )
+
+    def test_unsupported_and_generic_chat_backends_fail_closed(self):
+        with (
+            patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
+            patch.object(self.backend, "ucc_adapter_alive", return_value=False),
+        ):
+            selected = self.backend.select_backend()
+            result = self.backend.send_message(target="chat-001", content="hello")
+
+        self.assertEqual(selected["backend"], "unavailable")
+        self.assertEqual(selected["reason"], "unsupported_backend")
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "unsupported_backend")
+
+    def test_invalid_or_oversized_input_is_rejected_before_network(self):
+        with patch.object(self.backend, "_post_ucc") as post:
+            empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
+            oversized = self.backend.send_ucc_adapter(
+                target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
+            ).to_dict()
+
+        self.assertEqual(empty["code"], "invalid_input")
+        self.assertEqual(oversized["code"], "input_too_large")
+        post.assert_not_called()
+
+    def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
+        cases = (
+            {"target": "\ud800", "content": "hello"},
+            {"target": "chat-001", "content": "\ud800"},
+            {"target": "chat-001", "content": "hello", "account_id": "\ud800"},
+            {"target": "chat-001", "content": "hello", "message_id": "\ud800"},
+            {"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
+        )
+        with patch.object(self.backend, "_post_ucc") as post:
+            for kwargs in cases:
+                with self.subTest(field=next(reversed(kwargs))):
+                    result = self.backend.send_ucc_adapter(**kwargs).to_dict()
+                    self.assertEqual(result["code"], "invalid_input")
+        post.assert_not_called()
+
+    def test_deeply_nested_metadata_is_bounded_before_network(self):
+        metadata = {}
+        for _ in range(1200):
+            metadata = {"nested": metadata}
+
+        with patch.object(self.backend, "_post_ucc") as post:
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", metadata=metadata
+            ).to_dict()
+
+        self.assertEqual(result["code"], "input_too_large")
+        post.assert_not_called()
+
+    def test_adapter_origin_must_be_allowlisted(self):
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
+            patch.object(
+                self.backend,
+                "UCC_ADAPTER_ALLOWED_ORIGINS",
+                ("http://127.0.0.1:18800",),
+            ),
+        ):
+            self.assertIsNone(self.backend._adapter_base_url())
+
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
+            patch.object(
+                self.backend,
+                "UCC_ADAPTER_ALLOWED_ORIGINS",
+                ("https://adapter.example:8443",),
+            ),
+        ):
+            self.assertEqual(
+                self.backend._adapter_base_url(), "https://adapter.example:8443"
+            )
+
+    def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
+        invalid_origins = (
+            "ftp://127.0.0.1:18800",
+            "http://user:password@127.0.0.1:18800",
+            "http://127.0.0.1:18800#fragment",
+            "http://127.0.0.1:18800/base-path",
+        )
+        for origin in invalid_origins:
+            with self.subTest(origin=origin):
+                with (
+                    patch.object(self.backend, "UCC_ADAPTER_URL", origin),
+                    patch.object(
+                        self.backend,
+                        "UCC_ADAPTER_ALLOWED_ORIGINS",
+                        ("http://127.0.0.1:18800",),
+                    ),
+                ):
+                    self.assertIsNone(self.backend._adapter_base_url())
+
+    def test_adapter_path_rejects_decoded_traversal(self):
+        for path in (
+            "/api/%2e%2e/send",
+            "/api/%2E%2E/send",
+            "/api/%2525252e%2525252e/send",
+            "/api/%252525252e%252525252e/send",
+            "/api/../send",
+            "/api/%zz/send",
+        ):
+            with self.subTest(path=path):
+                self.assertIsNone(self.backend._safe_path(path))
+
+    def test_adapter_path_keeps_intended_safe_canonical_paths(self):
+        for path in (
+            "/api/internal/wechat/send",
+            "/internal/channels/wechat/send",
+            "/api/%7Ehealth/status",
+        ):
+            with self.subTest(path=path):
+                self.assertEqual(self.backend._safe_path(path), path)
+
+    def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
+        handler = self.backend._NoRedirectHandler()
+        self.assertIsNone(
+            handler.redirect_request(
+                None,
+                None,
+                302,
+                "redirect",
+                {"Location": "https://evil.example/steal"},
+                "https://evil.example/steal",
+            )
+        )
+
+        redirected = urllib.error.HTTPError(
+            "http://127.0.0.1:18800/api/internal/wechat/send",
+            302,
+            "redirect",
+            {"Location": "https://evil.example/steal"},
+            None,
+        )
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
+            patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello"
+            ).to_dict()
+
+        request = opened.call_args.args[0]
+        self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
+        self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
+        opened.assert_called_once()
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_http_error")
+
 
 if __name__ == "__main__":
     import unittest