feat(v2r1): package canonical authenticated ingress

active-runtime/app/aios_v2r1/__init__.py

@@ -0,0 +1,3 @@
+"""Canonical AIOS V2R1 runtime package."""
+
+__version__ = "2.1.0-rc.1"

active-runtime/app/aios_v2r1/cli.py

@@ -6,6 +6,7 @@ import json
 import sys
 
 from .cell_manager import CellManager
+from .ingress_service import serve
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -29,8 +30,25 @@ def build_parser() -> argparse.ArgumentParser:
     return parser
 
 
+def build_serve_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="aios-cell serve")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=18792)
+    return parser
+
+
 def main(argv: list[str] | None = None) -> int:
-    args = build_parser().parse_args(argv)
+    values = list(sys.argv[1:] if argv is None else argv)
+    service_mode = (
+        bool(values)
+        and values[0] == "serve"
+        and (len(values) == 1 or values[1].startswith("-"))
+    )
+    if service_mode:
+        serve_args = build_serve_parser().parse_args(values[1:])
+        serve(serve_args.host, serve_args.port)
+        return 0
+    args = build_parser().parse_args(values)
     manager = CellManager(args.registry)
     if args.command == "create":
         result = manager.create(args.tenant_id, args.cell_id, args.version, args.digest)

active-runtime/app/aios_v2r1/ingress_service.py

