fix(orchestrator): bound streamed request bodies

Dockerfile.orchestrator

@@ -9,12 +9,9 @@
 #     Keeping the content in one place means future orchestrator deps (e.g.
 #     iroh) are added here once, not duplicated per backend.
 #
-# It stays deliberately lean: the control plane is **stdlib-only** today, so
+# It stays deliberately lean: only the pinned FastAPI/Uvicorn control-plane
-# no third-party payload — none of the gateway's mitmproxy/git/gitleaks
+# stack is installed here — none of the gateway's mitmproxy/git/gitleaks
-# (that's Dockerfile.gateway) and no buildah (that's the firecracker
+# (that's Dockerfile.gateway) and no buildah (that's firecracker-only).
-# builder, and lives only in Dockerfile.orchestrator.fc). Keeping the
-# secret-dense control plane on a minimal dependency surface is the point
-# (PRD 0070's "secret concentration").
 #
 # Shares an exact multi-architecture Python/trixie manifest with the gateway
 # image. The version-qualified tag keeps the human-readable upstream version;
@@ -25,6 +22,11 @@ FROM ${PYTHON_BASE_IMAGE}
 
 WORKDIR /app
 
+COPY requirements.orchestrator.lock /tmp/requirements.orchestrator.lock
+RUN pip install --no-cache-dir --require-hashes \
+        -r /tmp/requirements.orchestrator.lock \
+    && rm /tmp/requirements.orchestrator.lock
+
 # The orchestrator content. Baked so the image is self-contained (runs from
 # a built image, no runtime bind-mount); the docker backend may still
 # bind-mount /app for dev live-reload, which simply overlays this copy.

bot_bottle/backend/firecracker/infra_artifact.py

@@ -54,6 +54,7 @@ _BUILD_INPUTS = {
         "image-build-args.json",
         "Dockerfile.orchestrator",
         "Dockerfile.orchestrator.fc",
+        "requirements.orchestrator.lock",
     ),
     "gateway": (
         "image-build-args.json",

bot_bottle/orchestrator/__init__.py

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
     from ..gateway import Gateway, GatewayError
     from .lifecycle import Orchestrator
     from .service import OrchestratorCore
-    from .server import OrchestratorServer, dispatch, make_server
+    from .server import OrchestratorServer, create_app, make_server
 
 
 # Facade name -> submodule that defines it. Lazy so importing a leaf (or the
@@ -67,8 +67,8 @@ _LAZY: dict[str, str] = {
     "GatewayError": "..gateway",
     "Orchestrator": ".lifecycle",
     "OrchestratorCore": ".service",
+    "create_app": ".server",
     "OrchestratorServer": ".server",
-    "dispatch": ".server",
     "make_server": ".server",
 }
 
@@ -100,7 +100,7 @@ __all__ = [
     "GatewayError",
     "Orchestrator",
     "OrchestratorCore",
+    "create_app",
     "OrchestratorServer",
-    "dispatch",
     "make_server",
 ]

bot_bottle/orchestrator/__main__.py

@@ -62,17 +62,14 @@ def main(argv: list[str] | None = None) -> int:
     orchestrator = OrchestratorCore(registry, broker, secret)
 
     server = make_server(orchestrator, host=args.host, port=args.port)
-    bound_host, bound_port = server.server_address[0], server.server_address[1]
     log.info(
         "orchestrator control plane listening",
-        context={"host": bound_host, "port": bound_port, "db": str(registry.db_path)},
+        context={"host": args.host, "port": args.port, "db": str(registry.db_path)},
     )
     try:
-        server.serve_forever()
+        server.run()
     except KeyboardInterrupt:
         log.info("orchestrator shutting down")
-    finally:
-        server.server_close()
     return 0
 
 

bot_bottle/orchestrator/api.py

@@ -0,0 +1,339 @@
+"""FastAPI control-plane routes for the orchestrator."""
+# pyright: reportUnusedFunction=false
+
+from __future__ import annotations
+
+import asyncio
+import math
+import sys
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict, StrictStr
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from ..orchestrator_auth import ROLE_CLI, ROLES
+from ..supervisor.types import TOOLS
+from ..trust_domain import CONTROL_PLANE
+from .http_contract import (
+    MAX_BODY_BYTES,
+    ORCHESTRATOR_AUTH_HEADER,
+    REQUEST_BODY_TIMEOUT_SECONDS,
+)
+from .service import OrchestratorCore
+
+_GATEWAY_ROUTES = frozenset({
+    ("POST", "/resolve"),
+    ("POST", "/supervise/propose"),
+    ("POST", "/supervise/poll"),
+})
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(extra="ignore", strict=True)
+
+
+class LaunchBody(_StrictModel):
+    source_ip: StrictStr
+    image_ref: StrictStr = ""
+    metadata: StrictStr = ""
+    policy: StrictStr = ""
+    tokens: dict[StrictStr, StrictStr] = {}
+    env_var_secret: StrictStr = ""
+
+
+class PolicyBody(_StrictModel):
+    policy: StrictStr
+
+
+class ReprovisionBody(_StrictModel):
+    env_var_secret: StrictStr
+
+
+class ReconcileBody(_StrictModel):
+    live_source_ips: list[StrictStr]
+    grace_seconds: float | None = None
+
+
+class IdentityBody(_StrictModel):
+    source_ip: StrictStr
+    identity_token: StrictStr = ""
+
+
+class AttributeBody(_StrictModel):
+    source_ip: StrictStr
+    identity_token: StrictStr
+
+
+class RespondBody(_StrictModel):
+    proposal_id: StrictStr
+    bottle_slug: StrictStr
+    decision: StrictStr
+    notes: StrictStr = ""
+    final_file: StrictStr | None = None
+
+
+class ProposeBody(IdentityBody):
+    tool: StrictStr
+    proposed_file: StrictStr
+    justification: StrictStr
+
+
+class PollBody(IdentityBody):
+    proposal_id: StrictStr
+
+
+class ControlPlaneBoundary:
+    """Reject unauthenticated and oversized requests before reading a body."""
+
+    def __init__(self, app: ASGIApp, signing_key: str) -> None:
+        self.app = app
+        self.signing_key = signing_key
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        method = scope["method"]
+        route = scope["path"].rstrip("/") or "/"
+        if not (method == "GET" and route == "/health"):
+            headers = dict(scope["headers"])
+            presented = headers.get(
+                ORCHESTRATOR_AUTH_HEADER.encode(), b"",
+            ).decode(errors="ignore")
+            role = CONTROL_PLANE.verify(presented, self.signing_key)
+            if role is None:
+                await self._reject(
+                    scope, send, 401, "control-plane authentication required",
+                )
+                return
+            allowed = ROLES if (method, route) in _GATEWAY_ROUTES else {ROLE_CLI}
+            if role not in allowed:
+                await self._reject(scope, send, 403, "insufficient role for this route")
+                return
+            scope.setdefault("state", {})["role"] = role
+        raw_length = dict(scope["headers"]).get(b"content-length")
+        if raw_length is not None:
+            try:
+                length = int(raw_length)
+            except ValueError:
+                await self._reject(scope, send, 400, "invalid Content-Length")
+                return
+            if length < 0:
+                await self._reject(scope, send, 400, "invalid Content-Length")
+                return
+            if length > MAX_BODY_BYTES:
+                await self._reject(scope, send, 413, "request body too large")
+                return
+        try:
+            body = await self._read_body(receive)
+        except _BodyTooLarge:
+            await self._reject(scope, send, 413, "request body too large")
+            return
+        except TimeoutError:
+            await self._reject(scope, send, 408, "request body read timed out")
+            return
+        try:
+            await self.app(scope, self._replay_body(body), send)
+        except Exception as exc:  # noqa: BLE001 - redact control-plane failures
+            sys.stderr.write(
+                f"orchestrator: {method} {route} failed "
+                f"[error_type={type(exc).__name__}]\n"
+            )
+            sys.stderr.flush()
+            await self._reject(scope, send, 500, "internal error")
+
+    @staticmethod
+    async def _reject(
+        scope: Scope, send: Send, status: int, error: str,
+    ) -> None:
+        response = JSONResponse({"error": error}, status_code=status)
+        await response(scope, ControlPlaneBoundary._empty_receive, send)
+
+    @staticmethod
+    async def _empty_receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    @staticmethod
+    async def _read_body(receive: Receive) -> bytes:
+        body = bytearray()
+        async with asyncio.timeout(REQUEST_BODY_TIMEOUT_SECONDS):
+            while True:
+                message = await receive()
+                if message["type"] != "http.request":
+                    break
+                body.extend(message.get("body", b""))
+                if len(body) > MAX_BODY_BYTES:
+                    raise _BodyTooLarge
+                if not message.get("more_body", False):
+                    break
+        return bytes(body)
+
+    @staticmethod
+    def _replay_body(body: bytes) -> Receive:
+        sent = False
+
+        async def replay() -> Message:
+            nonlocal sent
+            if sent:
+                return {"type": "http.disconnect"}
+            sent = True
+            return {"type": "http.request", "body": body, "more_body": False}
+
+        return replay
+
+
+class _BodyTooLarge(Exception):
+    """The streamed request exceeded the control-plane body limit."""
+
+
+def _required(value: str, name: str) -> str:
+    if not value:
+        raise HTTPException(400, f"{name} (string) is required")
+    return value
+
+
+def create_app(orch: OrchestratorCore, *, signing_key: str) -> FastAPI:
+    """Build the authenticated orchestrator ASGI application."""
+    key = signing_key.strip()
+    if not key:
+        raise ValueError(
+            "orchestrator control-plane signing key is required; "
+            "refusing to start without caller authentication"
+        )
+    app = FastAPI(
+        title="bot-bottle orchestrator",
+        docs_url=None,
+        redoc_url=None,
+        openapi_url=None,
+    )
+    app.add_middleware(ControlPlaneBoundary, signing_key=key)
+
+    @app.get("/health")
+    def health() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get("/gateway")
+    def gateway() -> dict[str, object]:
+        return orch.gateway_status()
+
+    @app.get("/bottles")
+    def bottles() -> dict[str, object]:
+        return {"bottles": [record.redacted() for record in orch.registry.all()]}
+
+    @app.post("/bottles", status_code=201)
+    def launch(body: LaunchBody) -> dict[str, str]:
+        rec = orch.launch_bottle(
+            _required(body.source_ip, "source_ip"),
+            image_ref=body.image_ref,
+            metadata=body.metadata,
+            policy=body.policy,
+            tokens=dict(body.tokens),
+            env_var_secret=body.env_var_secret,
+        )
+        return {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
+
+    @app.put("/bottles/{bottle_id}/policy")
+    def set_policy(bottle_id: str, body: PolicyBody) -> dict[str, object]:
+        if orch.set_policy(bottle_id, body.policy):
+            return {"updated": True}
+        raise HTTPException(404, "no such bottle")
+
+    @app.post("/bottles/{bottle_id}/reprovision_gateway")
+    def reprovision(bottle_id: str, body: ReprovisionBody) -> dict[str, object]:
+        secret = _required(body.env_var_secret, "env_var_secret")
+        if orch.reprovision_from_secret(bottle_id, secret):
+            return {"reprovisioned": True}
+        raise HTTPException(404, "no stored secrets for this bottle")
+
+    @app.delete("/bottles/{bottle_id}")
+    def teardown(bottle_id: str) -> dict[str, object]:
+        if orch.teardown_bottle(bottle_id):
+            return {"torn_down": True}
+        raise HTTPException(404, "no such bottle")
+
+    @app.post("/reconcile")
+    def reconcile(body: ReconcileBody) -> dict[str, object]:
+        if any(not ip for ip in body.live_source_ips):
+            raise HTTPException(400, "live_source_ips must contain non-empty strings")
+        kwargs: dict[str, float] = {}
+        if body.grace_seconds is not None:
+            if not math.isfinite(body.grace_seconds) or body.grace_seconds < 0:
+                raise HTTPException(
+                    400, "grace_seconds must be a non-negative finite number",
+                )
+            kwargs["grace_seconds"] = body.grace_seconds
+        return {"reaped": orch.reconcile(body.live_source_ips, **kwargs)}
+
+    @app.post("/attribute")
+    def attribute(body: AttributeBody) -> dict[str, str]:
+        rec = orch.attribute(body.source_ip, body.identity_token)
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return {"bottle_id": rec.bottle_id}
+
+    @app.get("/supervise/proposals")
+    def proposals() -> dict[str, object]:
+        return {"proposals": orch.supervise_pending()}
+
+    @app.post("/supervise/respond")
+    def respond(body: RespondBody) -> dict[str, object]:
+        ok, error = orch.supervise_respond(
+            _required(body.proposal_id, "proposal_id"),
+            bottle_slug=_required(body.bottle_slug, "bottle_slug"),
+            decision=_required(body.decision, "decision"),
+            notes=body.notes,
+            final_file=body.final_file,
+        )
+        if not ok:
+            raise HTTPException(409, error)
+        return {"responded": True}
+
+    @app.post("/supervise/propose", status_code=201)
+    def propose(body: ProposeBody) -> dict[str, str]:
+        source_ip = _required(body.source_ip, "source_ip")
+        if body.tool not in TOOLS:
+            raise HTTPException(400, f"tool (string) must be one of {TOOLS}")
+        rec = orch.resolve(source_ip, body.identity_token)
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        proposal_id = orch.supervise_queue_proposal(
+            rec.bottle_id,
+            tool=body.tool,
+            proposed_file=_required(body.proposed_file, "proposed_file"),
+            justification=_required(body.justification, "justification"),
+        )
+        return {"proposal_id": proposal_id}
+
+    @app.post("/supervise/poll")
+    def poll(body: PollBody) -> dict[str, object]:
+        rec = orch.resolve(
+            _required(body.source_ip, "source_ip"), body.identity_token,
+        )
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return orch.supervise_poll_response(
+            rec.bottle_id, _required(body.proposal_id, "proposal_id"),
+        )
+
+    @app.post("/resolve")
+    def resolve(body: IdentityBody) -> dict[str, object]:
+        rec = orch.resolve(
+            _required(body.source_ip, "source_ip"), body.identity_token,
+        )
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return {
+            "bottle_id": rec.bottle_id,
+            "policy": rec.policy,
+            "tokens": orch.tokens_for(rec.bottle_id),
+        }
+
+    return app
+
+
+__all__ = [
+    "ControlPlaneBoundary",
+    "MAX_BODY_BYTES",
+    "ORCHESTRATOR_AUTH_HEADER",
+    "create_app",
+]

bot_bottle/orchestrator/client.py

@@ -21,7 +21,7 @@ from dataclasses import dataclass
 from ..log import debug
 from ..orchestrator_auth import ROLE_CLI
 from ..trust_domain import CONTROL_PLANE
-from .server import ORCHESTRATOR_AUTH_HEADER
+from .http_contract import ORCHESTRATOR_AUTH_HEADER
 
 DEFAULT_TIMEOUT_SECONDS = 5.0
 

bot_bottle/orchestrator/http_contract.py

@@ -0,0 +1,11 @@
+"""Dependency-free constants shared by orchestrator HTTP clients and server."""
+
+ORCHESTRATOR_AUTH_HEADER = "x-bot-bottle-orchestrator-auth"
+MAX_BODY_BYTES = 1 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 10.0
+
+__all__ = [
+    "MAX_BODY_BYTES",
+    "ORCHESTRATOR_AUTH_HEADER",
+    "REQUEST_BODY_TIMEOUT_SECONDS",
+]

bot_bottle/orchestrator/server.py

@@ -1,502 +1,49 @@
-"""Orchestrator HTTP control plane (PRD 0070).
+"""Uvicorn transport for the FastAPI orchestrator control plane."""
-
-The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
-**HTTP** — the universal transport chosen in 0070 (works on every host; no
-vsock / unix-socket portability caveats):
-
-    GET    /health                    -> 200 {"status": "ok"}
-    GET    /gateway                   -> 200 {"configured", ["name","running"]}
-    GET    /bottles                   -> 200 {"bottles": [ <redacted record>, ...]}
-    POST   /bottles                   -> 201 {"bottle_id","identity_token"}  (launch)
-                                         body: {"source_ip", ["image_ref"],
-                                                ["metadata"], ["policy"],
-                                                ["tokens"], ["env_var_secret"]}
-    PUT    /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404  (live reload)
-                                         body: {"policy"}
-    POST   /bottles/<bottle_id>/reprovision_gateway
-                                      -> 200 {"reprovisioned": true} | 404
-                                         body: {"env_var_secret"}
-    DELETE /bottles/<bottle_id>        -> 200 {"torn_down": true} | 404  (teardown)
-    POST   /reconcile                 -> 200 {"reaped": [bottle_id, ...]}
-                                         body: {"live_source_ips": [...],
-                                                ["grace_seconds"]}
-    POST   /attribute                 -> 200 {"bottle_id"} | 403
-    POST   /resolve                   -> 200 {"bottle_id","policy"} | 403
-                                         body: {"source_ip","identity_token"}
-    GET    /supervise/proposals       -> 200 {"proposals": [ <proposal>, ...]}
-    POST   /supervise/respond         -> 200 {"responded": true} | 409  (operator)
-                                         body: {"proposal_id","bottle_slug",
-                                                "decision", ["notes"],["final_file"]}
-    POST   /supervise/propose         -> 201 {"proposal_id"} | 403  (agent)
-                                         body: {"source_ip","identity_token",
-                                                "tool","proposed_file","justification"}
-    POST   /supervise/poll            -> 200 {"status", ["notes"],["final_file"]} | 403
-                                         body: {"source_ip","identity_token",
-                                                "proposal_id"}
-
-The `/supervise/propose` + `/supervise/poll` pair is the **agent** half of the
-supervise flow: the data plane (supervise / egress / git-gate) queues a proposal
-and polls for its response over RPC instead of opening `bot-bottle.db` directly.
-`poll` is idempotent — it never archives, so a dropped connection can't lose an
-operator decision (the row is reaped when the bottle is torn down / reconciled).
-Both attribute the caller by `(source_ip, identity_token)` exactly like
-`/resolve`, so a bottle can only ever queue or read its own proposals.
-
-`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
-tear down) the bottle in the registry AND broker the backend-native launch
-via the orchestrator. Register/deregister without a launch are internal to
-`OrchestratorCore`, not exposed here.
-
-Routing/handling is the pure function `dispatch()` so it is unit-testable
-without a socket; `Handler` / `OrchestratorServer` / `make_server` are a
-thin stdlib adapter around it. Listing redacts identity tokens — they are
-returned only once, to the caller that launches the bottle.
-"""
 
 from __future__ import annotations
 
-import http.server
-import json
-import math
 import os
 import socket
-import socketserver
-import sys
 import threading
-import typing
-from urllib.parse import urlsplit
 
-from ..orchestrator_auth import ROLE_CLI, ROLES
+import uvicorn
+
 from ..trust_domain import CONTROL_PLANE
-from ..supervisor.types import TOOLS
+from .api import create_app
+from .http_contract import MAX_BODY_BYTES, ORCHESTRATOR_AUTH_HEADER
 from .service import OrchestratorCore
 
-# JSON body payload type (parsed request / rendered response).
+MAX_REQUESTS = 32
-Json = dict[str, object]
+KEEP_ALIVE_TIMEOUT_SECONDS = 10
-
-# The request header carrying the caller's role-scoped control-plane token (a
-# signed JWT naming the caller's role — see orchestrator_auth). The role gates which
-# routes the caller may reach: the data plane holds a `gateway` token good only
-# for the agent-facing lookups; the host CLI holds a `cli` token for the
-# operator/mutating routes. An agent that can merely *reach* the port holds no
-# token at all, and a compromised gateway holds only `gateway` — neither can
-# drive the operator routes (approve proposals, rewrite policy, read tokens).
-ORCHESTRATOR_AUTH_HEADER = "x-bot-bottle-orchestrator-auth"
-MAX_BODY_BYTES = 1 * 1024 * 1024
-REQUEST_TIMEOUT_SECONDS = 10.0
-MAX_REQUEST_THREADS = 32
-
-# The routes the data plane (role `gateway`) is allowed to reach — exactly the
-# per-request lookups PolicyResolver makes. Every other authenticated route is
-# operator-only. `cli` is a superset role: it may reach any route.
-_GATEWAY_ROUTES: frozenset[tuple[str, str]] = frozenset({
-    ("POST", "/resolve"),
-    ("POST", "/supervise/propose"),
-    ("POST", "/supervise/poll"),
-})
 
 
-def _allowed_roles(method: str, route: str) -> frozenset[str]:
+class OrchestratorServer:
-    """The roles permitted on `(method, route)`: `gateway` or `cli` on the
+    """Small lifecycle wrapper around Uvicorn with an eagerly bound socket."""
-    data-plane routes, `cli`-only everywhere else."""
-    if (method, route) in _GATEWAY_ROUTES:
-        return ROLES
-    return frozenset({ROLE_CLI})
 
+    def __init__(self, config: uvicorn.Config) -> None:
+        self._server = uvicorn.Server(config)
+        self._stopped = threading.Event()
+        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind((config.host, config.port))
+        self._socket.listen(config.backlog)
+        self.server_address = self._socket.getsockname()
 
-def _parse_json_object(body: bytes) -> Json:
+    def run(self) -> None:
-    """Parse a JSON object body. Raises ValueError for non-objects / bad JSON."""
-    if not body:
-        return {}
-    obj = json.loads(body)  # raises json.JSONDecodeError (a ValueError)
-    if not isinstance(obj, dict):
-        raise ValueError("request body must be a JSON object")
-    return obj
-
-
-def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
-    orch: OrchestratorCore, method: str, path: str, body: bytes, *, role: str | None = ROLE_CLI,
-) -> tuple[int, Json]:
-    """Route one control-plane request to a (status, payload) pair. Pure —
-    no I/O beyond the orchestrator — so it is fully testable without a socket.
-
-    `role` is the caller's verified control-plane role (`gateway` or `cli`), or
-    None for an unauthenticated request. Every route except `GET /health`
-    requires a role: a missing role is 401, and a role that
-    doesn't cover the route is 403 — so a `gateway` data-plane token can reach
-    `/resolve` + `/supervise/{propose,poll}` but not the operator routes
-    (rewrite policy, read injected tokens, approve its own supervise proposals).
-    The source-IP + identity-token checks inside `/resolve` and `/attribute`
-    authenticate the *bottle* a request is about, not the *caller*, so this role
-    gate is what protects the caller-privileged routes. Defaults `cli` so unit
-    tests of the routing logic don't have to thread it through."""
-    route = urlsplit(path).path.rstrip("/") or "/"
-
-    if method == "GET" and route == "/health":
-        return 200, {"status": "ok"}
-
-    # Role gate — every route below is a trusted-caller operation. Deny before
-    # touching the registry / broker / supervise store.
-    if role is None:
-        return 401, {"error": "control-plane authentication required"}
-    if role not in _allowed_roles(method, route):
-        return 403, {"error": "insufficient role for this route"}
-
-    if method == "GET" and route == "/gateway":
-        return 200, orch.gateway_status()
-
-    if method == "GET" and route == "/bottles":
-        return 200, {"bottles": [r.redacted() for r in orch.registry.all()]}
-
-    if method == "POST" and route == "/bottles":
         try:
-            data = _parse_json_object(body)
+            self._server.run(sockets=[self._socket])
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        image_ref = data.get("image_ref")
-        metadata = data.get("metadata")
-        policy = data.get("policy")
-        raw_tokens = data.get("tokens")
-        tokens = {
-            k: v for k, v in raw_tokens.items() if isinstance(k, str) and isinstance(v, str)
-        } if isinstance(raw_tokens, dict) else {}
-        env_var_secret = data.get("env_var_secret", "")
-        rec = orch.launch_bottle(
-            source_ip,
-            image_ref=image_ref if isinstance(image_ref, str) else "",
-            metadata=metadata if isinstance(metadata, str) else "",
-            policy=policy if isinstance(policy, str) else "",
-            tokens=tokens,
-            env_var_secret=env_var_secret if isinstance(env_var_secret, str) else "",
-        )
-        return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
-
-    if method == "PUT" and route.startswith("/bottles/") and route.endswith("/policy"):
-        bottle_id = route[len("/bottles/"):-len("/policy")]
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        policy = data.get("policy")
-        if not isinstance(policy, str):
-            return 400, {"error": "policy (string) is required"}
-        if orch.set_policy(bottle_id, policy):
-            return 200, {"updated": True}
-        return 404, {"error": "no such bottle"}
-
-    if (
-        method == "POST"
-        and route.startswith("/bottles/")
-        and route.endswith("/reprovision_gateway")
-    ):
-        bottle_id = route[len("/bottles/") : -len("/reprovision_gateway")]
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        env_var_secret = data.get("env_var_secret")
-        if not isinstance(env_var_secret, str) or not env_var_secret:
-            return 400, {"error": "env_var_secret (string) is required"}
-        if orch.reprovision_from_secret(bottle_id, env_var_secret):
-            return 200, {"reprovisioned": True}
-        return 404, {"error": "no stored secrets for this bottle"}
-
-    if method == "DELETE" and route.startswith("/bottles/"):
-        bottle_id = route[len("/bottles/"):]
-        if orch.teardown_bottle(bottle_id):
-            return 200, {"torn_down": True}
-        return 404, {"error": "no such bottle"}
-
-    if method == "POST" and route == "/reconcile":
-        # Host-driven self-heal: the caller enumerates its live bottles (only
-        # the host can see the backend) and the orchestrator drops rows for
-        # every other active bottle. Trusted-caller only — an agent that could
-        # reach this would be able to unregister its neighbours.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        raw_ips = data.get("live_source_ips")
-        if not isinstance(raw_ips, list):
-            return 400, {"error": "live_source_ips (list of strings) is required"}
-        if any(not isinstance(ip, str) or not ip for ip in raw_ips):
-            return 400, {"error": "live_source_ips must contain non-empty strings"}
-        live = raw_ips
-        grace = data.get("grace_seconds")
-        kwargs: dict[str, float] = {}
-        if grace is not None:
-            if isinstance(grace, bool) or not isinstance(grace, (int, float)):
-                return 400, {"error": "grace_seconds must be a non-negative finite number"}
-            parsed_grace = float(grace)
-            if not math.isfinite(parsed_grace) or parsed_grace < 0:
-                return 400, {"error": "grace_seconds must be a non-negative finite number"}
-            kwargs["grace_seconds"] = parsed_grace
-        return 200, {"reaped": orch.reconcile(live, **kwargs)}
-
-    if method == "POST" and route == "/attribute":
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        if not isinstance(source_ip, str) or not isinstance(token, str):
-            return 400, {"error": "source_ip and identity_token (strings) required"}
-        rec = orch.attribute(source_ip, token)
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        return 200, {"bottle_id": rec.bottle_id}
-
-    if method == "GET" and route == "/supervise/proposals":
-        # Operator TUI: pending supervise proposals across all bottles.
-        return 200, {"proposals": orch.supervise_pending()}
-
-    if method == "POST" and route == "/supervise/respond":
-        # Operator decision: apply (approve/modify rewrites egress policy),
-        # write the queued response, audit — all server-side on the one DB.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        proposal_id = data.get("proposal_id")
-        bottle_slug = data.get("bottle_slug")
-        decision = data.get("decision")
-        if not (isinstance(proposal_id, str) and proposal_id):
-            return 400, {"error": "proposal_id (string) is required"}
-        if not (isinstance(bottle_slug, str) and bottle_slug):
-            return 400, {"error": "bottle_slug (string) is required"}
-        if not (isinstance(decision, str) and decision):
-            return 400, {"error": "decision (string) is required"}
-        notes = data.get("notes")
-        final_file = data.get("final_file")
-        ok, err = orch.supervise_respond(
-            proposal_id,
-            bottle_slug=bottle_slug,
-            decision=decision,
-            notes=notes if isinstance(notes, str) else "",
-            final_file=final_file if isinstance(final_file, str) else None,
-        )
-        if ok:
-            return 200, {"responded": True}
-        return 409, {"error": err}
-
-    if method == "POST" and route == "/supervise/propose":
-        # Agent half: queue a proposal, attributed to the caller resolved from
-        # (source_ip, identity_token) — never a caller-supplied slug — so the
-        # data plane can't forge attribution. Fail-closed 403 when unattributed.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        tool = data.get("tool")
-        proposed_file = data.get("proposed_file")
-        justification = data.get("justification")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        if not isinstance(tool, str) or tool not in TOOLS:
-            return 400, {"error": f"tool (string) must be one of {TOOLS}"}
-        if not isinstance(proposed_file, str) or not proposed_file:
-            return 400, {"error": "proposed_file (string) is required"}
-        if not isinstance(justification, str) or not justification:
-            return 400, {"error": "justification (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        proposal_id = orch.supervise_queue_proposal(
-            rec.bottle_id, tool=tool, proposed_file=proposed_file,
-            justification=justification,
-        )
-        return 201, {"proposal_id": proposal_id}
-
-    if method == "POST" and route == "/supervise/poll":
-        # Agent half: non-blocking read of the caller's own proposal decision.
-        # Attributed like /propose, and scoped to the resolved bottle id, so a
-        # guessed proposal_id can never read another bottle's response.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        proposal_id = data.get("proposal_id")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        if not isinstance(proposal_id, str) or not proposal_id:
-            return 400, {"error": "proposal_id (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        return 200, orch.supervise_poll_response(rec.bottle_id, proposal_id)
-
-    if method == "POST" and route == "/resolve":
-        # The per-request lookup the multi-tenant gateway makes: returns the
-        # bottle's policy. Requires a matching (source_ip, identity_token)
-        # pair — a missing/empty/mismatched token fail-closes (403), no
-        # source-IP-only fallback.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        # tokens are the in-memory per-bottle egress auth values the gateway
-        # injects; served here, never persisted.
-        return 200, {
-            "bottle_id": rec.bottle_id,
-            "policy": rec.policy,
-            "tokens": orch.tokens_for(rec.bottle_id),
-        }
-
-    return 404, {"error": "not found"}
-
-
-class Handler(http.server.BaseHTTPRequestHandler):
-    """Thin stdlib adapter: read the body, call `dispatch`, write JSON."""
-
-    # Quiet by default (the orchestrator has its own logging); opt back into
-    # stdlib access logging with BOT_BOTTLE_ORCHESTRATOR_DEBUG.
-    def log_message(self, format: str, *args: typing.Any) -> None:  # noqa: A002
-        if os.environ.get("BOT_BOTTLE_ORCHESTRATOR_DEBUG"):
-            super().log_message(format, *args)
-
-    def _serve(self, method: str) -> None:
-        """Read the request body, dispatch it, and write the JSON reply. A
-        dispatch failure (e.g. a broker error) returns a 500 rather than
-        crashing the connection, so one bad request can't take the control
-        plane down for the caller."""
-        server = self.server
-        assert isinstance(server, OrchestratorServer)
-        role = server.role_for(self.headers.get(ORCHESTRATOR_AUTH_HEADER, ""))
-        route = urlsplit(self.path).path.rstrip("/") or "/"
-        if not (method == "GET" and route == "/health") and role is None:
-            self._write_json(
-                401, {"error": "control-plane authentication required"},
-            )
-            return
-        length_header = self.headers.get("Content-Length")
-        try:
-            length = int(length_header) if length_header is not None else 0
-        except ValueError:
-            self._write_json(400, {"error": "invalid Content-Length"})
-            return
-        if length < 0:
-            self._write_json(400, {"error": "invalid Content-Length"})
-            return
-        if length > MAX_BODY_BYTES:
-            self._write_json(413, {"error": "request body too large"})
-            return
-        try:
-            body = self.rfile.read(length) if length else b""
-        except (TimeoutError, socket.timeout):
-            self._write_json(408, {"error": "request body read timed out"})
-            return
-        try:
-            status: int
-            payload: Json
-            status, payload = dispatch(
-                server.orchestrator, method, self.path, body, role=role)
-        except Exception as e:  # noqa: BLE001 — the control plane must stay up
-            # Do not echo exception messages to the caller or logs: broker and
-            # persistence exceptions can contain request data. The operation,
-            # route, and exception type are enough to correlate a traceback.
-            sys.stderr.write(
-                f"orchestrator: {method} {self.path} failed "
-                f"[error_type={type(e).__name__}]\n"
-            )
-            sys.stderr.flush()
-            status, payload = 500, {"error": "internal error"}
-        self._write_json(status, payload)
-
-    def _write_json(self, status: int, payload: Json) -> None:
-        data = json.dumps(payload).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(data)))
-        self.end_headers()
-        self.wfile.write(data)
-
-    def do_GET(self) -> None:
-        self._serve("GET")
-
-    def do_POST(self) -> None:
-        self._serve("POST")
-
-    def do_PUT(self) -> None:
-        self._serve("PUT")
-
-    def do_DELETE(self) -> None:
-        self._serve("DELETE")
-
-
-class OrchestratorServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
-    """Threading HTTP server that carries the orchestrator for its handlers.
-
-    Holds the per-host control-plane *signing key* (from
-    `$BOT_BOTTLE_ORCHESTRATOR_TOKEN`, injected by the launcher into the
-    orchestrator process only) and verifies each request's role-scoped token
-    against it. Every route but `/health` requires a valid token whose role
-    covers the route. Construction fails when the key is absent so a new or
-    misconfigured launcher cannot accidentally expose an open control plane."""
-
-    daemon_threads = True
-    allow_reuse_address = True
-
-    def __init__(
-        self,
-        address: tuple[str, int],
-        orchestrator: OrchestratorCore,
-        *,
-        signing_key: str,
-    ) -> None:
-        self.orchestrator = orchestrator
-        self._signing_key = signing_key.strip()
-        if not self._signing_key:
-            raise ValueError(
-                "orchestrator control-plane signing key is required; "
-                "refusing to start without caller authentication"
-            )
-        self._request_slots = threading.BoundedSemaphore(MAX_REQUEST_THREADS)
-        super().__init__(address, Handler)
-
-    def get_request(self) -> tuple[socket.socket, typing.Any]:
-        request, client_address = super().get_request()
-        request.settimeout(REQUEST_TIMEOUT_SECONDS)
-        return request, client_address
-
-    def process_request(
-        self, request: typing.Any, client_address: typing.Any,
-    ) -> None:
-        # Bound concurrency before ThreadingMixIn creates a worker. Backpressure
-        # stays in the accept loop instead of allocating an unbounded thread per
-        # slow or malicious connection.
-        self._request_slots.acquire()
-        try:
-            super().process_request(request, client_address)
-        except BaseException:
-            self._request_slots.release()
-            raise
-
-    def process_request_thread(
-        self, request: typing.Any, client_address: typing.Any,
-    ) -> None:
-        try:
-            super().process_request_thread(request, client_address)
         finally:
