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>

tests/unit/test_orchestrator_broker_client.py

@@ -8,7 +8,11 @@ import unittest
 import urllib.error
 from unittest.mock import MagicMock, patch
 
-from bot_bottle.orchestrator.broker import BrokerAuthError, LaunchRequest
+from bot_bottle.orchestrator.broker import (
+    BrokerAuthError,
+    BrokerUnavailableError,
+    LaunchRequest,
+)
 from bot_bottle.orchestrator.broker_client import BrokerClient, BrokerClientError
 
 _URLOPEN = "bot_bottle.orchestrator.broker_client.urllib.request.urlopen"
@@ -58,14 +62,25 @@ class TestSubmit(unittest.TestCase):
             with self.assertRaises(BrokerAuthError):
                 self.c.submit("forged")
 
-    def test_502_raises_broker_client_error(self) -> None:
+    def test_502_is_a_definite_client_error(self) -> None:
+        # The host responded — it processed the request and did not launch, so a
+        # definite BrokerClientError (the caller may safely roll back).
         with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})):
             with self.assertRaises(BrokerClientError):
                 self.c.submit("tok")
 
-    def test_unreachable_raises_broker_client_error(self) -> None:
+    def test_unreachable_is_ambiguous_unavailable(self) -> None:
+        # No response at all — the request may already have launched, so the
+        # AMBIGUOUS BrokerUnavailableError (the caller must NOT roll back).
         with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
-            with self.assertRaises(BrokerClientError):
+            with self.assertRaises(BrokerUnavailableError):
+                self.c.submit("tok")
+
+    def test_timeout_is_ambiguous_unavailable(self) -> None:
+        # A dropped/late response after the request was sent is the exact orphan
+        # risk: the host may have launched. Must be ambiguous, not a definite fail.
+        with patch(_URLOPEN, side_effect=TimeoutError("read timed out")):
+            with self.assertRaises(BrokerUnavailableError):
                 self.c.submit("tok")
 
     def test_malformed_success_body_raises(self) -> None:
@@ -79,6 +94,24 @@ class TestSubmit(unittest.TestCase):
             with self.assertRaises(BrokerAuthError):
                 self.c.submit("forged")
 
+    def test_non_json_success_body_raises(self) -> None:
+        # A 200 whose body isn't JSON is tolerated into {} then fails the
+        # missing-field check — a definite client error, not a crash.
+        m = MagicMock()
+        m.__enter__.return_value.read.return_value = b"not json at all"
+        with patch(_URLOPEN, return_value=m):
+            with self.assertRaises(BrokerClientError):
+                self.c.submit("tok")
+
+    def test_unreadable_error_body_is_tolerated(self) -> None:
+        # An HTTPError whose body can't be read (fp=None) still classifies by
+        # status — the error detail is best-effort.
+        err = urllib.error.HTTPError(
+            "http://host/broker", 502, "err", {}, None)  # type: ignore[arg-type]
+        with patch(_URLOPEN, side_effect=err):
+            with self.assertRaises(BrokerClientError):
+                self.c.submit("tok")
+
 
 if __name__ == "__main__":
     unittest.main()