@@ -0,0 +1,461 @@
+"""Fail-closed loopback HTTP shell for V2R1 ingress processors."""
+from __future__ import annotations
+
+import ipaddress
+import json
+import socket
+import threading
+import time
+from collections.abc import Callable, Mapping
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+from urllib.parse import urlsplit
+
+
+MAX_REQUEST_BYTES = 1024 * 1024
+MAX_RESPONSE_BYTES = 4096
+REQUEST_IO_TIMEOUT_SECONDS = 0.5
+REQUEST_NETWORK_DEADLINE_SECONDS = 1.0
+HEX_DIGITS = frozenset("0123456789abcdefABCDEF")
+HEADER_NAME_BYTES = frozenset(
+    b"!#$%&'*+-.^_`|~0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+)
+INTERNAL_ERROR_BODY = b'{"status":"INTERNAL_ERROR"}'
+MUTATION_ROUTES = frozenset(
+    {"/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"}
+)
+Processor = Callable[[str, bytes], tuple[int, Mapping[str, Any]]]
+
+
+class RequestRejected(Exception):
+    def __init__(self, status: int, payload_status: str) -> None:
+        super().__init__(payload_status)
+        self.status = status
+        self.payload_status = payload_status
+
+
+class IngressHTTPServer(ThreadingHTTPServer):
+    """Threaded loopback server with an optional externally supplied processor."""
+
+    daemon_threads = False
+    block_on_close = True
+
+    def __init__(
+        self,
+        address: tuple[str, int],
+        handler: type[BaseHTTPRequestHandler],
+        processor: Processor | None,
+    ) -> None:
+        self._closing = threading.Event()
+        self._lifecycle_lock = threading.RLock()
+        self._connections_lock = threading.Lock()
+        self._connections: set[socket.socket] = set()
+        self.processor = processor
+        super().__init__(address, handler)
+
+    @property
+    def is_closing(self) -> bool:
+        return self._closing.is_set()
+
+    def register_connection(self, connection: socket.socket) -> None:
+        with self._connections_lock:
+            self._connections.add(connection)
+
+    def unregister_connection(self, connection: socket.socket) -> None:
+        with self._connections_lock:
+            self._connections.discard(connection)
+
+    def _interrupt_active_reads(self) -> None:
+        with self._connections_lock:
+            connections = tuple(self._connections)
+        for connection in connections:
+            try:
+                connection.shutdown(socket.SHUT_RD)
+            except OSError:
+                pass
+
+    def call_processor(self, path: str, body: bytes) -> tuple[bool, Any]:
+        """Start processing atomically with respect to server shutdown."""
+        with self._lifecycle_lock:
+            if self._closing.is_set() or self.processor is None:
+                return False, None
+            return True, self.processor(path, body)
+
+    def shutdown(self) -> None:
+        with self._lifecycle_lock:
+            self._closing.set()
+        self._interrupt_active_reads()
+        super().shutdown()
+
+    def server_close(self) -> None:
+        with self._lifecycle_lock:
+            self._closing.set()
+        self._interrupt_active_reads()
+        super().server_close()
+
+    def handle_error(
+        self,
+        request: socket.socket,
+        client_address: tuple[str, int],
+    ) -> None:
+        """Close unexpected handler failures without printing request data."""
+        try:
+            request.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+
+
+class _ValidatingHeaderReader:
+    """Validate raw header lines before the standard library normalizes them."""
+
+    def __init__(self, handler: "IngressRequestHandler", reader: Any) -> None:
+        self._handler = handler
+        self._reader = reader
+
+    def readline(self, limit: int = -1) -> bytes:
+        try:
+            line = self._reader.readline(limit)
+        except TimeoutError as exc:
+            raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
+        self._handler._ensure_network_read_active()
+        self._validate(line)
+        return line
+
+    @staticmethod
+    def _validate(line: bytes) -> None:
+        if line == b"\r\n":
+            return
+        if not line or not line.endswith(b"\r\n"):
+            raise RequestRejected(400, "BAD_REQUEST")
+        raw = line[:-2]
+        if not raw or raw[:1] in {b" ", b"\t"}:
+            raise RequestRejected(400, "BAD_REQUEST")
+        separator = raw.find(b":")
+        if separator <= 0:
+            raise RequestRejected(400, "BAD_REQUEST")
+        name = raw[:separator]
+        value = raw[separator + 1 :]
+        if any(byte not in HEADER_NAME_BYTES for byte in name):
+            raise RequestRejected(400, "BAD_REQUEST")
+        if any((byte < 0x20 and byte != 0x09) or byte == 0x7F for byte in value):
+            raise RequestRejected(400, "BAD_REQUEST")
+        if name.lower() == b"content-length":
+            numeric = value[1:] if value.startswith(b" ") else value
+            if not numeric or any(byte < 0x30 or byte > 0x39 for byte in numeric):
+                raise RequestRejected(400, "BAD_REQUEST")
+
+
+class IngressRequestHandler(BaseHTTPRequestHandler):
+    """Bounded JSON transport; application processing is injected later."""
+
+    server: IngressHTTPServer
+
+    def setup(self) -> None:
+        super().setup()
+        self.connection.settimeout(REQUEST_IO_TIMEOUT_SECONDS)
+        self._network_read_lock = threading.Lock()
+        self._network_read_expired = threading.Event()
+        self._network_read_deadline: float | None = None
+        self._network_read_timer: threading.Timer | None = None
+        self._network_read_active = False
+        self.server.register_connection(self.connection)
+
+    def finish(self) -> None:
+        self._complete_network_read()
+        try:
+            super().finish()
+        except OSError:
+            pass
+        finally:
+            self.server.unregister_connection(self.connection)
+
+    def _start_network_read(self) -> None:
+        self._complete_network_read()
+        timer = threading.Timer(
+            REQUEST_NETWORK_DEADLINE_SECONDS,
+            self._expire_network_read,
+        )
+        timer.name = f"aios-ingress-read-deadline-{id(self)}"
+        timer.daemon = True
+        with self._network_read_lock:
+            self._network_read_expired.clear()
+            self._network_read_deadline = (
+                time.monotonic() + REQUEST_NETWORK_DEADLINE_SECONDS
+            )
+            self._network_read_active = True
+            self._network_read_timer = timer
+        timer.start()
+
+    def _expire_network_read(self) -> None:
+        with self._network_read_lock:
+            if not self._network_read_active:
+                return
+            self._network_read_expired.set()
+        self.close_connection = True
+        try:
+            self.connection.shutdown(socket.SHUT_RD)
+        except OSError:
+            pass
+
+    def _complete_network_read(self) -> None:
+        timer: threading.Timer | None
+        with getattr(self, "_network_read_lock", threading.Lock()):
+            self._network_read_active = False
+            self._network_read_deadline = None
+            timer = self._network_read_timer
+            self._network_read_timer = None
+        if timer is not None:
+            timer.cancel()
+            if timer is not threading.current_thread():
+                timer.join(timeout=REQUEST_IO_TIMEOUT_SECONDS)
+
+    def _ensure_network_read_active(self) -> None:
+        if self.server.is_closing:
+            raise RequestRejected(503, "SERVICE_UNAVAILABLE")
+        deadline = self._network_read_deadline
+        if self._network_read_expired.is_set() or (
+            deadline is not None and time.monotonic() >= deadline
+        ):
+            self._network_read_expired.set()
+            raise RequestRejected(408, "REQUEST_TIMEOUT")
+
+    def parse_request(self) -> bool:
+        original_reader = self.rfile
+        self.rfile = _ValidatingHeaderReader(self, original_reader)  # type: ignore[assignment]
+        try:
+            return super().parse_request()
+        finally:
+            self.rfile = original_reader
+
+    def handle_one_request(self) -> None:
+        self._start_network_read()
+        try:
+            super().handle_one_request()
+        except RequestRejected as rejection:
+            self.close_connection = True
+            self._send_safely(rejection.status, {"status": rejection.payload_status})
+        except TimeoutError:
+            self.close_connection = True
+            self._send_safely(408, {"status": "REQUEST_TIMEOUT"})
+        except (BrokenPipeError, ConnectionResetError, OSError):
+            self.close_connection = True
+        except Exception:
+            self.close_connection = True
+            self._send_safely(500, {"status": "INTERNAL_ERROR"})
+        finally:
+            self._complete_network_read()
+
+    def send_error(
+        self,
+        code: int,
+        message: str | None = None,
+        explain: str | None = None,
+    ) -> None:
+        """Replace BaseHTTPRequestHandler HTML errors with bounded JSON."""
+        self.close_connection = True
+        if code == 501:
+            self._method_not_allowed()
+            return
+        self._send(int(code), {"status": "HTTP_ERROR"})
+
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+    def _send_safely(self, status: int, payload: Mapping[str, Any]) -> None:
+        try:
+            self._send(status, payload)
+        except OSError:
+            self.close_connection = True
+
+    def _path(self) -> str:
+        target = self.path
+        if not target or any(
+            ord(character) < 0x20 or ord(character) == 0x7F for character in target
+        ):
+            raise RequestRejected(400, "BAD_REQUEST")
+        for index, character in enumerate(target):
+            if character == "%" and (
+                index + 2 >= len(target)
+                or any(value not in HEX_DIGITS for value in target[index + 1 : index + 3])
+            ):
+                raise RequestRejected(400, "BAD_REQUEST")
+        try:
+            parts = urlsplit(target)
+        except ValueError as exc:
+            raise RequestRejected(400, "BAD_REQUEST") from exc
+        if parts.fragment:
+            raise RequestRejected(400, "BAD_REQUEST")
+        if target.startswith("/"):
+            if parts.scheme or parts.netloc:
+                raise RequestRejected(400, "BAD_REQUEST")
+        elif parts.scheme in {"http", "https"} and parts.netloc:
+            if parts.hostname is None:
+                raise RequestRejected(400, "BAD_REQUEST")
+            try:
+                _ = parts.port
+            except ValueError as exc:
+                raise RequestRejected(400, "BAD_REQUEST") from exc
+        else:
+            raise RequestRejected(400, "BAD_REQUEST")
+        return parts.path or "/"
+
+    @staticmethod
+    def _encoded_response(
+        status: object,
+        payload: object,
+    ) -> tuple[int, bytes]:
+        if (
+            type(status) is not int
+            or not 200 <= status <= 599
+            or status in {204, 205, 304}
+            or not isinstance(payload, Mapping)
+        ):
+            return 500, INTERNAL_ERROR_BODY
+        try:
+            body = json.dumps(
+                dict(payload),
+                ensure_ascii=False,
+                separators=(",", ":"),
+                sort_keys=True,
+                allow_nan=False,
+            ).encode("utf-8")
+        except Exception:
+            return 500, INTERNAL_ERROR_BODY
+        if len(body) > MAX_RESPONSE_BYTES:
+            return 500, INTERNAL_ERROR_BODY
+        return status, body
+
+    def _send(self, status: int, payload: Mapping[str, Any]) -> None:
+        self._complete_network_read()
+        status, body = self._encoded_response(status, payload)
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        if getattr(self, "command", None) != "HEAD":
+            self.wfile.write(body)
+
+    def _method_not_allowed(self) -> None:
+        self._send(405, {"status": "METHOD_NOT_ALLOWED"})
+
+    def _reject(self, status: int, payload_status: str) -> None:
+        self.close_connection = True
+        self._send(status, {"status": payload_status})
+
+    def _content_length(self) -> int:
+        if self.headers.get_all("Transfer-Encoding", []):
+            raise RequestRejected(400, "BAD_REQUEST")
+        values = self.headers.get_all("Content-Length", [])
+        if len(values) > 1:
+            raise RequestRejected(400, "BAD_REQUEST")
+        if not values:
+            return 0
+        raw_length = values[0]
+        if (
+            not raw_length
+            or not raw_length.isascii()
+            or not raw_length.isdigit()
+        ):
+            raise RequestRejected(400, "BAD_REQUEST")
+        try:
+            length = int(raw_length)
+        except ValueError as exc:
+            raise RequestRejected(400, "BAD_REQUEST") from exc
+        if length > MAX_REQUEST_BYTES:
+            raise RequestRejected(413, "REQUEST_TOO_LARGE")
+        return length
+
+    def _read_exact_body(self, length: int) -> bytes:
+        try:
+            body = self.rfile.read(length)
+        except TimeoutError as exc:
+            raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
+        self._ensure_network_read_active()
+        if len(body) != length:
+            raise RequestRejected(400, "BAD_REQUEST")
+        return body
+
+    def do_GET(self) -> None:
+        path = self._path()
+        if path == "/health/live":
+            self._send(200, {"status": "LIVE"})
+        elif path == "/health/ready":
+            if self.server.processor is None:
+                self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
+            else:
+                self._send(200, {"status": "READY"})
+        elif path in MUTATION_ROUTES:
+            self._method_not_allowed()
+        else:
+            self._send(404, {"status": "NOT_FOUND"})
+
+    def do_POST(self) -> None:
+        path = self._path()
+        if path not in MUTATION_ROUTES:
+            if path in {"/health/live", "/health/ready"}:
+                self._method_not_allowed()
+            else:
+                self._send(404, {"status": "NOT_FOUND"})
+            return
+        try:
+            length = self._content_length()
+        except RequestRejected as rejection:
+            self._reject(rejection.status, rejection.payload_status)
+            return
+        processor = self.server.processor
+        if processor is None:
+            self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
+            return
+        if self.server.is_closing:
+            self._reject(503, "SERVICE_UNAVAILABLE")
+            return
+        try:
+            body = self._read_exact_body(length)
+        except RequestRejected as rejection:
+            self._reject(rejection.status, rejection.payload_status)
+            return
+        self._complete_network_read()
+        try:
+            started, result = self.server.call_processor(path, body)
+            if not started:
+                self._reject(503, "SERVICE_UNAVAILABLE")
+                return
+            status, payload = result
+        except Exception:
+            self._send(500, {"status": "INTERNAL_ERROR"})
+            return
+        self._send(status, payload)
+
+    do_DELETE = _method_not_allowed
+    do_HEAD = _method_not_allowed
+    do_OPTIONS = _method_not_allowed
+    do_PATCH = _method_not_allowed
+    do_PUT = _method_not_allowed
+
+
+def build_server(
+    host: str = "127.0.0.1",
+    port: int = 18792,
+    *,
+    processor: Processor | None = None,
+) -> IngressHTTPServer:
+    """Build a loopback-only HTTP server without starting it."""
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError as exc:
+        raise ValueError("host must be a loopback IP address") from exc
+    if address.version != 4 or not address.is_loopback:
+        raise ValueError("host must be an IPv4 loopback address")
+    return IngressHTTPServer((host, port), IngressRequestHandler, processor)
+
+
+def serve(host: str = "127.0.0.1", port: int = 18792) -> None:
+    """Run the unconfigured loopback shell until interrupted."""
+    server = build_server(host, port)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        return
+    finally:
+        server.server_close()

active-runtime/pyproject.toml

@@ -0,0 +1,17 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "aios-v2r1-runtime"
+version = "2.1.0rc1"
+description = "AIOS V2R1 canonical runtime package"
+requires-python = ">=3.11"
+dependencies = []
+
+[project.scripts]
+aios-cell = "aios_v2r1.cli:main"
+
+[tool.setuptools]
+package-dir = {"" = "app"}
+packages = ["aios_v2r1"]

openclaw/plugins/aios-tenant-boundary/dist/index.js

@@ -1,31 +1,48 @@
 import {
   classifyRun,
   decideToolCall,
+  resolveSenderId,
   runKey,
 } from "./policy.js";
 
-const activeRuns = new Map();
 const MAX_RUN_AGE_MS = 60 * 60 * 1000;
 
-function prune(now = Date.now()) {
+function prune(activeRuns, now = Date.now()) {
   for (const [key, value] of activeRuns.entries()) {
     if (now - value.recordedAt > MAX_RUN_AGE_MS) activeRuns.delete(key);
   }
 }
 
+function approvalTimeout(config) {
+  const value = config.approvalTimeoutMs;
+  return Number.isInteger(value) && value >= 1000 && value <= 300000
+    ? value
+    : 60000;
+}
+
 export default {
   id: "aios-tenant-boundary",
   name: "AIOS Tenant Boundary",
   description: "Fail-closed user-origin and side-effect policy for Tenant Cells.",
 
   register(api) {
+    const activeRuns = new Map();
+
     api.on(
       "before_agent_run",
       async (event, ctx) => {
-        prune();
+        prune(activeRuns);
         const config = event.context?.pluginConfig ?? api.pluginConfig ?? {};
         const trust = classifyRun(ctx, config);
         const key = runKey(event, ctx);
+        const existing = key ? activeRuns.get(key) : undefined;
+        if (existing && existing.senderId !== trust.senderId) {
+          return {
+            outcome: "block",
+            reason: "run_scope_sender_conflict",
+            message: "This run scope is already bound to a different authenticated sender.",
+          };
+        }
         if (key) {
           activeRuns.set(key, { ...trust, recordedAt: Date.now() });
         }
@@ -49,10 +66,14 @@ export default {
     api.on(
       "before_tool_call",
       async (event, ctx) => {
-        prune();
+        prune(activeRuns);
         const config = event.context?.pluginConfig ?? api.pluginConfig ?? {};
         const key = runKey(event, ctx);
-        const trust = key ? activeRuns.get(key) : undefined;
+        let trust = key ? activeRuns.get(key) : undefined;
+        const senderId = resolveSenderId(ctx);
+        if (trust && senderId && trust.senderId !== senderId) {
+          trust = { trusted: false, reason: "run_scope_sender_conflict" };
+        }
         const decision = decideToolCall(event.toolName, trust, config);
 
         if (decision.action === "block") {
@@ -67,7 +88,7 @@ export default {
               title: `Approve ${String(event.toolName)}`,
               description: "Authenticated user requested a side-effecting tool call in this Tenant Cell.",
               severity: "warning",
-              timeoutMs: Number(config.approvalTimeoutMs ?? 60000),
+              timeoutMs: approvalTimeout(config),
               allowedDecisions: ["allow-once", "deny"],
               pluginId: "aios-tenant-boundary",
             },
@@ -78,9 +99,13 @@ export default {
       { priority: 100 },
     );
 
-    api.on("agent_end", async (event, ctx) => {
-      const key = runKey(event, ctx);
-      if (key) activeRuns.delete(key);
-    });
+    api.on(
+      "agent_end",
+      async (event, ctx) => {
+        const key = runKey(event, ctx);
+        if (key) activeRuns.delete(key);
+      },
+      { priority: 100 },
+    );
   },
 };

openclaw/plugins/aios-tenant-boundary/dist/policy.js

@@ -66,6 +66,26 @@ export function decideToolCall(toolName, runTrust, config = {}) {
 }
 
 export function runKey(event = {}, ctx = {}) {
-  const key = event.runId || ctx.runId;
-  return typeof key === "string" && key.trim() ? key.trim() : undefined;
+  const normalize = (value) => typeof value === "string" && value.trim()
+    ? value.trim()
+    : undefined;
+  const sameField = (eventValue, contextValue) => {
+    const fromEvent = normalize(eventValue);
+    const fromContext = normalize(contextValue);
+    return {
+      valid: !(fromEvent && fromContext && fromEvent !== fromContext),
+      value: fromEvent || fromContext,
+    };
+  };
+  const runId = sameField(event.runId, ctx.runId);
+  const agentId = sameField(event.agentId, ctx.agentId);
+  const sessionId = sameField(event.sessionId, ctx.sessionId);
+  const sessionKey = sameField(event.sessionKey, ctx.sessionKey);
+  if (!runId.valid || !agentId.valid || !sessionId.valid || !sessionKey.valid) {
+    return undefined;
+  }
+  const selectedSession = sessionId.value || sessionKey.value;
+  return runId.value && agentId.value && selectedSession
+    ? JSON.stringify([runId.value, agentId.value, selectedSession])
+    : undefined;
 }

openclaw/plugins/aios-tenant-boundary/openclaw.plugin.json

@@ -2,7 +2,7 @@
   "id": "aios-tenant-boundary",
   "name": "AIOS Tenant Boundary",
   "description": "Enforces AIOS V2R1 tenant provenance and tool policy boundaries.",
-  "version": "2.1.0-rc1",
+  "version": "2.1.0-rc.1",
   "entry": "dist/index.js",
   "configSchema": {
     "type": "object",
@@ -12,7 +12,14 @@
       "trustedSenderIds": { "type": "array", "items": { "type": "string" }, "default": [] },
       "blockedSenderIds": { "type": "array", "items": { "type": "string" }, "default": [] },
       "readOnlyTools": { "type": "array", "items": { "type": "string" } },
-      "requireApprovalForHumanSideEffects": { "type": "boolean", "default": true }
+      "requireApprovalForHumanSideEffects": { "type": "boolean", "default": true },
+      "blockUnknownAgentRuns": { "type": "boolean", "default": false },
+      "approvalTimeoutMs": {
+        "type": "integer",
+        "minimum": 1000,
+        "maximum": 300000,
+        "default": 60000
+      }
     }
   }
 }

openclaw/plugins/aios-tenant-boundary/test/compat.test.js

@@ -0,0 +1,64 @@
+import assert from "node:assert/strict";
+import { readFile } from "node:fs/promises";
+import test from "node:test";
+
+import plugin from "../dist/index.js";
+
+const packageJson = JSON.parse(
+  await readFile(new URL("../package.json", import.meta.url), "utf8"),
+);
+const manifest = JSON.parse(
+  await readFile(new URL("../openclaw.plugin.json", import.meta.url), "utf8"),
+);
+
+function registrations(config = {}) {
+  const seen = [];
+  plugin.register({
+    pluginConfig: config,
+    on(name, handler, options) {
+      seen.push({ name, handler, options });
+    },
+  });
+  return seen;
+}
+
+test("package, manifest, and executable test command are release compatible", () => {
+  assert.equal(packageJson.version, "2.1.0-rc.1");
+  assert.equal(manifest.version, "2.1.0-rc.1");
+  assert.equal(packageJson.scripts.test, "node --test test/*.test.js");
+});
+
+test("schema declares every runtime-read property and exact B1 bounds", () => {
+  const properties = manifest.configSchema.properties;
+  const runtimeKeys = [
+    "approvalTimeoutMs",
+    "blockUnknownAgentRuns",
+    "blockedSenderIds",
+    "readOnlyTools",
+    "requireApprovalForHumanSideEffects",
+    "requireExplicitTrustedSenders",
+    "trustedSenderIds",
+  ];
+  assert.deepEqual(Object.keys(properties).sort(), runtimeKeys);
+  assert.deepEqual(properties.blockUnknownAgentRuns, {
+    type: "boolean",
+    default: false,
+  });
+  assert.deepEqual(properties.approvalTimeoutMs, {
+    type: "integer",
+    minimum: 1000,
+    maximum: 300000,
+    default: 60000,
+  });
+});
+
+test("all three hooks register exactly once at priority 100", () => {
+  const seen = registrations();
+  assert.deepEqual(
+    seen.map(({ name }) => name).sort(),
+    ["agent_end", "before_agent_run", "before_tool_call"],
+  );
+  for (const registration of seen) {
+    assert.deepEqual(registration.options, { priority: 100 });
+  }
+});

openclaw/plugins/aios-tenant-boundary/test/loader.test.mjs

@@ -0,0 +1,50 @@
+import assert from "node:assert/strict";
+import { readFile } from "node:fs/promises";
+import { realpath } from "node:fs/promises";
+
+const [mode, inspectPath] = process.argv.slice(2);
+assert.ok(["default", "opt-in"].includes(mode), `unsupported mode: ${mode}`);
+assert.ok(inspectPath, "inspect JSON path is required");
+
+const data = JSON.parse(await readFile(inspectPath, "utf8"));
+const plugin = data.plugin;
+assert.equal(plugin.id, "aios-tenant-boundary");
+assert.equal(plugin.version, "2.1.0-rc.1");
+assert.equal(plugin.status, "loaded");
+assert.equal(plugin.activated, true);
+assert.equal(await realpath(plugin.rootDir), await realpath(new URL("..", import.meta.url)));
+assert.equal(await realpath(plugin.source), await realpath(new URL("../dist/index.js", import.meta.url)));
+assert.equal(data.install.source, "path");
+assert.equal(await realpath(data.install.installPath), await realpath(plugin.rootDir));
+
+const hooks = data.typedHooks
+  .map(({ name, priority }) => ({ name, priority }))
+  .sort((a, b) => a.name.localeCompare(b.name));
+const expectedNames = mode === "default"
+  ? ["before_tool_call"]
+  : ["agent_end", "before_agent_run", "before_tool_call"];
+assert.deepEqual(hooks.map(({ name }) => name), expectedNames);
+assert.ok(hooks.every(({ priority }) => priority === 100));
+
+const diagnostics = data.diagnostics ?? [];
+assert.equal(
+  diagnostics.filter(({ level, severity }) =>
+    ["error", "fatal"].includes(String(level ?? severity).toLowerCase())).length,
+  0,
+);
+if (mode === "default") {
+  assert.deepEqual(
+    diagnostics.map(({ level, message }) => ({ level, message })).sort((a, b) => a.message.localeCompare(b.message)),
+    ["agent_end", "before_agent_run"].map((hook) => ({
+      level: "warn",
+      message: `typed hook "${hook}" blocked because non-bundled plugins must set plugins.entries.aios-tenant-boundary.hooks.allowConversationAccess=true`,
+    })).sort((a, b) => a.message.localeCompare(b.message)),
+  );
+} else {
+  assert.deepEqual(diagnostics, []);
+}
+
+const properties = plugin.configJsonSchema?.properties ?? {};
+for (const key of ["approvalTimeoutMs", "blockUnknownAgentRuns"]) {
+  assert.ok(Object.hasOwn(properties, key), `missing runtime schema property: ${key}`);
+}

openclaw/plugins/aios-tenant-boundary/test/policy.test.js

@@ -0,0 +1,349 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import plugin from "../dist/index.js";
+import {
+  classifyRun,
+  decideToolCall,
+  resolveSenderId,
+  runKey,
+} from "../dist/policy.js";
+
+function registration(config = {}) {
+  const handlers = new Map();
+  plugin.register({
+    pluginConfig: config,
+    on(name, handler) {
+      handlers.set(name, handler);
+    },
+  });
+  return {
+    beforeAgentRun: handlers.get("before_agent_run"),
+    beforeToolCall: handlers.get("before_tool_call"),
+    agentEnd: handlers.get("agent_end"),
+  };
+}
+
+function agentContext(id, senderId) {
+  return {
+    runId: `run-${id}`,
+    agentId: "agent-main",
+    sessionId: "session-uuid-123",
+    sessionKey: "agent:main:weixin:direct:user",
+    ...(senderId ? { senderId } : {}),
+  };
+}
+
+function toolContext(id, extra = {}) {
+  return {
+    runId: `run-${id}`,
+    agentId: "agent-main",
+    sessionId: "session-uuid-123",
+    sessionKey: "agent:main:weixin:direct:user",
+    ...extra,
+  };
+}
+
+test("sender classification covers missing, blocked, allowlisted, and automation", () => {
+  assert.equal(resolveSenderId({ channelContext: { sender: { id: " nested " } } }), "nested");
+  assert.deepEqual(classifyRun({}, {}), {
+    trusted: false,
+    origin: "system_or_unknown",
+    reason: "sender_id_missing",
+  });
+  assert.equal(
+    classifyRun({ senderId: "blocked" }, { blockedSenderIds: ["blocked"] }).reason,
+    "sender_blocked",
+  );
+  assert.equal(
+    classifyRun(
+      { senderId: "alice" },
+      { requireExplicitTrustedSenders: true, trustedSenderIds: ["alice"] },
+    ).trusted,
+    true,
+  );
+  assert.deepEqual(classifyRun({ jobId: "job-1" }, {}), {
+    trusted: false,
+    origin: "automation",
+    reason: "job_id_present",
+  });
+});
+
+test("unknown agent runs default to pass but fail closed when explicitly enabled", async () => {
+  const defaultHandlers = registration();
+  assert.deepEqual(
+    await defaultHandlers.beforeAgentRun({}, agentContext("default")),
+    { outcome: "pass" },
+  );
+
+  const enforced = registration({ blockUnknownAgentRuns: true });
+  const denied = await enforced.beforeAgentRun({}, agentContext("denied"));
+  assert.equal(denied.outcome, "block");
+  assert.equal(denied.reason, "sender_id_missing");
+  assert.doesNotMatch(denied.reason + denied.message, /undefined|\r|\n/);
+
+  assert.deepEqual(
+    await enforced.beforeAgentRun({}, { ...agentContext("job"), jobId: "job-1" }),
+    { outcome: "pass" },
+  );
+});
+
+test("read-only tools pass without provenance and side effects fail with a stable reason", async () => {
+  assert.deepEqual(decideToolCall("read", undefined, {}), {
+    action: "allow",
+    code: "read_only",
+  });
+  assert.deepEqual(decideToolCall("shell", undefined, {}), {
+    action: "block",
+    code: "run_provenance_missing",
+  });
+
+  const handlers = registration();
+  assert.equal(await handlers.beforeToolCall({ toolName: "read" }, {}), undefined);
+  const blocked = await handlers.beforeToolCall({ toolName: "shell" }, {});
+  assert.equal(blocked.block, true);
+  assert.equal(
+    blocked.blockReason,
+    "AIOS V2R1 origin policy denied tool call: run_provenance_missing",
+  );
+});
+
+test("trusted human side effects request approval with exact timeout validation", async () => {
+  for (const [configured, expected] of [
+    [undefined, 60000],
+    [1000, 1000],
+    [300000, 300000],
+    [999, 60000],
+    [300001, 60000],
+    [60000.5, 60000],
+    ["1000", 60000],
+    [null, 60000],
+  ]) {
+    const config = {
+      requireExplicitTrustedSenders: true,
+      trustedSenderIds: ["alice"],
+    };
+    if (configured !== undefined) config.approvalTimeoutMs = configured;
+    const handlers = registration(config);
+    const id = `approval-${String(configured)}`;
+    await handlers.beforeAgentRun({}, agentContext(id, "alice"));
+    const result = await handlers.beforeToolCall({ toolName: "shell" }, toolContext(id));
+    assert.equal(result.requireApproval.timeoutMs, expected, String(configured));
+    assert.deepEqual(result.requireApproval.allowedDecisions, ["allow-once", "deny"]);
+  }
+});
+
+test("active run provenance is isolated by registration, run, and sender", async () => {
+  const config = {
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice", "bob"],
+  };
+  const first = registration(config);
+  const second = registration(config);
+  await first.beforeAgentRun({}, agentContext("shared", "alice"));
+
+  const otherSender = await first.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("shared", { senderId: "bob" }),
+  );
+  assert.equal(otherSender.block, true);
+  assert.match(otherSender.blockReason, /run_scope_sender_conflict$/);
+
+  const otherRegistration = await second.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("shared"),
+  );
+  assert.equal(otherRegistration.block, true);
+  assert.match(otherRegistration.blockReason, /run_provenance_missing$/);
+});
+
+test("agent_end cleanup removes only the matching shared scope", async () => {
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice", "bob"],
+  });
+  await handlers.beforeAgentRun({}, agentContext("alice-scope", "alice"));
+  await handlers.beforeAgentRun({}, agentContext("bob-scope", "bob"));
+  await handlers.agentEnd({}, toolContext("alice-scope"));
+
+  const alice = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("alice-scope"),
+  );
+  assert.equal(alice.block, true);
+
+  const bob = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("bob-scope"),
+  );
+  assert.ok(bob.requireApproval);
+});
+
+test("run provenance expires after the bounded age", async (t) => {
+  const originalNow = Date.now;
+  t.after(() => { Date.now = originalNow; });
+  let now = 1_000_000;
+  Date.now = () => now;
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice"],
+  });
+  await handlers.beforeAgentRun({}, agentContext("expiring", "alice"));
+  now += 60 * 60 * 1000 + 1;
+  const result = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("expiring"),
+  );
+  assert.equal(result.block, true);
+  assert.match(result.blockReason, /run_provenance_missing$/);
+});
+
+test("host-shaped lifecycle keeps distinct sessionId and sessionKey across all hooks", async () => {
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice"],
+  });
+  const before = agentContext("host-shape", "alice");
+  const tool = toolContext("host-shape");
+  const end = toolContext("host-shape");
+  for (const ctx of [before, tool, end]) {
+    assert.equal(ctx.sessionId, "session-uuid-123");
+    assert.equal(ctx.sessionKey, "agent:main:weixin:direct:user");
+    assert.equal(ctx.runId, "run-host-shape");
+    assert.equal(ctx.agentId, "agent-main");
+  }
+
+  await handlers.beforeAgentRun({}, before);
+  const result = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    tool,
+  );
+  assert.ok(result.requireApproval);
+  assert.equal(result.block, undefined);
+  await handlers.agentEnd({}, end);
+  const cleaned = await handlers.beforeToolCall({ toolName: "shell" }, tool);
+  assert.equal(cleaned.block, true);
+  assert.match(cleaned.blockReason, /run_provenance_missing$/);
+});
+
+test("shared scope collision fails closed without overwriting original sender", async () => {
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice", "bob"],
+  });
+  assert.deepEqual(
+    await handlers.beforeAgentRun({}, agentContext("collision", "alice")),
+    { outcome: "pass" },
+  );
+  const collision = await handlers.beforeAgentRun(
+    {},
+    agentContext("collision", "bob"),
+  );
+  assert.equal(collision.outcome, "block");
+  assert.equal(collision.reason, "run_scope_sender_conflict");
+  assert.doesNotMatch(collision.message, /alice|bob/);
+
+  const original = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    toolContext("collision"),
+  );
+  assert.ok(original.requireApproval);
+});
+
+test("shared scope normalization is exact and incomplete scope fails closed", async () => {
+  const bothAliases = runKey({}, {
+    runId: " run-1 ",
+    agentId: " agent-1 ",
+    sessionId: " session-uuid-123 ",
+    sessionKey: " agent:main:weixin:direct:user ",
+  });
+  assert.equal(
+    bothAliases,
+    JSON.stringify(["run-1", "agent-1", "session-uuid-123"]),
+  );
+
+  const fallback = runKey(
+    { sessionKey: " agent:main:weixin:direct:user " },
+    {
+      runId: "run-1",
+      agentId: "agent-1",
+      sessionKey: "agent:main:weixin:direct:user",
+    },
+  );
+  assert.equal(
+    fallback,
+    JSON.stringify(["run-1", "agent-1", "agent:main:weixin:direct:user"]),
+  );
+  assert.equal(runKey(
+    { sessionId: "event-session" },
+    {
+      runId: "run-1",
+      agentId: "agent-1",
+      sessionId: "context-session",
+      sessionKey: "agent:main:weixin:direct:user",
+    },
+  ), undefined);
+  assert.equal(runKey({}, {
+    runId: "run-1",
+    agentId: "agent-1",
+    sessionKey: "agent:main:weixin:direct:user",
+    sessionId: "session-uuid-123",
+  }), bothAliases);
+  assert.equal(runKey({}, {
+    runId: "run-1",
+    agentId: "agent-1",
+    sessionKey: "agent:main:weixin:direct:user",
+  }), fallback);
+  assert.equal(runKey({}, {
+    agentId: "agent-1", sessionId: "session-uuid-123",
+  }), undefined);
+
+  const handlers = registration();
+  const missing = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    { agentId: "agent-1", sessionId: "session-uuid-123" },
+  );
+  assert.equal(missing.block, true);
+  assert.match(missing.blockReason, /run_provenance_missing$/);
+});
+
+test("sessionKey fallback requires sessionId absent from both hook contexts", async () => {
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice"],
+  });
+  const before = {
+    runId: "run-fallback",
+    agentId: "agent-main",
+    sessionKey: "agent:main:weixin:direct:user",
+    senderId: "alice",
+  };
+  const tool = {
+    runId: "run-fallback",
+    agentId: "agent-main",
+    sessionKey: "agent:main:weixin:direct:user",
+  };
+  await handlers.beforeAgentRun({}, before);
+  assert.ok((await handlers.beforeToolCall({ toolName: "shell" }, tool)).requireApproval);
+
+  const mismatchedSelection = await handlers.beforeToolCall(
+    { toolName: "shell" },
+    { ...tool, sessionId: "session-uuid-123" },
+  );
+  assert.equal(mismatchedSelection.block, true);
+  assert.match(mismatchedSelection.blockReason, /run_provenance_missing$/);
+});
+
+test("same-named event and context sessionId mismatch fails closed", async () => {
+  const handlers = registration({
+    requireExplicitTrustedSenders: true,
+    trustedSenderIds: ["alice"],
+  });
+  await handlers.beforeAgentRun({}, agentContext("event-mismatch", "alice"));
+  const result = await handlers.beforeToolCall(
+    { toolName: "shell", sessionId: "different-session" },
+    toolContext("event-mismatch"),
+  );
+  assert.equal(result.block, true);
+  assert.match(result.blockReason, /run_provenance_missing$/);
+});

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_agent_event_bus.py

@@ -1,5 +1,5 @@
-from active_runtime.app.agent_event_bus import AgentEventBus
-from active_runtime.app.agent_event_dashboard import AgentEventDashboard
+from app.agent_event_bus import AgentEventBus
+from app.agent_event_dashboard import AgentEventDashboard
 
 
 def test_unified_event_flow():

tests/test_agent_observability_loop.py

@@ -1,4 +1,4 @@
-from active_runtime.app.agent_observability_hooks import AgentObservabilityHooks
+from app.agent_observability_hooks import AgentObservabilityHooks
 
 
 def test_agent_observability_closed_loop():

tests/test_architect_governance.py

@@ -1,5 +1,5 @@
-from active_runtime.app.architect_governance_agent import ArchitectGovernanceAgent
-from active_runtime.app.architect_governance_dashboard import ArchitectGovernanceDashboard
+from app.architect_governance_agent import ArchitectGovernanceAgent
+from app.architect_governance_dashboard import ArchitectGovernanceDashboard
 
 
 def test_architect_agent_decides_release_level():

tests/test_architect_research_timeline.py

@@ -1,4 +1,4 @@
-from active_runtime.app.architect_research_timeline import ArchitectResearchTimeline
+from app.architect_research_timeline import ArchitectResearchTimeline
 
 
 def test_research_timeline_tracks_version_decision():

