fix(orchestrator): address review on host control server transport (#468)

Codex review on #496:

- **High — ambiguous delivery no longer orphans a launched bottle.** A
  timeout / dropped response from the host controller is now the ambiguous
  BrokerUnavailableError (distinct from the definite BrokerAuthError /
  BrokerClientError). OrchestratorCore.launch_bottle keeps the registry
  row on the ambiguous case instead of deregistering — deregistering would
  orphan a running container with no record (reconcile reaps rows, never
  containers). The row is left for reconcile to reap iff the bottle is not
  actually live. Definite failures still roll back, so a real failure
  leaves no orphan row.
- **Medium — the privileged endpoint bounds request bodies.** The host
  server rejects an oversized Content-Length with 413 before reading it,
  and sets a per-request socket timeout, so a caller that can merely reach
  the socket (no signed token) can't exhaust memory or a handler thread.

Tests: ambiguous-keep vs definite-rollback in the launch path; the
BrokerUnavailableError/BrokerClientError split in BrokerClient; the 413
body cap + handler error paths (driven in-thread, since daemon request
threads lose coverage) plus a deterministic real-socket check that
declares an oversized Content-Length but sends a sliver (rejection on the
header, no unread-body reset race); and the __main__ entrypoint broker
selection. Diff-coverage 98%; pyright clean; pylint 9.8.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

bot_bottle/orchestrator/broker.py

@@ -38,7 +38,21 @@ _ALLOWED_OPS = ("launch", "teardown")
 class BrokerAuthError(Exception):
     """A broker request failed provenance or schema verification —
     bad/absent signature, malformed token, or a payload that doesn't match
-    the fixed launch-request shape. Fail-closed: the broker must not act."""
+    the fixed launch-request shape. Fail-closed: the broker must not act.
+
+    A **definite** negative: nothing was launched, so a caller may safely roll
+    back as if the op never happened."""
+
+
+class BrokerUnavailableError(Exception):
+    """A brokered request could not be carried to a verdict: the broker (or the
+    wire to it) was unreachable, timed out, or dropped the response.
+
+    Crucially **ambiguous** — unlike `BrokerAuthError`, the op MAY already have
+    taken effect on the backend before the response was lost, so a caller must
+    NOT assume it did nothing (e.g. must not roll a registry row back as if no
+    launch happened, which would orphan a running container). Only the in-process
+    brokers never raise this; the out-of-process `BrokerClient` does."""
 
 
 @dataclass(frozen=True)
@@ -179,6 +193,7 @@ class StubBroker(LaunchBroker):
 
 __all__ = [
     "BrokerAuthError",
+    "BrokerUnavailableError",
     "LaunchRequest",
     "SubmitBroker",
     "LaunchBroker",

bot_bottle/orchestrator/broker_client.py

@@ -20,15 +20,17 @@ import json
 import urllib.error
 import urllib.request
 
-from .broker import BrokerAuthError, LaunchRequest
+from .broker import BrokerAuthError, BrokerUnavailableError, LaunchRequest
 
 DEFAULT_TIMEOUT_SECONDS = 5.0
 
 
 class BrokerClientError(RuntimeError):
-    """A brokered launch/teardown could not be delivered to the host control
-    server: unreachable, or an unexpected status other than the fail-closed 401
-    (which surfaces as `BrokerAuthError`, matching the in-process broker)."""
+    """The host control server *responded*, but with an unexpected status other
+    than the fail-closed 401 (which surfaces as `BrokerAuthError`) — e.g. a 502
+    backend failure or a malformed body. A definite negative: the host processed
+    the request and it did not launch. (A *no-response* failure — unreachable /
+    timeout / dropped — is the ambiguous `BrokerUnavailableError` instead.)"""
 
 
 class BrokerClient:
@@ -45,9 +47,11 @@ class BrokerClient:
         verified and acted on.
 
         Raises `BrokerAuthError` on a fail-closed 401 (bad provenance/schema —
-        the same exception the in-process broker raises), or `BrokerClientError`
-        if the controller is unreachable, times out, or returns any other
-        non-success status."""
+        the same exception the in-process broker raises); `BrokerClientError` if
+        the host *responds* with any other non-success status or a malformed
+        body (a definite negative); or `BrokerUnavailableError` if no response is
+        obtained (unreachable / timeout / dropped) — the **ambiguous** case, where
+        the host may already have acted, so the caller must not roll back."""
         data = json.dumps({"token": token}).encode()
         req = urllib.request.Request(
             f"{self._base}/broker", data=data, method="POST",
@@ -65,8 +69,11 @@ class BrokerClient:
             raise BrokerClientError(
                 f"POST /broker: HTTP {e.code} {detail}".rstrip()
             ) from e
-        except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
-            raise BrokerClientError(f"POST /broker: {e}") from e
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            # No usable response — unreachable, timed out, or the connection
+            # dropped mid-exchange. Ambiguous: the request may already have
+            # launched the bottle, so this is NOT a definite failure.
+            raise BrokerUnavailableError(f"POST /broker: {e}") from e
 
 
 def _json_object(raw: bytes) -> dict[str, object]:

bot_bottle/orchestrator/host_server.py

@@ -57,6 +57,16 @@ BROKER_SECRET_ENV = "BOT_BOTTLE_BROKER_SECRET"
 # (8099) — a separate privileged component listening on its own socket.
 DEFAULT_PORT = 8091
 
+# Cap on the request body. A signed broker request is tiny, so rejecting anything
+# larger *before reading it* keeps a caller that can merely reach the socket (no
+# signed token needed) from exhausting memory or a handler thread with a huge
+# Content-Length — the signed token, not mere reachability, is the authority.
+MAX_BODY_BYTES = 64 * 1024
+
+# Per-request socket timeout, bounding how long a stalled / slow-loris caller can
+# hold a handler thread on this privileged listener.
+REQUEST_TIMEOUT_SECONDS = 15
+
 
 def _parse_json_object(body: bytes) -> Json:
     """Parse a JSON object body. Raises ValueError for non-objects / bad JSON."""
@@ -130,6 +140,10 @@ def dispatch(  # pylint: disable=too-many-return-statements
 class Handler(http.server.BaseHTTPRequestHandler):
     """Thin stdlib adapter: read the body, call `dispatch`, write JSON."""
 
+    # Socket timeout per request (applied by StreamRequestHandler.setup) so a
+    # stalled caller can't pin a handler thread on this privileged listener.
+    timeout = REQUEST_TIMEOUT_SECONDS
+
     # Quiet by default; opt back into stdlib access logging with
     # BOT_BOTTLE_HOST_CONTROLLER_DEBUG (the controller has its own logging).
     def log_message(self, format: str, *args: typing.Any) -> None:  # noqa: A002
@@ -137,12 +151,21 @@ class Handler(http.server.BaseHTTPRequestHandler):
             super().log_message(format, *args)
 
     def _serve(self, method: str) -> None:
-        """Read the request body, dispatch it, and write the JSON reply. A
-        dispatch that raises (it shouldn't — dispatch is total) still returns a
-        500 rather than dropping the connection."""
+        """Read the request body (bounded), dispatch it, and write the JSON
+        reply. A dispatch that raises (it shouldn't — dispatch is total) still
+        returns a 500 rather than dropping the connection."""
         server = self.server
         assert isinstance(server, HostControlServer)
-        length = int(self.headers.get("Content-Length") or 0)
+        try:
+            length = int(self.headers.get("Content-Length") or 0)
+        except ValueError:
+            self._reply(400, {"error": "invalid Content-Length"})
+            return
+        if length < 0 or length > MAX_BODY_BYTES:
+            # Reject before reading: nothing legitimate is this big, so an
+            # oversized declared length is a bug or a resource-exhaustion attempt.
+            self._reply(413, {"error": "request body too large"})
+            return
         body = self.rfile.read(length) if length > 0 else b""
         try:
             status, payload = dispatch(server.broker, method, self.path, body)
@@ -150,6 +173,10 @@ class Handler(http.server.BaseHTTPRequestHandler):
             sys.stderr.write(f"host controller: {method} {self.path} failed: {e!r}\n")
             sys.stderr.flush()
             status, payload = 500, {"error": f"internal error: {e}"}
+        self._reply(status, payload)
+
+    def _reply(self, status: int, payload: typing.Mapping[str, object]) -> None:
+        """Write one JSON response with an explicit Content-Length."""
         data = json.dumps(payload).encode()
         self.send_response(status)
         self.send_header("Content-Type", "application/json")

bot_bottle/orchestrator/service.py

@@ -25,7 +25,7 @@ import json
 from collections.abc import Iterable
 from datetime import datetime, timezone
 
-from .broker import LaunchRequest, SubmitBroker, sign_request
+from .broker import BrokerUnavailableError, LaunchRequest, SubmitBroker, sign_request
 from .store.registry_store import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore
 from .supervisor import (
     AuditEntry,
@@ -111,14 +111,23 @@ class OrchestratorCore:
             image_ref=image_ref,
             slot=slot,
         )
-        launched = False
         try:
             self._broker.submit(sign_request(req, self._secret))
-            launched = True
-        finally:
-            if not launched:
-                self.registry.deregister(rec.bottle_id)
-                self._tokens.pop(rec.bottle_id, None)
+        except BrokerUnavailableError:
+            # Ambiguous delivery failure (timeout / dropped response): the broker
+            # may already have launched the bottle before the response was lost.
+            # Do NOT deregister — that would orphan a running container with no
+            # registry row (reconcile reaps rows, never containers). Keep the row
+            # so reconcile reaps it iff the bottle is not actually live; surface
+            # the error so the caller knows the launch is unconfirmed.
+            raise
+        except Exception:
+            # A definite failure — a fail-closed rejection, a backend launch
+            # error, or the host reporting it did not launch: nothing is running,
+            # so roll the registry entry back to leave no orphan.
+            self.registry.deregister(rec.bottle_id)
+            self._tokens.pop(rec.bottle_id, None)
+            raise
         return rec
 
     def teardown_bottle(self, bottle_id: str) -> bool: