didericis/bot-bottle
1
0
Fork 0
Code Issues 26 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

feat(orchestrator): slice 13d(i) — containerize the orchestrator (validated on docker)

Browse Source 
Real-on-host testing surfaced two issues the unit-mocked slices couldn't:

1. The host (NixOS) firewall DROPS container->host traffic, so a host-process
   orchestrator is unreachable from the gateway container. Fix: run the
   orchestrator AS a container on the shared gateway network (PRD 0070's
   "virtualize the orchestrator") — the gateway reaches it by container name
   over docker DNS (container<->container, no firewall), and the host CLI
   reaches it via a published loopback port.
2. The control plane crashed the connection on a dispatch error (e.g. a
   broker failure) instead of returning 500.

Changes:
- lifecycle: OrchestratorProcess (host process) -> OrchestratorService
  (containers). Runs the control plane in the bundle image with the repo
  bind-mounted (orchestrator is stdlib-only), register-only stub broker so it
  needs NO docker socket (the backend launches agents; the host manages both
  containers). Registry DB persists via a host-root mount. ensure_running is
  an idempotent singleton over both containers.
- gateway: BOT_BOTTLE_ORCHESTRATOR_URL is now the orchestrator's *by-name*
  URL on the shared network (dropped the host.docker.internal hack).
- control_plane: _serve wraps dispatch — a failure returns 500, never crashes
  the connection.
- OrchestratorProcess default broker -> stub (register-only) for docker.

Validated live end-to-end: both containers up, gateway->orchestrator by name
OK, register -> resolve-by-source-IP returns the bottle's policy from inside
the gateway.

pyright 0 errors; pylint 9.83/10; unit suite green (1760 tests; the 13
test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
This commit is contained in:
didericis
2026-07-13 23:07:49 -04:00
parent 610c4173a5
commit 0c2d0aca63
7 changed files with 185 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_consolidated_launch.py
+3 -3
View File
@@ -40,13 +40,13 @@ def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
class TestLaunchConsolidated(unittest.TestCase):
def _run(self, client: Mock, provision: Mock | None = None):
process = MagicMock()
process.ensure_running.return_value = "http://orch:8080"
service = MagicMock()
service.ensure_running.return_value = "http://orch:8080"
with patch(f"{_MOD}._network_cidr", return_value="172.18.0.0/16"), \
patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()):
return launch_consolidated(_egress_plan(), _git_plan(), process=process)
return launch_consolidated(_egress_plan(), _git_plan(), service=service)
def test_allocates_ip_registers_and_provisions(self) -> None:
client = _client()
tests/unit/test_orchestrator_lifecycle.py
+45 -45
View File
@@ -1,4 +1,4 @@
"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070)."""
"""Unit: orchestrator+gateway container lifecycle — idempotent singleton (PRD 0070)."""
from __future__ import annotations
@@ -6,81 +6,81 @@ import tempfile
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, Mock, patch
from bot_bottle.orchestrator.lifecycle import (
OrchestratorProcess,
ORCHESTRATOR_NAME,
OrchestratorService,
OrchestratorStartError,
)
from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen"
_RUN = "bot_bottle.orchestrator.lifecycle.run_docker"
_GATEWAY = "bot_bottle.orchestrator.lifecycle.DockerGateway"
_SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
_MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
def _health(status: int) -> MagicMock:
"""A urlopen() context-manager whose `.status` is `status`."""
m = MagicMock()
m.__enter__.return_value.status = status
return m
class TestOrchestratorProcess(unittest.TestCase):
class TestOrchestratorService(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.p = OrchestratorProcess(port=8099)
self.svc = OrchestratorService(port=8099)
def test_url(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.p.url)
def test_urls(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.svc.url)
# The gateway reaches the control plane by container name over docker DNS.
self.assertEqual(f"http://{ORCHESTRATOR_NAME}:8099", self.svc.internal_url)
def test_is_healthy_true_on_200(self) -> None:
def test_is_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.p.is_healthy())
def test_is_healthy_false_on_error(self) -> None:
self.assertTrue(self.svc.is_healthy())
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.p.is_healthy())
self.assertFalse(self.svc.is_healthy())
def test_ensure_running_noop_when_already_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)), patch(_POPEN) as popen:
self.assertEqual(self.p.url, self.p.ensure_running())
popen.assert_not_called() # a live control plane is left untouched
def test_ensure_running_healthy_still_ensures_gateway_but_not_orchestrator(self) -> None:
with patch(_URLOPEN, return_value=_health(200)), \
patch(_GATEWAY) as gw_cls, patch(_RUN) as run:
self.assertEqual(self.svc.url, self.svc.ensure_running())
gw = gw_cls.return_value
gw.ensure_running.assert_called() # gateway kept up
run.assert_not_called() # no orchestrator container run
def test_ensure_running_spawns_then_waits_for_health(self) -> None:
# First check (before spawn) fails; after spawn the poll succeeds.
def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
run = Mock(return_value=Mock(returncode=0, stderr=""))
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_POPEN) as popen, patch(_SLEEP):
url = self.p.ensure_running()
self.assertEqual(self.p.url, url)
popen.assert_called_once()
def test_ensure_running_raises_on_startup_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_POPEN), patch(_SLEEP), \
patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.p.ensure_running(startup_timeout=1.0)
def test_argv_includes_gateway_and_broker(self) -> None:
argv = OrchestratorProcess(port=8099, broker="docker", gateway=True)._argv()
self.assertIn("--gateway", argv)
patch(_GATEWAY), patch(_RUN, run), patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
argv = runs[0]
self.assertIn(ORCHESTRATOR_NAME, argv)
self.assertIn("--broker", argv)
self.assertEqual("stub", argv[argv.index("--broker") + 1]) # register-only, no socket
self.assertIn("bot_bottle.orchestrator", argv)
self.assertEqual("docker", argv[argv.index("--broker") + 1])
self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
def test_argv_omits_gateway_when_disabled(self) -> None:
self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv())
def test_ensure_running_raises_on_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_GATEWAY), patch(_RUN, return_value=Mock(returncode=0, stderr="")), \
patch(_SLEEP), patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.svc.ensure_running(startup_timeout=1.0)
def test_spawn_launches_detached_and_logs(self) -> None:
with patch(_POPEN) as popen:
self.p._spawn()
popen.assert_called_once()
kwargs = popen.call_args.kwargs
self.assertTrue(kwargs["start_new_session"]) # outlives the CLI
self.assertTrue((Path(self._tmp.name) / "orchestrator.log").exists())
def test_stop_removes_orchestrator_and_gateway(self) -> None:
with patch(_RUN) as run, patch(_GATEWAY) as gw_cls:
self.svc.stop()
rms = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "rm", "--force"]]
self.assertTrue(any(ORCHESTRATOR_NAME in a for a in rms))
gw_cls.return_value.stop.assert_called_once()
if __name__ == "__main__":