tests/test_architect_weekly_research.py

@@ -1,4 +1,4 @@
-from active_runtime.app.architect_weekly_research_agent import ArchitectWeeklyResearchAgent
+from app.architect_weekly_research_agent import ArchitectWeeklyResearchAgent
 
 
 def test_architect_research_decides_release_type():

tests/test_architecture_change_classifier.py

@@ -1,4 +1,4 @@
-from active_runtime.app.architecture_change_classifier import ArchitectureChangeClassifier
+from app.architecture_change_classifier import ArchitectureChangeClassifier
 
 
 def test_small_change_can_auto_process():

tests/test_architecture_owner_gate.py

@@ -1,4 +1,4 @@
-from active_runtime.app.architecture_owner_gate import ArchitectureOwnerGate
+from app.architecture_owner_gate import ArchitectureOwnerGate
 
 
 def test_architecture_change_blocks_without_owner():

tests/test_change_governance.py

@@ -1,4 +1,4 @@
-from active_runtime.app.change_governance_dashboard import ChangeGovernanceDashboard
+from app.change_governance_dashboard import ChangeGovernanceDashboard
 
 
 def test_change_governance_blocks_missing_evidence():

tests/test_governance_event_bridge.py

@@ -1,4 +1,4 @@
-from active_runtime.app.governance_event_bridge import GovernanceEventBridge
+from app.governance_event_bridge import GovernanceEventBridge
 
 
 def test_governance_events_are_published():

