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_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]: