feat(v2r1): package canonical authenticated ingress

active-runtime/app/aios_v2r1/cli.py

@@ -39,12 +39,7 @@ def build_serve_parser() -> argparse.ArgumentParser:
 
 def main(argv: list[str] | None = None) -> int:
     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:
+    if values and values[0] == "serve":
         serve_args = build_serve_parser().parse_args(values[1:])
         serve(serve_args.host, serve_args.port)
         return 0

active-runtime/app/aios_v2r1/ingress_service.py

@@ -3,9 +3,6 @@ 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
@@ -14,31 +11,16 @@ 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
+    daemon_threads = True
 
     def __init__(
         self,
@@ -46,387 +28,39 @@ class IngressHTTPServer(ThreadingHTTPServer):
         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._server_close_lock = threading.Lock()
         self.processor = processor
         super().__init__(address, handler)
 
-    @property
-    def is_closing(self) -> bool:
-        return self._closing.is_set()
-
-    def get_request(self) -> tuple[socket.socket, tuple[str, int]]:
-        """Register each accepted socket before handler scheduling can begin."""
-        connection, client_address = super().get_request()
-        with self._lifecycle_lock:
-            if self._closing.is_set():
-                reject = True
-            else:
-                with self._connections_lock:
-                    self._connections.add(connection)
-                reject = False
-        if reject:
-            self._shutdown_and_close_connection(connection)
-            raise OSError("server is closing")
-        return connection, client_address
-
-    def unregister_connection(self, connection: socket.socket) -> None:
-        with self._connections_lock:
-            self._connections.discard(connection)
-
-    def _shutdown_and_close_connection(self, connection: socket.socket) -> None:
-        try:
-            connection.shutdown(socket.SHUT_RDWR)
-        except OSError:
-            pass
-        try:
-            connection.close()
-        except OSError:
-            pass
-        finally:
-            self.unregister_connection(connection)
-
-    def _close_accepted_connections(self) -> None:
-        with self._connections_lock:
-            connections = tuple(self._connections)
-        for connection in connections:
-            self._shutdown_and_close_connection(connection)
-
-    def process_request(
-        self,
-        request: socket.socket,
-        client_address: tuple[str, int],
-    ) -> None:
-        """Linearize request admission with the transition to closing."""
-        with self._lifecycle_lock:
-            if self._closing.is_set():
-                reject = True
-            else:
-                reject = False
-                try:
-                    super().process_request(request, client_address)
-                except Exception:
-                    self._shutdown_and_close_connection(request)
-                    raise
-        if reject:
-            self._shutdown_and_close_connection(request)
-
-    def shutdown_request(self, request: socket.socket) -> None:
-        try:
-            request.shutdown(socket.SHUT_RDWR)
-        except OSError:
-            pass
-        try:
-            super().shutdown_request(request)
-        finally:
-            self.unregister_connection(request)
-
-    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 _begin_closing(self) -> None:
-        with self._lifecycle_lock:
-            self._closing.set()
-
-    def shutdown(self) -> None:
-        self._begin_closing()
-        self._close_accepted_connections()
-        super().shutdown()
-
-    def server_close(self) -> None:
-        """Interrupt accepted sockets before the base class joins handlers."""
-        with self._server_close_lock:
-            self._begin_closing()
-            self._close_accepted_connections()
-            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
-
-    def finish(self) -> None:
-        self._complete_network_read()
-        try:
-            super().finish()
-        except OSError:
-            pass
-
-    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
+        return urlsplit(self.path).path
 
     def _send(self, status: int, payload: Mapping[str, Any]) -> None:
-        self._complete_network_read()
-        status, body = self._encoded_response(status, payload)
+        body = json.dumps(
+            dict(payload), ensure_ascii=False, separators=(",", ":"), sort_keys=True
+        ).encode("utf-8")
+        if len(body) > MAX_RESPONSE_BYTES:
+            status = 500
+            body = b'{"status":"INTERNAL_ERROR"}'
         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":
+        if self.command != "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":
@@ -449,34 +83,22 @@ class IngressRequestHandler(BaseHTTPRequestHandler):
             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")
+            length = int(self.headers.get("Content-Length", "0"))
+            if length < 0 or length > MAX_REQUEST_BYTES:
+                self._send(413, {"status": "REQUEST_TOO_LARGE"})
                 return
-            status, payload = result
+            body = self.rfile.read(length)
+            status, payload = processor(path, body)
+            if not isinstance(status, int) or not isinstance(payload, Mapping):
+                raise TypeError("invalid processor response")
+            self._send(status, payload)
         except Exception:
             self._send(500, {"status": "INTERNAL_ERROR"})
-            return
-        self._send(status, payload)
 
     do_DELETE = _method_not_allowed
     do_HEAD = _method_not_allowed
@@ -496,8 +118,8 @@ def build_server(
         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")
+    if not address.is_loopback:
+        raise ValueError("host must be a loopback IP address")
     return IngressHTTPServer((host, port), IngressRequestHandler, processor)
 
 
@@ -506,7 +128,5 @@ def serve(host: str = "127.0.0.1", port: int = 18792) -> None:
     server = build_server(host, port)
     try:
         server.serve_forever()
-    except KeyboardInterrupt:
-        return
     finally:
         server.server_close()

tests/test_v2r1_package_identity.py

@@ -1,14 +1,9 @@
 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]
@@ -28,38 +23,3 @@ def test_packaging_metadata_declares_only_the_canonical_app_package() -> None:
     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/v2r1/test_ingress_service.py

@@ -2,29 +2,15 @@ from __future__ import annotations
 
 import http.client
 import json
-import os
-import signal
-import socket
-import subprocess
-import sys
 import threading
-import textwrap
-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,
-)
+from aios_v2r1.ingress_service import build_server
 
 
 @contextmanager