tests/test_governance_timeline.py

@@ -1,4 +1,4 @@
-from active_runtime.app.governance_timeline_dashboard import GovernanceTimelineDashboard
+from app.governance_timeline_dashboard import GovernanceTimelineDashboard
 
 
 def test_governance_timeline_detects_ready_flow():

tests/test_search_agent_adapter.py

@@ -1,4 +1,4 @@
-from active_runtime.app.search_agent_adapter import AgentSearchContext, OpenClawSearchClient
+from app.search_agent_adapter import AgentSearchContext, OpenClawSearchClient
 
 
 class FakeOrchestrator:

tests/test_search_architect_integration.py

@@ -1,4 +1,4 @@
-from active_runtime.app.search_architect_integration import SearchArchitectIntegration
+from app.search_architect_integration import SearchArchitectIntegration
 
 
 def test_search_architect_recommends_release_for_new_capability():

tests/test_search_gateway.py

@@ -1,4 +1,4 @@
-from active_runtime.app.search_gateway import (
+from app.search_gateway import (
     SearchEvidence,
     SearchGateway,
     SearchProviderAttempt,

tests/test_search_gateway_dashboard.py

@@ -1,4 +1,4 @@
-from active_runtime.app.search_gateway_dashboard import SearchGatewayDashboard
+from app.search_gateway_dashboard import SearchGatewayDashboard
 
 
 def test_search_gateway_dashboard_reports_provider_failures():

tests/test_search_gateway_general_capability.py

@@ -1,5 +1,5 @@
-from active_runtime.app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
-from active_runtime.app.search_profile_resolver import SearchProfileResolver
+from app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
+from app.search_profile_resolver import SearchProfileResolver
 
 
 class FailingProvider:

tests/test_search_gateway_orchestrator.py

@@ -1,4 +1,4 @@
-from active_runtime.app.search_gateway_orchestrator import SearchGatewayOrchestrator
+from app.search_gateway_orchestrator import SearchGatewayOrchestrator
 
 
 def test_search_orchestrator_has_audit_registry_event_layers():

tests/test_search_provider_registry.py

@@ -1,5 +1,5 @@
-from active_runtime.app.search_provider_registry import SearchProviderRegistry, StaticArchiveProvider
-from active_runtime.app.search_gateway import SearchGateway, SearchRequest
+from app.search_provider_registry import SearchProviderRegistry, StaticArchiveProvider
+from app.search_gateway import SearchGateway, SearchRequest
 
 
 def test_provider_registry_has_default_layers():

tests/test_task_center.py

@@ -1,4 +1,4 @@
-from active_runtime.app.task_center import AIOSTaskRecord, TaskCenter, TaskNotificationStatus, TaskProgress
+from app.task_center import AIOSTaskRecord, TaskCenter, TaskNotificationStatus, TaskProgress
 
 
 def test_task_center_tracks_progress_and_notification():

tests/test_task_dashboard_view.py

@@ -1,4 +1,4 @@
-from active_runtime.app.task_dashboard_view import TaskDashboardView
+from app.task_dashboard_view import TaskDashboardView
 
 
 def test_task_dashboard_contains_operations_views():

tests/test_task_event_and_statistics.py

@@ -1,5 +1,5 @@
-from active_runtime.app.task_event_bridge import TaskEventBridge
-from active_runtime.app.task_statistics import TaskStatistics
+from app.task_event_bridge import TaskEventBridge
+from app.task_statistics import TaskStatistics
 
 
 def test_task_event_bridge_creates_events():

tests/test_task_lifecycle.py

@@ -1,4 +1,4 @@
-from active_runtime.app.task_lifecycle import TaskLifecycle
+from app.task_lifecycle import TaskLifecycle
 
 
 def test_owner_approval_button_appears_for_waiting_task():

tests/test_task_orchestrator.py

@@ -1,4 +1,4 @@
-from active_runtime.app.task_orchestrator import TaskOrchestrator
+from app.task_orchestrator import TaskOrchestrator
 
 
 def test_command_creates_task():

tests/test_v1r4c1_risk_fixes.py

@@ -1,4 +1,4 @@
-from active_runtime.app.git_repo_locator import GitRepoLocator
+from app.git_repo_locator import GitRepoLocator
 
 
 def test_repo_locator_uses_configurable_root():

tests/test_v2r1_package_identity.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import importlib.util
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+import aios_v2r1
+from aios_v2r1 import cli
+from aios_v2r1.ingress_service import IngressHTTPServer
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_canonical_package_has_one_identity_and_no_compatibility_shim() -> None:
+    assert aios_v2r1.__name__ == "aios_v2r1"
+    assert aios_v2r1.__package__ == "aios_v2r1"
+    assert aios_v2r1.__version__ == "2.1.0-rc.1"
+    assert importlib.util.find_spec("active_runtime") is None
+    assert not (ROOT / "active_runtime").exists()
+
+
+def test_packaging_metadata_declares_only_the_canonical_app_package() -> None:
+    metadata = (ROOT / "active-runtime" / "pyproject.toml").read_text()
+    assert 'packages = ["aios_v2r1"]' in metadata
+    assert 'package-dir = {"" = "app"}' in metadata
+    assert "active_runtime" not in metadata
+    assert 'aios-cell = "aios_v2r1.cli:main"' in metadata
+
+
+def test_registry_path_named_serve_keeps_legacy_status_dispatch(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    calls: list[tuple[str, str]] = []
+
+    @dataclass
+    class Result:
+        tenant_id: str
+        state: str
+
+    class FakeManager:
+        def __init__(self, registry: str) -> None:
+            calls.append(("registry", registry))
+
+        def status(self, tenant_id: str) -> Result:
+            calls.append(("status", tenant_id))
+            return Result(tenant_id=tenant_id, state="healthy")
+
+    monkeypatch.setattr(cli, "CellManager", FakeManager)
+    monkeypatch.setattr(
+        cli,
+        "serve",
+        lambda host, port: pytest.fail("legacy registry command entered service mode"),
+    )
+
+    assert cli.main(["serve", "status", "tenant-a"]) == 0
+    assert calls == [("registry", "serve"), ("status", "tenant-a")]
+    assert '"tenant_id": "tenant-a"' in capsys.readouterr().out
+
+
+def test_packaged_server_waits_for_non_daemon_request_handlers() -> None:
+    assert IngressHTTPServer.daemon_threads is False
+    assert IngressHTTPServer.block_on_close is True

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

tests/v2r1/test_ingress_service.py

@@ -0,0 +1,838 @@
+from __future__ import annotations
+
+import http.client
+import json
+import os
+import signal
+import socket
+import subprocess
+import sys
+import threading
+import time
+from collections.abc import Callable
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator
+
+import pytest
+
+from aios_v2r1 import cli
+from aios_v2r1.ingress_service import (
+    MAX_RESPONSE_BYTES,
+    MAX_REQUEST_BYTES,
+    IngressHTTPServer,
+    IngressRequestHandler,
+    build_server,
+)
+
+
+@contextmanager
+def running_server(processor: Callable[[str, bytes], tuple[int, dict[str, object]]] | None = None) -> Iterator[tuple[str, int]]:
+    server = build_server("127.0.0.1", 0, processor=processor)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        host, port = server.server_address
+        yield str(host), int(port)
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def request(
+    address: tuple[str, int],
+    method: str,
+    path: str,
+    body: bytes | None = None,
+) -> tuple[int, bytes, dict[str, str]]:
+    connection = http.client.HTTPConnection(*address, timeout=5)
+    try:
+        connection.request(method, path, body=body)
+        response = connection.getresponse()
+        payload = response.read()
+        return response.status, payload, {key.lower(): value for key, value in response.getheaders()}
+    finally:
+        connection.close()
+
+
+def raw_request(
+    address: tuple[str, int],
+    method: bytes,
+    path: bytes = b"/health/live",
+) -> bytes:
+    request_bytes = (
+        method + b" " + path + b" HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
+    )
+    with socket.create_connection(address, timeout=5) as connection:
+        connection.sendall(request_bytes)
+        chunks: list[bytes] = []
+        while True:
+            chunk = connection.recv(8192)
+            if not chunk:
+                return b"".join(chunks)
+            chunks.append(chunk)
+
+
+def raw_exchange(
+    address: tuple[str, int],
+    request_bytes: bytes,
+    *,
+    shutdown_write: bool = False,
+    receive_timeout: float = 2.0,
+) -> tuple[bytes, bool]:
+    chunks: list[bytes] = []
+    timed_out = False
+    with socket.create_connection(address, timeout=5) as connection:
+        connection.settimeout(receive_timeout)
+        connection.sendall(request_bytes)
+        if shutdown_write:
+            connection.shutdown(socket.SHUT_WR)
+        while True:
+            try:
+                chunk = connection.recv(8192)
+            except TimeoutError:
+                timed_out = True
+                break
+            except ConnectionResetError:
+                break
+            if not chunk:
+                break
+            chunks.append(chunk)
+    return b"".join(chunks), timed_out
+
+
+def parse_raw_response(response: bytes) -> tuple[int, bytes, bytes]:
+    headers, body = response.split(b"\r\n\r\n", 1)
+    status = int(headers.split(b"\r\n", 1)[0].split(b" ", 2)[1])
+    return status, headers, body
+
+
+def post_request_bytes(headers: bytes, body: bytes = b"") -> bytes:
+    return (
+        b"POST /v1/ingress HTTP/1.1\r\n"
+        b"Host: localhost\r\nConnection: close\r\n"
+        + headers
+        + b"\r\n"
+        + body
+    )
+
+
+def wait_until_listening(process: subprocess.Popen[str], address: tuple[str, int]) -> None:
+    deadline = time.monotonic() + 5
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            _, stderr = process.communicate()
+            pytest.fail(f"server exited before listening: {stderr}")
+        try:
+            with socket.create_connection(address, timeout=0.1):
+                return
+        except OSError:
+            time.sleep(0.05)
+    pytest.fail("server did not start listening")
+
+
+def test_live_and_unconfigured_readiness_are_exact_and_bounded() -> None:
+    with running_server() as address:
+        live = request(address, "GET", "/health/live")
+        ready = request(address, "GET", "/health/ready")
+
+    assert live[:2] == (200, b'{"status":"LIVE"}')
+    assert ready[:2] == (503, b'{"status":"SERVICE_NOT_CONFIGURED"}')
+    assert live[2]["content-type"] == "application/json"
+    assert ready[2]["content-length"] == str(len(ready[1]))
+    assert len(live[1]) <= 4096
+    assert len(ready[1]) <= 4096
+
+
+@pytest.mark.parametrize(
+    "route",
+    ["/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"],
+)
+def test_mutation_routes_fail_closed_without_reading_or_mutating(route: str) -> None:
+    with running_server() as address:
+        status, payload, _ = request(
+            address,
+            "POST",
+            route,
+            b'{"sender":"do-not-log","body":"do-not-log"}',
+        )
+    assert status == 503
+    assert payload == b'{"status":"SERVICE_NOT_CONFIGURED"}'
+
+
+def test_unknown_routes_and_wrong_methods_use_bounded_sanitized_codes() -> None:
+    with running_server() as address:
+        missing = request(address, "GET", "/missing")
+        wrong_method = request(address, "GET", "/v1/ingress")
+        unsupported = request(address, "PUT", "/health/live")
+
+    assert missing[:2] == (404, b'{"status":"NOT_FOUND"}')
+    assert wrong_method[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}')
+    assert unsupported[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}')
+    assert all(len(result[1]) <= 4096 for result in (missing, wrong_method, unsupported))
+
+
+@pytest.mark.parametrize(
+    ("method", "expected_status"),
+    [
+        (b"METHOD_SECRET_REFLECTION_TOKEN", b"405"),
+        (b"METHOD_SECRET_REFLECTION_TOKEN" + b"X" * 66_000, b"414"),
+    ],
+    ids=["arbitrary", "overlong"],
+)
+def test_arbitrary_and_overlong_methods_use_bounded_json_without_reflection(
+    method: bytes, expected_status: bytes
+) -> None:
+    with running_server() as address:
+        response = raw_request(address, method)
+
+    headers, body = response.split(b"\r\n\r\n", 1)
+    status_line = headers.split(b"\r\n", 1)[0]
+    assert status_line.split(b" ", 2)[1] == expected_status
+    assert b"Content-Type: application/json" in headers
+    assert len(body) <= 4096
+    assert b"METHOD_SECRET_REFLECTION_TOKEN" not in response
+    assert b"<html" not in response.lower()
+
+
+@pytest.mark.parametrize("invalid_status", [True, False, 99, 100, 101, 199, 204, 205, 304, 600, 999, "200", 200.0, None])
+def test_invalid_processor_status_is_sanitized_before_writing_status_line(
+    invalid_status: object,
+) -> None:
+    def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
+        return invalid_status, {"status": "MUST_NOT_ESCAPE"}  # type: ignore[return-value]
+
+    with running_server(processor) as address:
+        response = raw_request(address, b"POST", b"/v1/ingress")
+
+    headers, body = response.split(b"\r\n\r\n", 1)
+    assert headers.split(b"\r\n", 1)[0].startswith(b"HTTP/1.0 500 ")
+    assert body == b'{"status":"INTERNAL_ERROR"}'
+    assert b"MUST_NOT_ESCAPE" not in response
+
+
+def test_processor_is_injected_and_exceptions_are_not_disclosed(capsys: pytest.CaptureFixture[str]) -> None:
+    secret = "sender-secret-body-token"
+
+    def exploding_processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
+        assert route == "/v1/ingress"
+        assert json.loads(body) == {"value": secret}
+        raise RuntimeError(secret)
+
+    with running_server(exploding_processor) as address:
+        ready = request(address, "GET", "/health/ready")
+        failed = request(
+            address,
+            "POST",
+            "/v1/ingress",
+            json.dumps({"value": secret}).encode(),
+        )
+
+    captured = capsys.readouterr()
+    assert ready[:2] == (200, b'{"status":"READY"}')
+    assert failed[:2] == (500, b'{"status":"INTERNAL_ERROR"}')
+    assert secret not in failed[1].decode()
+    assert secret not in captured.out
+    assert secret not in captured.err
+
+
+@pytest.mark.parametrize(
+    ("headers", "body"),
+    [
+        (b"Transfer-Encoding: secret-framing-token\r\n", b"secret-body"),
+        (b"Content-Length: 0\r\nContent-Length: 0\r\n", b""),
+        (b"Content-Length: 1\r\nContent-Length: 2\r\n", b"x"),
+        (b"Content-Length: invalid-secret-length\r\n", b""),
+        (b"Content-Length: -1\r\n", b""),
+        (f"Content-Length: {MAX_REQUEST_BYTES + 1}\r\n".encode(), b""),
+        (b"Content-Length: 0 \r\n", b""),
+        (b"Content-Length:\t0\r\n", b""),
+    ],
+    ids=[
+        "transfer-encoding",
+        "duplicate-same",
+        "duplicate-conflicting",
+        "invalid",
+        "negative",
+        "oversize",
+        "trailing-ows",
+        "leading-ows",
+    ],
+)
+def test_invalid_post_framing_is_bounded_and_never_calls_processor(
+    headers: bytes,
+    body: bytes,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    calls: list[bytes] = []
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        calls.append(payload)
+        return 201, {"status": "MUTATED"}
+
+    with running_server(processor) as address:
+        response, timed_out = raw_exchange(
+            address,
+            post_request_bytes(headers, body),
+        )
+
+    status, response_headers, response_body = parse_raw_response(response)
+    captured = capsys.readouterr()
+    assert not timed_out
+    assert 400 <= status <= 499
+    assert b"Content-Type: application/json" in response_headers
+    assert len(response_body) <= 4096
+    assert b"secret-framing-token" not in response
+    assert b"secret-body" not in response
+    assert b"invalid-secret-length" not in response
+    assert calls == []
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+
+
+def test_short_eof_body_is_rejected_before_processor() -> None:
+    calls: list[bytes] = []
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        calls.append(payload)
+        return 201, {"status": "MUTATED"}
+
+    with running_server(processor) as address:
+        response, timed_out = raw_exchange(
+            address,
+            post_request_bytes(b"Content-Length: 10\r\n", b"abc"),
+            shutdown_write=True,
+        )
+
+    status, _, body = parse_raw_response(response)
+    assert not timed_out
+    assert 400 <= status <= 499
+    assert len(body) <= 4096
+    assert calls == []
+
+
+def test_partial_body_timeout_is_bounded_and_rejected_before_processor() -> None:
+    processor_called = threading.Event()
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        processor_called.set()
+        return 201, {"status": "MUTATED"}
+
+    with running_server(processor) as address:
+        started = time.monotonic()
+        response, timed_out = raw_exchange(
+            address,
+            post_request_bytes(b"Content-Length: 10\r\n", b"abc"),
+            receive_timeout=1.5,
+        )
+        elapsed = time.monotonic() - started
+        called_after_client_close = processor_called.wait(timeout=0.5)
+
+    status, _, body = parse_raw_response(response)
+    assert not timed_out
+    assert elapsed < 1.5
+    assert status == 408
+    assert len(body) <= 4096
+    assert not called_after_client_close
+
+
+def test_server_close_drains_partial_body_handler_and_prevents_late_processor(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    processor_called = threading.Event()
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        processor_called.set()
+        return 201, {"status": "MUTATED"}
+
+    server = build_server("127.0.0.1", 0, processor=processor)
+    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+    server_thread.start()
+    host, port = server.server_address
+    client = socket.create_connection((str(host), int(port)), timeout=5)
+    try:
+        client.sendall(post_request_bytes(b"Content-Length: 10\r\n", b"abc"))
+        time.sleep(0.1)
+        started = time.monotonic()
+        server.shutdown()
+        server.server_close()
+        close_elapsed = time.monotonic() - started
+    finally:
+        client.close()
+        server_thread.join(timeout=5)
+
+    called_after_close = processor_called.wait(timeout=0.5)
+    captured = capsys.readouterr()
+    assert server.daemon_threads is False
+    assert server.block_on_close is True
+    assert close_elapsed < 3
+    assert not server_thread.is_alive()
+    assert not called_after_close
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+    assert "Fatal Python error" not in captured.err
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as rebound:
+        rebound.bind((str(host), int(port)))
+
+
+@pytest.mark.parametrize(
+    "target",
+    [
+        b"http://[::1/v1/ingress",
+        b"http://127.0.0.1:bad/v1/ingress",
+        b"relative-target",
+        b"/v1/ingress%ZZ",
+    ],
+    ids=[
+        "invalid-ipv6-bracket",
+        "invalid-absolute-port",
+        "invalid-origin-form",
+        "invalid-percent-escape",
+    ],
+)
+def test_malformed_request_targets_return_sanitized_400_without_processor(
+    target: bytes,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    calls: list[bytes] = []
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        calls.append(payload)
+        return 201, {"status": "MUTATED"}
+
+    request_bytes = (
+        b"POST "
+        + target
+        + b" HTTP/1.1\r\nHost: localhost\r\n"
+        + b"Content-Length: 0\r\nConnection: close\r\n\r\n"
+    )
+    with running_server(processor) as address:
+        response, timed_out = raw_exchange(address, request_bytes)
+
+    status, headers, body = parse_raw_response(response)
+    captured = capsys.readouterr()
+    assert not timed_out
+    assert status == 400
+    assert b"Content-Type: application/json" in headers
+    assert len(body) <= 4096
+    assert target not in response
+    assert calls == []
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+
+
+def test_server_handle_error_never_prints_payload_traceback(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    secret = "unexpected-handler-secret"
+
+    class ExplodingHandler(IngressRequestHandler):
+        def do_GET(self) -> None:
+            raise RuntimeError(secret)
+
+    server = IngressHTTPServer(("127.0.0.1", 0), ExplodingHandler, None)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        host, port = server.server_address
+        raw_exchange(
+            (str(host), int(port)),
+            b"GET /health/live HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+    captured = capsys.readouterr()
+    assert secret not in captured.out
+    assert secret not in captured.err
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+
+
+@pytest.mark.parametrize(
+    "malformed_header",
+    [
+        b"Bad Name: value\r\n",
+        b"Bad\tName: value\r\n",
+        b"Trailing-Name : value\r\n",
+        b" X-Leading: value\r\n",
+        b"\tX-Leading: value\r\n",
+        b":authority: localhost\r\n",
+        b"X-Obs-Fold: first\r\n continuation\r\n",
+        b"X-Obs-Fold: first\r\n\tcontinuation\r\n",
+        b"X-Control: value\x00secret\r\n",
+        b"X-Carriage: value\rsecret\r\n",
+        b"X-Bare-LF: value\n",
+        b"X-\xff: value\r\n",
+    ],
+    ids=[
+        "space-in-name",
+        "tab-in-name",
+        "whitespace-before-colon",
+        "leading-space",
+        "leading-tab",
+        "pseudo-header",
+        "obs-fold-space",
+        "obs-fold-tab",
+        "nul-in-value",
+        "cr-in-value",
+        "bare-lf",
+        "non-ascii-name",
+    ],
+)
+def test_malformed_raw_header_grammar_is_rejected_before_processor(
+    malformed_header: bytes,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    calls: list[bytes] = []
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        calls.append(payload)
+        return 201, {"status": "MUTATED"}
+
+    request_bytes = (
+        b"POST /v1/ingress HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        + malformed_header
+        + b"Content-Length: 0\r\nConnection: close\r\n\r\n"
+    )
+    with running_server(processor) as address:
+        response, timed_out = raw_exchange(address, request_bytes)
+
+    assert not timed_out
+    assert response
+    status, headers, body = parse_raw_response(response)
+    captured = capsys.readouterr()
+    assert status == 400
+    assert b"Content-Type: application/json" in headers
+    assert json.loads(body) == {"status": "BAD_REQUEST"}
+    assert len(body) <= MAX_RESPONSE_BYTES
+    assert b"secret" not in response
+    assert calls == []
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+
+
+def trickle_request_until_response(
+    address: tuple[str, int],
+    prefix: bytes,
+    *,
+    receive_timeout: float = 1.8,
+) -> tuple[bytes, bool, float]:
+    connection = socket.create_connection(address, timeout=5)
+    stop = threading.Event()
+
+    def sender() -> None:
+        while not stop.wait(0.1):
+            try:
+                connection.sendall(b"x")
+            except OSError:
+                return
+
+    connection.sendall(prefix)
+    sender_thread = threading.Thread(target=sender, daemon=True)
+    sender_thread.start()
+    connection.settimeout(receive_timeout)
+    chunks: list[bytes] = []
+    timed_out = False
+    started = time.monotonic()
+    try:
+        while True:
+            try:
+                chunk = connection.recv(8192)
+            except TimeoutError:
+                timed_out = True
+                break
+            except ConnectionResetError:
+                break
+            if not chunk:
+                break
+            chunks.append(chunk)
+    finally:
+        elapsed = time.monotonic() - started
+        stop.set()
+        try:
+            connection.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+        connection.close()
+        sender_thread.join(timeout=1)
+    return b"".join(chunks), timed_out, elapsed
+
+
+@pytest.mark.parametrize(
+    "prefix",
+    [
+        (
+            b"POST /v1/ingress HTTP/1.1\r\n"
+            b"Host: localhost\r\nX-Slow-Header: "
+        ),
+        (
+            b"POST /v1/ingress HTTP/1.1\r\n"
+            b"Host: localhost\r\nContent-Length: 1000\r\n"
+            b"Connection: close\r\n\r\nx"
+        ),
+    ],
+    ids=["slow-header", "slow-body"],
+)
+def test_absolute_network_deadline_cannot_be_reset_by_slow_trickle(
+    prefix: bytes,
+) -> None:
+    processor_called = threading.Event()
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        processor_called.set()
+        return 201, {"status": "MUTATED"}
+
+    with running_server(processor) as address:
+        response, timed_out, elapsed = trickle_request_until_response(address, prefix)
+        called_after_cleanup = processor_called.wait(timeout=0.4)
+
+    assert not timed_out
+    assert elapsed < 1.8
+    assert response
+    status, _, body = parse_raw_response(response)
+    assert status == 408
+    assert json.loads(body) == {"status": "REQUEST_TIMEOUT"}
+    assert len(body) <= MAX_RESPONSE_BYTES
+    assert not called_after_cleanup
+
+
+def test_shutdown_drains_several_slow_clients_without_late_work_or_threads(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    processor_called = threading.Event()
+
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        processor_called.set()
+        return 201, {"status": "MUTATED"}
+
+    before_non_daemon = {id(thread) for thread in threading.enumerate() if not thread.daemon}
+    server = build_server("127.0.0.1", 0, processor=processor)
+    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+    server_thread.start()
+    host, port = server.server_address
+    clients: list[socket.socket] = []
+    stops: list[threading.Event] = []
+    senders: list[threading.Thread] = []
+
+    def start_client(prefix: bytes) -> None:
+        client = socket.create_connection((str(host), int(port)), timeout=5)
+        client.sendall(prefix)
+        stop = threading.Event()
+
+        def sender() -> None:
+            while not stop.wait(0.1):
+                try:
+                    client.sendall(b"x")
+                except OSError:
+                    return
+
+        sender_thread = threading.Thread(target=sender, daemon=True)
+        sender_thread.start()
+        clients.append(client)
+        stops.append(stop)
+        senders.append(sender_thread)
+
+    header_prefix = (
+        b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\nX-Slow: "
+    )
+    body_prefix = (
+        b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\n"
+        b"Content-Length: 1000\r\nConnection: close\r\n\r\nx"
+    )
+    for index in range(6):
+        start_client(header_prefix if index % 2 == 0 else body_prefix)
+    time.sleep(0.15)
+
+    closed = threading.Event()
+    close_elapsed: list[float] = []
+
+    def close_server() -> None:
+        started = time.monotonic()
+        try:
+            server.shutdown()
+            server.server_close()
+        finally:
+            close_elapsed.append(time.monotonic() - started)
+            closed.set()
+
+    closer = threading.Thread(target=close_server)
+    closer.start()
+    closed_within_bound = closed.wait(timeout=2.5)
+    for stop in stops:
+        stop.set()
+    for client in clients:
+        try:
+            client.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+        client.close()
+    for sender_thread in senders:
+        sender_thread.join(timeout=1)
+    closer.join(timeout=5)
+    server_thread.join(timeout=5)
+
+    called_after_close = processor_called.wait(timeout=0.4)
+    deadline = time.monotonic() + 1
+    leaked_non_daemon: list[threading.Thread] = []
+    while time.monotonic() < deadline:
+        leaked_non_daemon = [
+            thread
+            for thread in threading.enumerate()
+            if not thread.daemon
+            and id(thread) not in before_non_daemon
+            and thread.is_alive()
+        ]
+        if not leaked_non_daemon:
+            break
+        time.sleep(0.02)
+    captured = capsys.readouterr()
+    assert closed_within_bound
+    assert close_elapsed and close_elapsed[0] < 2.5
+    assert not server_thread.is_alive()
+    assert not closer.is_alive()
+    assert not leaked_non_daemon
+    assert not called_after_close
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+    assert "Fatal Python error" not in captured.err
+    restarted = build_server(str(host), int(port))
+    restarted.server_close()
+
+
+def test_completed_framing_cancels_deadline_before_slow_processor() -> None:
+    def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
+        time.sleep(1.2)
+        return 201, {"status": "MUTATED"}
+
+    with running_server(processor) as address:
+        status, body, _ = request(address, "POST", "/v1/ingress", b"")
+
+    assert status == 201
+    assert json.loads(body) == {"status": "MUTATED"}
+    assert not [
+        thread
+        for thread in threading.enumerate()
+        if thread.name.startswith("aios-ingress-read-deadline")
+        and thread.is_alive()
+    ]
+
+
+@pytest.mark.parametrize(
+    "payload_factory",
+    [
+        pytest.param(lambda: [], id="list"),
+        pytest.param(lambda: "not-an-object", id="string"),
+        pytest.param(lambda: None, id="null"),
+        pytest.param(lambda: {"value": float("nan")}, id="nan"),
+        pytest.param(lambda: {"value": float("inf")}, id="infinity"),
+        pytest.param(lambda: {"value": float("-inf")}, id="negative-infinity"),
+        pytest.param(lambda: {"value": object()}, id="nonserializable"),
+        pytest.param(
+            lambda: {"value": "secret-oversize-" + "x" * MAX_RESPONSE_BYTES},
+            id="oversize",
+        ),
+    ],
+)
+def test_invalid_processor_payload_is_fixed_bounded_json_500(
+    payload_factory: Callable[[], object],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    payload = payload_factory()
+
+    def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
+        return 201, payload  # type: ignore[return-value]
+
+    with running_server(processor) as address:
+        response = raw_request(address, b"POST", b"/v1/ingress")
+
+    status, headers, body = parse_raw_response(response)
+    captured = capsys.readouterr()
+    assert status == 500
+    assert b"Content-Type: application/json" in headers
+    assert body == b'{"status":"INTERNAL_ERROR"}'
+    assert json.loads(body) == {"status": "INTERNAL_ERROR"}
+    assert len(body) <= MAX_RESPONSE_BYTES
+    assert b"secret-oversize" not in response
+    assert "Traceback" not in captured.out
+    assert "Traceback" not in captured.err
+
+
+def test_recursive_processor_payload_uses_nonrecursive_fixed_500() -> None:
+    recursive: dict[str, object] = {}
+    recursive["self"] = recursive
+
+    def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
+        return 201, recursive
+
+    with running_server(processor) as address:
+        status, body, _ = request(address, "POST", "/v1/ingress", b"")
+
+    assert status == 500
+    assert body == b'{"status":"INTERNAL_ERROR"}'
+    assert json.loads(body) == {"status": "INTERNAL_ERROR"}
+
+
+def test_server_rejects_non_loopback_bind() -> None:
+    with pytest.raises(ValueError, match="loopback"):
+        build_server("0.0.0.0", 0)
+
+
+def test_af_inet_server_rejects_ipv6_loopback_explicitly() -> None:
+    with pytest.raises(ValueError, match="IPv4 loopback"):
+        build_server("::1", 0)
+
+
+def test_sigint_shutdown_exits_zero_without_traceback_and_releases_port() -> None:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation:
+        reservation.bind(("127.0.0.1", 0))
+        port = int(reservation.getsockname()[1])
+
+    env = os.environ.copy()
+    package_root = str(Path(__file__).resolve().parents[2] / "active-runtime" / "app")
+    env["PYTHONPATH"] = package_root + os.pathsep + env.get("PYTHONPATH", "")
+    process = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "aios_v2r1.cli",
+            "serve",
+            "--host",
+            "127.0.0.1",
+            "--port",
+            str(port),
+        ],
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        wait_until_listening(process, ("127.0.0.1", port))
+        process.send_signal(signal.SIGINT)
+        stdout, stderr = process.communicate(timeout=5)
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.communicate()
+
+    assert process.returncode == 0
+    assert "Traceback" not in stdout
+    assert "Traceback" not in stderr
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as rebound:
+        rebound.bind(("127.0.0.1", port))
+
+
+def test_cli_preserves_registry_commands_and_dispatches_top_level_serve(monkeypatch: pytest.MonkeyPatch) -> None:
+    parsed = cli.build_parser().parse_args(["registry.sqlite", "status", "tenant-a"])
+    assert parsed.registry == "registry.sqlite"
+    assert parsed.command == "status"
+
+    called: list[tuple[str, int]] = []
+    monkeypatch.setattr(cli, "serve", lambda host, port: called.append((host, port)))
+    assert cli.main(["serve", "--host", "127.0.0.1", "--port", "19001"]) == 0
+    assert called == [("127.0.0.1", 19001)]