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,289 @@
+"""Fail-closed loopback HTTP shell for V2R1 ingress processors."""
+from __future__ import annotations
+
+import ipaddress
+import json
+import socket
+import threading
+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
+HEX_DIGITS = frozenset("0123456789abcdefABCDEF")
+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.processor = processor
+        super().__init__(address, handler)
+
+    @property
+    def is_closing(self) -> bool:
+        return self._closing.is_set()
+
+    def shutdown(self) -> None:
+        self._closing.set()
+        super().shutdown()
+
+    def server_close(self) -> None:
+        self._closing.set()
+        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 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)
+
+    def handle_one_request(self) -> None:
+        try:
+            super().handle_one_request()
+        except RequestRejected as rejection:
+            self.close_connection = True
+            self._send_safely(rejection.status, {"status": rejection.payload_status})
+        except (BrokenPipeError, ConnectionResetError, TimeoutError, OSError):
+            self.close_connection = True
+        except Exception:
+            self.close_connection = True
+            self._send_safely(500, {"status": "INTERNAL_ERROR"})
+
+    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 "/"
+
+    def _send(self, status: int, payload: Mapping[str, Any]) -> None:
+        if type(status) is not int or not 100 <= status <= 599:
+            status = 500
+            payload = {"status": "INTERNAL_ERROR"}
+        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":
+            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].strip()
+        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
+        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
+        if self.server.is_closing:
+            self._reject(503, "SERVICE_UNAVAILABLE")
+            return
+        try:
+            status, payload = processor(path, body)
+            if (
+                type(status) is not int
+                or not 100 <= status <= 599
+                or not isinstance(payload, Mapping)
+            ):
+                raise TypeError("invalid processor response")
+        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"]

tests/test_agent_event_bus.py

@@ -1,5 +1,5 @@
-from active_runtime.app.agent_event_bus import AgentEventBus
+from app.agent_event_bus import AgentEventBus
-from active_runtime.app.agent_event_dashboard import AgentEventDashboard
+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 app.architect_governance_agent import ArchitectGovernanceAgent
-from active_runtime.app.architect_governance_dashboard import ArchitectGovernanceDashboard
+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 app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
-from active_runtime.app.search_profile_resolver import SearchProfileResolver
+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 app.search_provider_registry import SearchProviderRegistry, StaticArchiveProvider
-from active_runtime.app.search_gateway import SearchGateway, SearchRequest
+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 app.task_event_bridge import TaskEventBridge
-from active_runtime.app.task_statistics import TaskStatistics
+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/v2r1/test_ingress_service.py

@@ -0,0 +1,508 @@
+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_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, 600, 999, "200", 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""),
+    ],
+    ids=[
+        "transfer-encoding",
+        "duplicate-same",
+        "duplicate-conflicting",
+        "invalid",
+        "negative",
+        "oversize",
+    ],
+)
+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
+
+
+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)]