@@ -57,82 +43,6 @@ def request(
         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")
@@ -174,45 +84,6 @@ def test_unknown_routes_and_wrong_methods_use_bounded_sanitized_codes() -> None:
     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"
 
@@ -238,835 +109,11 @@ def test_processor_is_injected_and_exceptions_are_not_disclosed(capsys: pytest.C
     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
-    restarted = build_server(str(host), int(port))
-    restarted.server_close()
-
-
-@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 run_pre_header_sigint_probe(
-    timeout: float = 1.2,
-) -> tuple[int, str, str, float, bool, int]:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation:
-        reservation.bind(("127.0.0.1", 0))
-        port = int(reservation.getsockname()[1])
-
-    script = textwrap.dedent(
-        """
-        import os
-        import signal
-        import socket
-        import sys
-        import threading
-        import time
-
-        from aios_v2r1.ingress_service import (
-            IngressHTTPServer,
-            IngressRequestHandler,
-        )
-
-        accepted_before_setup = threading.Event()
-
-        class PreHeaderReadinessHandler(IngressRequestHandler):
-            def setup(self):
-                accepted_before_setup.set()
-                try:
-                    self.request.recv(1)
-                except OSError:
-                    pass
-                super().setup()
-
-        server = IngressHTTPServer(
-            ("127.0.0.1", int(sys.argv[1])),
-            PreHeaderReadinessHandler,
-            None,
-        )
-
-        def connect_and_interrupt():
-            client = socket.create_connection(server.server_address, timeout=2)
-            if not accepted_before_setup.wait(timeout=2):
-                os._exit(71)
-            os.kill(os.getpid(), signal.SIGINT)
-            time.sleep(10)
-            client.close()
-
-        trigger = threading.Thread(target=connect_and_interrupt, daemon=True)
-        trigger.start()
-        try:
-            server.serve_forever(poll_interval=0.01)
-        except KeyboardInterrupt:
-            pass
-        finally:
-            server.server_close()
-        """
-    )
-    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, "-c", script, str(port)],
-        env=env,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True,
-    )
-    started = time.monotonic()
-    timed_out = False
-    try:
-        stdout, stderr = process.communicate(timeout=timeout)
-    except subprocess.TimeoutExpired:
-        timed_out = True
-        process.kill()
-        stdout, stderr = process.communicate(timeout=2)
-    return (
-        int(process.returncode),
-        stdout,
-        stderr,
-        time.monotonic() - started,
-        timed_out,
-        port,
-    )
-
-
-def test_sigint_closes_pre_header_readiness_socket_in_25_bounded_runs() -> None:
-    for iteration in range(25):
-        returncode, stdout, stderr, elapsed, timed_out, port = (
-            run_pre_header_sigint_probe()
-        )
-        assert not timed_out, f"pre-header SIGINT hung on iteration {iteration}"
-        assert elapsed < 1.2
-        assert returncode == 0
-        assert "Traceback" not in stdout
-        assert "Traceback" not in stderr
-        assert "Fatal Python error" not in stderr
-        restarted = build_server("127.0.0.1", port)
-        restarted.server_close()
-
-
-def test_server_close_closes_socket_accepted_before_handler_setup() -> None:
-    server = build_server("127.0.0.1", 0)
-    client = socket.create_connection(server.server_address, timeout=2)
-    accepted, _ = server.get_request()
-    client.settimeout(0.4)
-    try:
-        server.server_close()
-        server.server_close()
-        try:
-            remaining = client.recv(1)
-        except ConnectionResetError:
-            remaining = b""
-        assert remaining == b""
-    finally:
-        accepted.close()
-        client.close()
-        server.server_close()
-
-
-def test_process_request_rejects_socket_after_server_closing() -> None:
-    handler_started = threading.Event()
-
-    class RecordingHandler:
-        def __init__(self, *args: object) -> None:
-            handler_started.set()
-
-    server = IngressHTTPServer(
-        ("127.0.0.1", 0),
-        RecordingHandler,  # type: ignore[arg-type]
-        None,
-    )
-    server.server_close()
-    request_socket, peer = socket.socketpair()
-    peer.settimeout(0.4)
-    try:
-        server.process_request(request_socket, ("127.0.0.1", 1))
-        try:
-            remaining = peer.recv(1)
-        except ConnectionResetError:
-            remaining = b""
-        assert remaining == b""
-        assert not handler_started.wait(timeout=0.1)
-    finally:
-        request_socket.close()
-        peer.close()
-        server.server_close()
-
-
-
-def test_accept_and_server_close_race_is_bounded_in_25_runs() -> None:
-    errors: list[BaseException] = []
-    for _ in range(25):
-        class RecordingHandler:
-            def __init__(self, *args: object) -> None:
-                return
-
-        server = IngressHTTPServer(
-            ("127.0.0.1", 0),
-            RecordingHandler,  # type: ignore[arg-type]
-            None,
-        )
-        client = socket.create_connection(server.server_address, timeout=2)
-        barrier = threading.Barrier(3)
-
-        def accept_and_process() -> None:
-            try:
-                barrier.wait(timeout=1)
-                request_socket, client_address = server.get_request()
-                server.process_request(request_socket, client_address)
-            except OSError:
-                return
-            except BaseException as exc:
-                errors.append(exc)
-
-        def close_during_accept() -> None:
-            try:
-                barrier.wait(timeout=1)
-                server.server_close()
-            except BaseException as exc:
-                errors.append(exc)
-
-        accepter = threading.Thread(target=accept_and_process)
-        closer = threading.Thread(target=close_during_accept)
-        accepter.start()
-        closer.start()
-        barrier.wait(timeout=1)
-        accepter.join(timeout=1)
-        closer.join(timeout=1)
-        client.close()
-        server.server_close()
-        assert not accepter.is_alive()
-        assert not closer.is_alive()
-    assert not errors
-
-def test_server_close_is_idempotent_with_idle_and_partial_clients(
-    capsys: pytest.CaptureFixture[str],
-) -> None:
-    server = build_server("127.0.0.1", 0)
-    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
-    server_thread.start()
-    clients: list[socket.socket] = []
-    payloads = [
-        None,
-        None,
-        b"GET /health/live HTTP/1.1\r\nHost: localhost\r\n",
-        b"POST /v1/ingress HTTP/1.1\r\nContent-Length: 10\r\n\r\nx",
-    ]
-    for payload in payloads:
-        client = socket.create_connection(server.server_address, timeout=2)
-        if payload is not None:
-            client.sendall(payload)
-        clients.append(client)
-    time.sleep(0.05)
-    server.shutdown()
-
-    errors: list[BaseException] = []
-
-    def close_server() -> None:
-        try:
-            server.server_close()
-        except BaseException as exc:
-            errors.append(exc)
-
-    closers = [threading.Thread(target=close_server) for _ in range(4)]
-    for closer in closers:
-        closer.start()
-    for closer in closers:
-        closer.join(timeout=2)
-    server_thread.join(timeout=2)
-    for client in clients:
-        client.close()
-    captured = capsys.readouterr()
-    assert not errors
-    assert not [closer for closer in closers if closer.is_alive()]
-    assert not server_thread.is_alive()
-    assert "Traceback" not in captured.out
-    assert "Traceback" not in captured.err
-    assert "Fatal Python error" not in captured.err
-
-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"