-            self._request_slots.release()
+            self._stopped.set()
 
-    def role_for(self, presented: str) -> str | None:
+    def serve_forever(self) -> None:
-        """The verified caller role, or None for a missing/invalid token."""
+        self.run()
-        return CONTROL_PLANE.verify(presented, self._signing_key)
+
+    def shutdown(self) -> None:
+        self._server.should_exit = True
+        self._stopped.wait(timeout=5)
+
+    def server_close(self) -> None:
+        self._socket.close()
 
 
 def make_server(
@@ -506,18 +53,28 @@ def make_server(
     *,
     signing_key: str | None = None,
 ) -> OrchestratorServer:
-    """Build an authenticated control-plane server.
+    """Build a bounded Uvicorn server around the orchestrator application."""
-
-    ``signing_key=None`` reads the owning process's injected environment.
-    Empty or missing keys are rejected by :class:`OrchestratorServer`.
-    """
     key = CONTROL_PLANE.key_from_env() if signing_key is None else signing_key
-    return OrchestratorServer(
+    app = create_app(orchestrator, signing_key=key)
-        (host, port), orchestrator, signing_key=key,
+    config = uvicorn.Config(
+        app,
+        host=host,
+        port=port,
+        access_log=bool(os.environ.get("BOT_BOTTLE_ORCHESTRATOR_DEBUG")),
+        log_level="info",
+        limit_concurrency=MAX_REQUESTS,
+        timeout_keep_alive=KEEP_ALIVE_TIMEOUT_SECONDS,
+        server_header=False,
     )
+    return OrchestratorServer(config)
 
 
 __all__ = [
-    "dispatch", "Handler", "OrchestratorServer", "make_server", "Json",
+    "KEEP_ALIVE_TIMEOUT_SECONDS",
-    "ORCHESTRATOR_AUTH_HEADER", "MAX_BODY_BYTES",
+    "MAX_BODY_BYTES",
+    "MAX_REQUESTS",
+    "ORCHESTRATOR_AUTH_HEADER",
+    "OrchestratorServer",
+    "create_app",
+    "make_server",
 ]

bot_bottle/resources.py

@@ -44,6 +44,7 @@ BUNDLED_RESOURCES: tuple[str, ...] = (
     "Dockerfile.orchestrator",
     "Dockerfile.orchestrator.fc",
     "requirements.gateway.lock",
+    "requirements.orchestrator.lock",
     "nix/firecracker-netpool.nix",
     "scripts/firecracker-netpool.sh",
 )

requirements-dev.txt

@@ -2,9 +2,12 @@
 # The bot-bottle project itself has no runtime dependencies.
 # These tools are used for code quality checks in CI/CD.
 
+-r requirements.orchestrator.in
 pylint>=3.0.0
 pyright>=1.1.411
 coverage>=7.0.0
 # PEP 517 build front-end used by tests/unit/test_wheel_install.py to build and
 # install a real wheel (proves the installed distribution is self-contained).
 build>=1.0.0
+# FastAPI's in-process TestClient transport.
+httpx>=0.28.0

requirements.orchestrator.in

@@ -0,0 +1,3 @@
+# Runtime dependencies baked only into the orchestrator images.
+fastapi==0.140.0
+uvicorn==0.51.0

requirements.orchestrator.lock

@@ -0,0 +1,182 @@
+#
+# This file is autogenerated by pip-compile with Python 3.13
+# by the following command:
+#
+#    pip-compile --generate-hashes --output-file=requirements.orchestrator.lock requirements.orchestrator.in
+#
+annotated-doc==0.0.4 \
+    --hash=sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320 \
+    --hash=sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4
+    # via fastapi
+annotated-types==0.8.0 \
+    --hash=sha256:13b2beaad985e05e2d6407ee4c4f35590b11f8d693a258a561055cac8f64cab7 \
+    --hash=sha256:f072f4d804ea359e4eaf198b1af7a8b0943881a87f31bb764f8bf219bb9419e0
+    # via pydantic
+anyio==4.14.2 \
+    --hash=sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494 \
+    --hash=sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f
+    # via starlette
+click==8.4.2 \
+    --hash=sha256:9a6cea6e60b17ebe0a44c5cc636d94f09bd66142c1cd7d8b4cd731c4917a15f6 \
+    --hash=sha256:e6f9f66136c816745b9d65817da91d61d957fb16e02e4dcd0552553c5a197b76
+    # via uvicorn
+fastapi==0.140.0 \
+    --hash=sha256:e951c0a0d9540bf5d9a2a9e078fd415da2ab7e312d435139e7d9e2e7fe9f0b23 \
+    --hash=sha256:f338951b82fd74ca8f843163aec43ea1a1ce84d515415a50fa98fa25572a5544
+    # via -r requirements.orchestrator.in
+h11==0.16.0 \
+    --hash=sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1 \
+    --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86
+    # via uvicorn
+idna==3.18 \
+    --hash=sha256:7f952cbe720b688055e3f87de14f5c3e5fdaa8bc3928985c4077ca689de849a2 \
+    --hash=sha256:ffb385a7e039654cef1ab9ef32c6fafe283c0c0467bba1d9029738ce4a14a848
+    # via anyio
+pydantic==2.13.4 \
+    --hash=sha256:45a282cde31d808236fd7ea9d919b128653c8b38b393d1c4ab335c62924d9aba \
+    --hash=sha256:c40756b57adaa8b1efeeced5c196f3f3b7c435f90e84ea7f443901bec8099ef6
+    # via fastapi
+pydantic-core==2.46.4 \
+    --hash=sha256:00c603d540afdd6b80eb39f078f33ebd46211f02f33e34a32d9f053bba711de0 \
+    --hash=sha256:0186750b482eefa11d7f435892b09c5c606193ef3375bcf94aa00ae6bfb66262 \
+    --hash=sha256:041bde0a48fd37cf71cab1c9d56d3e8625a3793fef1f7dd232b3ff37e978ecda \
+    --hash=sha256:0c563b08bca408dc7f65f700633d8442fffb2421fc47b8101377e9fd65051ff0 \
+    --hash=sha256:0cbe8b01f948de4286c74cdd6c667aceb38f5c1e26f0693b3983d9d74887c65e \
+    --hash=sha256:0ce40cd7b21210e99342afafbd4d0f76d784eb5b1d60f3bdc566be4983c6c73b \
+    --hash=sha256:0e96592440881c74a213e5ad528e2b24d3d4f940de2766bed9010ab1d9e51594 \
+    --hash=sha256:10e17cbb10a330363733efc4d7c4d0dd827ac0909b8f6a6542298fed1ea62f29 \
+    --hash=sha256:133878133d271ade3d41d1bfb2a45ec38dbdbda40bc065921c6b04e4630127e2 \
+    --hash=sha256:14d4edf427bdcf950a8a02d7cb44a08614388dd6e1bdcbf4f67504fa7887da9c \
+    --hash=sha256:14f4c5d6db102bd796a627bbb3a17b4cf4574b9ae861d8b7c9a9661c6dd3362d \
+    --hash=sha256:17299feefe090f2caa5b8e37222bb5f663e4935a8bfa6931d4102e5df1a9f398 \
+    --hash=sha256:184c081504d17f1c1066e430e117142b2c77d9448a97f7b65c6ac9fd9aee238d \
+    --hash=sha256:18e5ceec2ab67e6d5f1a9085e5a24c9c4e2ac4545730bfe668680bca05e555f3 \
+    --hash=sha256:19e51f073cd3df251856a8a4189fbdf1de4012c3ebacfb1884f94f1eb406079f \
+    --hash=sha256:1a7dd0b3ee80d90150e3495a3a13ac34dbcbfd4f012996a6a1d8900e91b5c0fb \
+    --hash=sha256:1d8ba486450b14f3b1d63bc521d410ec7565e52f887b9fb671791886436a42f7 \
+    --hash=sha256:2108ba5c1c1eca18030634489dc544844144ee36357f2f9f780b93e7ddbb44b5 \
+    --hash=sha256:228ee9bae8bef5b1e97ec58302f80357c37199e0d0a99174e138d28e6957b9d9 \
+    --hash=sha256:23ace664830ee0bfe014a0c7bc248b1f7f25ed7ad103852c317624a1083af462 \
+    --hash=sha256:2412e734dcb48da14d4e4006b82b46b74f2518b8a26ee7e58c6844a6cd6d03c4 \
+    --hash=sha256:29c61fc04a3d840155ff08e475a04809278972fe6aef51e2720554e96367e34b \
+    --hash=sha256:2f84c03c8607173d16b5a854ec68a2f9079ae03237a54fb506d13af47e1d018d \
+    --hash=sha256:3009f12e4e90b7f88b4f9adb1b0c4a3d58fe7820f3238c190047209d148026df \
+    --hash=sha256:3245406455a5d98187ec35530fd772b1d799b26667980872c8d4614991e2c4a2 \
+    --hash=sha256:3447661d99f75a3683a4cf5c87da72f2161964611864dbbeac7fbb118bb4bfc0 \
+    --hash=sha256:372429a130e469c9cd698925ce5fc50940b7a1336b0d82038e63d5bbc4edc519 \
+    --hash=sha256:395aebd9183f9d112f569aeb5b2214d1a10a33bec8456447f7fbdfa51d38d4cd \
+    --hash=sha256:3a233125ac121aa3ffba9a2b59edfc4a985a76092dc8279586ab4b71390875e7 \
+    --hash=sha256:3be77f45df024d789a672ae34f8b06fb346c4f9f46ea714956660ea4862e89ac \
+    --hash=sha256:3bf92c5d0e00fefaab325a4d27828fe6b6e2a21848686b5b60d2d9eeb09d76c6 \
+    --hash=sha256:3ecbc122d18468d06ca279dc26a8c2e2d5acb10943bb35e36ae92096dc3b5565 \
+    --hash=sha256:3fb702cd90b0446a3a1c5e470bfa0dd23c0233b676a9099ddcc964fa6ca13898 \
+    --hash=sha256:428e04521a40150c85216fc8b85e8d39fece235a9cf5e383761238c7fa9b96fb \
+    --hash=sha256:432c179df7874eeb73307aad2df0755e1ae0efa61ff0ea89b93e194411ae3928 \
+    --hash=sha256:4a05d69cba51d852c5c3e92758653245a50c0b646ced0cf05bd793ed592839d6 \
+    --hash=sha256:4c63ebc82684aa89d9a3bcbd13d515b3be44250dc68dd3bd81526c1cb31286c3 \
+    --hash=sha256:4fc73cb559bdb54b1134a706a2802a4cddd27a0633f5abb7e53056268751ac6a \
+    --hash=sha256:4fcbe087dbc2068af7eda3aa87634eba216dbda64d1ae73c8684b621d33f6596 \
+    --hash=sha256:56cb4851bcaf3d117eddcef4fe66afd750a50274b0da8e22be256d10e5611987 \
+    --hash=sha256:5855698a4856556d86e8e6cd8434bc3ac0314ee8e12089ae0e143f64c6256e4e \
+    --hash=sha256:5a4330cdbc57162e4b3aa303f588ba752257694c9c9be3e7ebb11b4aca659b5d \
+    --hash=sha256:5b712b53160b79a5850310b912a5ef8e57e56947c8ad690c227f5c9d7e561712 \
+    --hash=sha256:5d5902252db0d3cedf8d4a1bc68f70eeb430f7e4c7104c8c476753519b423008 \
+    --hash=sha256:617d7e2ca7dcb8c5cf6bcb8c59b8832c94b36196bbf1cbd1bfb56ed341905edd \
+    --hash=sha256:62f875393d7f270851f20523dd2e29f082bcc82292d66db2b64ea71f64b6e1c1 \
+    --hash=sha256:633147d34cf4550417f12e2b1a0383973bdf5cdfde212cb09e9a581cf10820be \
+    --hash=sha256:66ce7632c22d837c95301830e111ad0128a32b8207533b60896a96c4915192ea \
+    --hash=sha256:6b3ace8194b0e5204818c92802dcdca7fc6d88aabbb799d7c795540d9cd6d292 \
+    --hash=sha256:6f2eeda33a839975441c86a4119e1383c50b47faf0cbb5176985565c6bb02c33 \
+    --hash=sha256:7027560ee92211647d0d34e3f7cd6f50da56399d26a9c8ad0da286d3869a53f3 \
+    --hash=sha256:7283d57845ecf5a163403eb0702dfc220cc4fbdd18919cb5ccea4f95ee1cdab4 \
+    --hash=sha256:7a5f930472650a82629163023e630d160863fce524c616f4e5186e5de9d9a49b \
+    --hash=sha256:7bfb192b3f4b9e8a89b6277b6ce787564f62cfd272055f6e685726b111dc7826 \
+    --hash=sha256:811ff8e9c313ab425368bcbb36e5c4ebd7108c2bbf4e4089cfbb0b01eff63fac \
+    --hash=sha256:8233f2947cf85404441fd7e0085f53b10c93e0ee78611099b5c7237e36aacbf7 \
+    --hash=sha256:82cf5301172168103724d49a1444d3378cb20cdee30b116a1bd6031236298a5d \
+    --hash=sha256:8358a950c8909158e3df31538a7e4edc2d7265a7c54b47f0864d9e5bae9dcebf \
+    --hash=sha256:85bb3611ff1802f3ee7fdd7dbff26b56f343fb432d57a4728fdd49b6ef35e2f4 \
+    --hash=sha256:86e1a4418c6cd97d60c95c71164158eaf7324fae7b0923264016baa993eba6fc \
+    --hash=sha256:8b9bab013d1c7a79d3501ff86d0bc9c31bf587db4551677b96bec07df78c6b15 \
+    --hash=sha256:8c5dac79fa1614d1e06ca695109c6105923bd9c7d1d6c918d4e637b7e6b32fd3 \
+    --hash=sha256:8d0820e8192167f80d88d64038e609c31452eeca865b4e1d9950a27a4609b00b \
+    --hash=sha256:8daafc69c93ee8a0204506a3b6b30f586ef54028f52aeeeb5c4cfc5184fd5914 \
+    --hash=sha256:9037063db01f09b09e237c282b6792bd4da634b5402c4e7f0c61effed7701a04 \
+    --hash=sha256:905a0ed8ea6f2d61c1738835f99b699348d7857379083e5fc497fa0c967a407c \
+    --hash=sha256:90884113d8b48f760e9587002789ddd741e76ab9f89518cd1e43b1f1a52ec44b \
+    --hash=sha256:91a06d2e259ecfbd8c901d70c3c507900458498142b3026a296b7de4d1322cc9 \
+    --hash=sha256:926c9541b14b12b1681dca8a0b75feb510b06c6341b70a8e500c2fdcff837cce \
+    --hash=sha256:9401557acd873c3a7f3eb9383edef8ac4968f9510e340f4808d427e75667e7b4 \
+    --hash=sha256:9551187363ffc0de2a00b2e47c25aeaeb1020b69b668762966df15fc5659dd5a \
+    --hash=sha256:962ccbab7b642487b1d8b7df90ef677e03134cf1fd8880bf698649b22a69371f \
+    --hash=sha256:97e7cf2be5c77b7d1a9713a05605d49460d02c6078d38d8bef3cbe323c548424 \
+    --hash=sha256:9aa768456404a8bf48a4406685ac2bec8e72b62c69313734fa3b73cf33b3a894 \
+    --hash=sha256:9bc519fbf2b7578398853d815009ae5e4d4603d12f4e3f91da8c06852d3da3e9 \
+    --hash=sha256:9d56801be94b86a9da183e5f3766e6310752b99ff647e38b09a9500d88e46e76 \
+    --hash=sha256:9f444c499b3eefd3a92e348059471ea0c3a6e303d9c1cec09fa748fd9f895201 \
+    --hash=sha256:9fa8ae11da9e2b3126c6426f147e0fba88d96d65921799bb30c6abd1cb2c97fb \
+    --hash=sha256:a0f62d0a58f4e7da165457e995725421e0064f2255d8eccebc49f41bbc23b109 \
+    --hash=sha256:a396dcc17e5a0b164dbe026896245a4fa9ff402edca1dff0be3d53a517f74de4 \
+    --hash=sha256:aaa2a54443eff1950ba5ddc6b6ccda0d9c84a364276a62f969bdf2a390650848 \
+    --hash=sha256:ad785e92e6dc634c21555edc8bd6b64957ab844541bcb96a1366c202951ae526 \
+    --hash=sha256:af8244b2bef6aaad6d92cda81372de7f8c8d36c9f0c3ea36e827c60e7d9467a0 \
+    --hash=sha256:b078afbc25f3a1436c7a1d2cd3e322497ee99615ba97c563566fdf46aff1ee01 \
+    --hash=sha256:b2f69dec1725e79a012d920df1707de5caf7ed5e08f3be4435e25803efc47458 \
+    --hash=sha256:b8458003118a712e66286df6a707db01c52c0f52f7db8e4a38f0da1d3b94fc4e \
+    --hash=sha256:bb63e0198ca18aad131c089b9204c23079c3afa95487e561f4c522d519e55aba \
+    --hash=sha256:bfec22eab3c8cc2ceec0248aec886624116dc079afa027ecc8ad4a7e62010f8a \
+    --hash=sha256:c1747f85cee84c26985853c6f3d9bd3e75da5212912443fa111c113b9c246f39 \
+    --hash=sha256:c1b3f518abeca3aa13c712fd202306e145abf59a18b094a6bafb2d2bbf59192c \
+    --hash=sha256:c50f2528cf200c5eed56faf3f4e22fcd5f38c157a8b78576e6ba3168ec35f000 \
+    --hash=sha256:c68fcd102d71ea85c5b2dfac3f4f8476eff42a9e078fd5faefff6d145063536b \
+    --hash=sha256:c7a7bd4e39e8e4c12c39cd480356842b6a8a06e41b23a55a5e3e191718838ddf \
+    --hash=sha256:c94f0688e7b8d0a67abf40e57a7eaaecd17cc9586706a31b76c031f63df052b4 \
+    --hash=sha256:cbaf13819775b7f769bf4a1f066cb6df7a28d4480081a589828ef190226881cd \
+    --hash=sha256:cd2213145bcc2ba85884d0ac63d222fece9209678f77b9b4d76f054c561adb28 \
+    --hash=sha256:ce5c1d2a8b27468f433ca974829c44060b8097eedc39933e3c206a90ee49c4a9 \
+    --hash=sha256:d396ec2b979760aaf3218e76c24e65bd0aca24983298653b3a9d7a45f9e47b30 \
+    --hash=sha256:d51026d73fcfd93610abc7b27789c26b313920fcfb20e27462d74a7f8b06e983 \
+    --hash=sha256:d80ee3d731373b24cebbc10d689ca4ee1875caf0d5703a245db18efd4dd37fc1 \
+    --hash=sha256:d995260fdf4e1db774581b4900e0f832abe3c7c84996726bbc161b19c8f29e76 \
+    --hash=sha256:da4b951fe36dc7c3a1ccb4e3cd1747c3542b8c9ceede8fc86cae054e764485f5 \
+    --hash=sha256:daa27d92c36f24388fe3ad306b174781c747627f134452e4f128ea00ce1fe8c4 \
+    --hash=sha256:db06ffe51636ffe9ca531fe9023dd64bdd794be8754cb5df57c5498ae5b518a7 \
+    --hash=sha256:e0d65b8c354be7fb5f720c3caa8bc940bc2d20ce749c8e06135f07f8ed95dd7c \
+    --hash=sha256:e68b7a074f65a2fd746c52a7ce6142ab7006074ac269ace0c25cd8ba171f8066 \
+    --hash=sha256:e739fee756ba1010f8bcccb534252e85a35fe45ae92c295a06059ce58b74ccd3 \
+    --hash=sha256:e846ae7835bf0703ae43f534ab79a867146dadd59dc9ca5c8b53d5c8f7c9ef02 \
+    --hash=sha256:e9c26f834c65f5752f3f06cb08cb86a913ceb7274d0db6e267808a708b46bc89 \
+    --hash=sha256:ea793e075b70290d89d8142074262885d3f7da19634845135751bd6344f73b50 \
+    --hash=sha256:f027324c56cd5406ca49c124b0db10e56c69064fec039acc571c29020cc87c76 \
+    --hash=sha256:f13a646d65d09fbf1bc6b3a9635d30095c8e7e5cc419ff35ecc563c5fd04cd49 \
+    --hash=sha256:f47286a97f0bc9b8859519809077b91b2cefe4ae47fcbf5e466a009c1c5d742b \
+    --hash=sha256:f747929cf940cddb5b3668a390056ddd5ba2e5010615ea2dcf4f9c4f3ab8791d \
+    --hash=sha256:f99626688942fb746e545232e7726926f3be91b5975f8b55327665fafda991c7 \
+    --hash=sha256:f9fa868638bf362d3d138ea55829cefb3d5f4b0d7f142234382a15e2485dbec4 \
+    --hash=sha256:fbdb89b3e1c94a30cc5edfce477c6e6a5dc4d8f84665b455c27582f211a1c72c \
+    --hash=sha256:fc010ab034c8c7452522748bf937df58020d256ccae0874463d1f4d01758af8e \
+    --hash=sha256:fc3e9034a63de20e15e8ade85358bc6efc614008cab72898b4b4952bea0509ff \
+    --hash=sha256:fd8b3d9fd264be37976686c7f65cd52a83f5e84f4bfd2adf9c1d469676bbb6ae
+    # via pydantic
+starlette==1.3.1 \
+    --hash=sha256:05d0213193f2fbaae60e2ecb593b4add4262ad4e46536b54abe36f11a71724e0 \
+    --hash=sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6
+    # via fastapi
+typing-extensions==4.16.0 \
+    --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 \
+    --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5
+    # via
+    #   fastapi
+    #   pydantic
+    #   pydantic-core
+    #   typing-inspection
+typing-inspection==0.4.2 \
+    --hash=sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7 \
+    --hash=sha256:ba561c48a67c5958007083d386c3295464928b01faa735ab8547c5692e87f464
+    # via
+    #   fastapi
+    #   pydantic
+uvicorn==0.51.0 \
+    --hash=sha256:5d38af6cd620f2ae3849fb44fd4879e0890aa1febe8d47eb355fb45d93fe6a5b \
+    --hash=sha256:f6f4b69b657c312f516dd2d268ab9ae6f254b11e4bac504f37b2ab58b24dd0b0
+    # via -r requirements.orchestrator.in

setup.py

@@ -26,6 +26,7 @@ _BUNDLED_RESOURCES = (
     "Dockerfile.orchestrator",
     "Dockerfile.orchestrator.fc",
     "requirements.gateway.lock",
+    "requirements.orchestrator.lock",
     "nix/firecracker-netpool.nix",
     "scripts/firecracker-netpool.sh",
 )

tests/unit/test_infra_artifact.py

@@ -114,6 +114,7 @@ class TestVersionInputs(unittest.TestCase):
             '{"PYTHON_BASE_IMAGE": "python:pinned"}\n',
         )
         (root / "requirements.gateway.lock").write_text("mitmproxy==11.1.3\n")
+        (root / "requirements.orchestrator.lock").write_text("fastapi==0.140.0\n")
         (root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
 
     def test_pyproject_toml_change_bumps_version(self) -> None:
@@ -165,6 +166,21 @@ class TestVersionInputs(unittest.TestCase):
             )
         self.assertNotEqual(before, after)
 
+    def test_orchestrator_lock_change_bumps_orchestrator_version(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            self._fake_repo(root)
+            before = ia.infra_artifact_version(
+                "init", "orchestrator", repo_root=root,
+            )
+            (root / "requirements.orchestrator.lock").write_text(
+                "fastapi==0.140.0 --hash=sha256:changed\n",
+            )
+            after = ia.infra_artifact_version(
+                "init", "orchestrator", repo_root=root,
+            )
+        self.assertNotEqual(before, after)
+
     def test_base_image_argument_change_bumps_both_role_versions(self) -> None:
         with tempfile.TemporaryDirectory() as d:
             root = Path(d)

tests/unit/test_orchestrator_client.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import json
+import subprocess
+import sys
 import unittest
 import urllib.error
 from unittest.mock import MagicMock, patch
@@ -20,6 +22,30 @@ from bot_bottle.orchestrator.client import (
 _URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
 
 
+class TestImportBoundary(unittest.TestCase):
+    def test_host_client_does_not_import_server_dependencies(self) -> None:
+        script = """
+import importlib.abc
+import sys
+
+class BlockServerDependencies(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.split(".", 1)[0] in {"fastapi", "uvicorn"}:
+            raise ImportError(f"host import reached {fullname}")
+        return None
+
+sys.meta_path.insert(0, BlockServerDependencies())
+import bot_bottle.orchestrator.client
+"""
+        result = subprocess.run(
+            [sys.executable, "-c", script],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        self.assertEqual(0, result.returncode, result.stderr)
+
+
 class TestHostAuthToken(unittest.TestCase):
     def test_mints_a_cli_token_from_the_host_key(self) -> None:
         # The CLI mints its `cli` token from the control-plane trust domain's

tests/unit/test_orchestrator_server.py

@@ -6,6 +6,7 @@ server tests), plus one real-socket round-trip to prove the handler wiring.
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import http.client
 import io
@@ -17,13 +18,17 @@ import threading
 import unittest
 import urllib.error
 import urllib.request
+import httpx
 from contextlib import closing
+from collections.abc import Iterator
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY, mint
+from bot_bottle.orchestrator import api as orchestrator_api
 from bot_bottle.orchestrator.broker import StubBroker
-from bot_bottle.orchestrator.server import MAX_BODY_BYTES, dispatch, make_server
+from bot_bottle.orchestrator.http_contract import ORCHESTRATOR_AUTH_HEADER
+from bot_bottle.orchestrator.server import MAX_BODY_BYTES, create_app, make_server
 from bot_bottle.orchestrator.store.registry_store import BottleRecord, RegistryStore
 from bot_bottle.orchestrator.service import OrchestratorCore
 from bot_bottle.orchestrator.store.store_manager import StoreManager
@@ -38,6 +43,47 @@ def _body(obj: object) -> bytes:
     return json.dumps(obj).encode()
 
 
+def dispatch(
+    orchestrator: OrchestratorCore,
+    method: str,
+    path: str,
+    body: bytes,
+    *,
+    role: str | None = ROLE_CLI,
+) -> tuple[int, dict[str, object]]:
+    """Exercise the real ASGI application without a network socket."""
+    key = "in-process-dispatch-key"
+    headers = {"content-type": "application/json"}
+    if role is not None:
+        headers[ORCHESTRATOR_AUTH_HEADER] = mint(role, key)
+
+    async def request() -> httpx.Response:
+        transport = httpx.ASGITransport(app=create_app(orchestrator, signing_key=key))
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="http://orchestrator",
+            follow_redirects=True,
+        ) as client:
+            return await client.request(
+                method, path, content=body, headers=headers,
+            )
+
+    response = asyncio.run(request())
+    payload = response.json()
+    if response.status_code == 422:
+        detail = payload.get("detail", []) if isinstance(payload, dict) else []
+        field = ""
+        if isinstance(detail, list) and detail and isinstance(detail[0], dict):
+            location = detail[0].get("loc", ())
+            if isinstance(location, (list, tuple)) and len(location) > 1:
+                field = str(location[1])
+        suffix = f": {field}" if field else ""
+        return 400, {"error": f"invalid request body{suffix}"}
+    if isinstance(payload, dict) and "detail" in payload and "error" not in payload:
+        payload = {"error": payload["detail"]}
+    return response.status_code, payload
+
+
 def _orchestrator(db_path: Path) -> OrchestratorCore:
     store = RegistryStore(db_path)
     store.migrate()
@@ -361,6 +407,76 @@ class TestServerRoundTrip(unittest.TestCase):
         self.assertNotIn("SENSITIVE", output)
 
 
+class TestControlPlaneBoundary(unittest.IsolatedAsyncioTestCase):
+    @staticmethod
+    def _scope(key: str) -> dict[str, object]:
+        return {
+            "type": "http",
+            "asgi": {"version": "3.0"},
+            "http_version": "1.1",
+            "method": "POST",
+            "scheme": "http",
+            "path": "/bottles",
+            "raw_path": b"/bottles",
+            "query_string": b"",
+            "headers": [(
+                ORCHESTRATOR_AUTH_HEADER.encode(),
+                mint(ROLE_CLI, key).encode(),
+            )],
+            "client": ("127.0.0.1", 1),
+            "server": ("127.0.0.1", 80),
+            "state": {},
+        }
+
+    async def test_chunked_oversized_body_returns_413(self) -> None:
+        key = "stream-limit-key"
+        called = False
+        first: dict[str, object] = {
+            "type": "http.request",
+            "body": b"x" * MAX_BODY_BYTES,
+            "more_body": True,
+        }
+        last: dict[str, object] = {
+            "type": "http.request", "body": b"x", "more_body": False,
+        }
+        chunks: Iterator[dict[str, object]] = iter([first, last])
+        sent: list[dict[str, object]] = []
+
+        async def receive() -> dict[str, object]:
+            return next(chunks)
+
+        async def send(message: dict[str, object]) -> None:
+            sent.append(message)
+
+        async def inner(*_args: object) -> None:
+            nonlocal called
+            called = True
+
+        boundary = orchestrator_api.ControlPlaneBoundary(inner, key)
+        await boundary(self._scope(key), receive, send)  # type: ignore[arg-type]
+        self.assertFalse(called)
+        self.assertEqual(413, sent[0]["status"])
+
+    async def test_slow_stream_returns_408(self) -> None:
+        key = "stream-timeout-key"
+        sent: list[dict[str, object]] = []
+
+        async def receive() -> dict[str, object]:
+            await asyncio.sleep(1)
+            return {"type": "http.request", "body": b"", "more_body": False}
+
+        async def send(message: dict[str, object]) -> None:
+            sent.append(message)
+
+        async def inner(*_args: object) -> None:
+            self.fail("timed-out body reached the application")
+
+        boundary = orchestrator_api.ControlPlaneBoundary(inner, key)
+        with patch.object(orchestrator_api, "REQUEST_BODY_TIMEOUT_SECONDS", 0.01):
+            await boundary(self._scope(key), receive, send)  # type: ignore[arg-type]
+        self.assertEqual(408, sent[0]["status"])
+
+
 class TestOrchestratorAuth(unittest.TestCase):
     """Role-scoped control-plane tokens (issue #400 / #469 review): every route
     but /health needs a valid token, and the token's role gates which routes it

tests/unit/test_wheel_install.py

@@ -62,7 +62,10 @@ class TestWheelInstall(unittest.TestCase):
 
         # Installing the freshly-built wheel must succeed — fail if it doesn't.
         install = subprocess.run(
-            [str(cls.venv_py), "-m", "pip", "install", "--quiet", str(wheels[0])],
+            [
+                str(cls.venv_py), "-m", "pip", "install", "--quiet",
+                "--force-reinstall", "--no-deps", str(wheels[0]),
+            ],
             capture_output=True, text=True, check=False,
         )
         if install.returncode != 0: