From 96f75599a1d0bf284f206178b7b85dcc6d7acaca Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 13 Jul 2026 23:07:49 -0400 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20slice=2013d(i)=20?= =?UTF-8?q?=E2=80=94=20containerize=20the=20orchestrator=20(validated=20on?= =?UTF-8?q?=20docker)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck --- .../backend/docker/consolidated_launch.py | 8 +- bot_bottle/orchestrator/__main__.py | 8 +- bot_bottle/orchestrator/control_plane.py | 13 +- bot_bottle/orchestrator/gateway.py | 16 +- bot_bottle/orchestrator/lifecycle.py | 184 ++++++++++-------- tests/unit/test_consolidated_launch.py | 6 +- tests/unit/test_orchestrator_lifecycle.py | 90 ++++----- 7 files changed, 185 insertions(+), 140 deletions(-) diff --git a/bot_bottle/backend/docker/consolidated_launch.py b/bot_bottle/backend/docker/consolidated_launch.py index 21f8b5e..634a8dc 100644 --- a/bot_bottle/backend/docker/consolidated_launch.py +++ b/bot_bottle/backend/docker/consolidated_launch.py @@ -26,7 +26,7 @@ from ...egress import EgressPlan from ...git_gate import GitGatePlan from ...orchestrator.client import OrchestratorClient from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK -from ...orchestrator.lifecycle import OrchestratorProcess +from ...orchestrator.lifecycle import OrchestratorService from ...orchestrator.registration import registration_inputs from .gateway_net import next_free_ip from .gateway_provision import deprovision_git_gate, provision_git_gate @@ -92,7 +92,7 @@ def launch_consolidated( git_gate_plan: GitGatePlan, *, image_ref: str = "", - process: OrchestratorProcess | None = None, + service: OrchestratorService | None = None, gateway_name: str = GATEWAY_NAME, network: str = GATEWAY_NETWORK, ) -> LaunchContext: @@ -100,8 +100,8 @@ def launch_consolidated( bottle, and provision its git-gate state. Returns the agent's attach context. Raises `ConsolidatedLaunchError` (or the primitives' own errors) if any step fails — the caller tears down on failure.""" - process = process or OrchestratorProcess() - url = process.ensure_running() + service = service or OrchestratorService() + url = service.ensure_running() client = OrchestratorClient(url) cidr = _network_cidr(network) diff --git a/bot_bottle/orchestrator/__main__.py b/bot_bottle/orchestrator/__main__.py index b5b49a3..9a8d15d 100644 --- a/bot_bottle/orchestrator/__main__.py +++ b/bot_bottle/orchestrator/__main__.py @@ -51,7 +51,13 @@ def main(argv: list[str] | None = None) -> int: # anything; 'docker' runs real containers (firecracker drops in later). secret = secrets.token_bytes(32) broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret) - gateway: Gateway | None = DockerGateway() if args.gateway else None + # Standalone `--gateway` (not the consolidated flow, where the host + # lifecycle runs the gateway). The gateway resolves against this same + # process; the URL is only reachable when they share a docker network. + gateway: Gateway | None = ( + DockerGateway(orchestrator_url=f"http://{args.host}:{args.port}") + if args.gateway else None + ) orchestrator = Orchestrator(registry, broker, secret, gateway) # One persistent per-host gateway, shared by every bottle: build the diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index ca0b869..a7459c5 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -34,6 +34,7 @@ import http.server import json import os import socketserver +import sys import typing from urllib.parse import urlsplit @@ -151,12 +152,20 @@ 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.""" + """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, ControlPlaneServer) length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) if length > 0 else b"" - status, payload = dispatch(server.orchestrator, method, self.path, body) + try: + status, payload = dispatch(server.orchestrator, method, self.path, body) + except Exception as e: # noqa: BLE001 — the control plane must stay up + sys.stderr.write(f"orchestrator: {method} {self.path} failed: {e!r}\n") + sys.stderr.flush() + status, payload = 500, {"error": f"internal error: {e}"} data = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") diff --git a/bot_bottle/orchestrator/gateway.py b/bot_bottle/orchestrator/gateway.py index 0b7bb50..573938d 100644 --- a/bot_bottle/orchestrator/gateway.py +++ b/bot_bottle/orchestrator/gateway.py @@ -91,12 +91,17 @@ class DockerGateway(Gateway): *, name: str = GATEWAY_NAME, network: str = GATEWAY_NETWORK, + orchestrator_url: str = "", build_context: Path | None = None, dockerfile: str | None = GATEWAY_DOCKERFILE, ) -> None: self.image_ref = image_ref self.name = name self.network = network + # The control-plane URL the gateway's data plane resolves per bottle + # against — reached by container name over docker DNS on the shared + # network (container↔container, no host firewall). Empty → single-tenant. + self._orchestrator_url = orchestrator_url self._build_context = build_context or _REPO_ROOT self._dockerfile = dockerfile @@ -146,7 +151,7 @@ class DockerGateway(Gateway): # Clear any stale (stopped) container holding the fixed name, then # start fresh. `rm --force` on an absent name is a tolerated no-op. run_docker(["docker", "rm", "--force", self.name]) - proc = run_docker([ + argv = [ "docker", "run", "--detach", "--name", self.name, "--label", GATEWAY_LABEL, @@ -154,8 +159,13 @@ class DockerGateway(Gateway): # Persist the self-generated CA so it survives restarts (agents # trust it) — see GATEWAY_CA_VOLUME. "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}", - self.image_ref, - ]) + ] + if self._orchestrator_url: + # Makes the gateway's egress / git / supervise daemons multi-tenant: + # each request resolves source-IP -> policy against the control plane. + argv += ["--env", f"BOT_BOTTLE_ORCHESTRATOR_URL={self._orchestrator_url}"] + argv.append(self.image_ref) + proc = run_docker(argv) if proc.returncode != 0: raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}") diff --git a/bot_bottle/orchestrator/lifecycle.py b/bot_bottle/orchestrator/lifecycle.py index 405b99a..0fffbf5 100644 --- a/bot_bottle/orchestrator/lifecycle.py +++ b/bot_bottle/orchestrator/lifecycle.py @@ -1,91 +1,141 @@ -"""Orchestrator process lifecycle (PRD 0070, docker slice). +"""Orchestrator + gateway lifecycle (PRD 0070, docker slice). -Before the CLI can register or launch bottles against the consolidated -model, exactly one orchestrator control plane — and the single per-host -gateway it manages — must be running. This starts the orchestrator -dev-harness (`python -m bot_bottle.orchestrator`) as a background host -process and health-checks it. +Runs the orchestrator control plane **as a container** on the shared gateway +network, alongside the gateway container. This is the PRD's "virtualize the +orchestrator": container↔container between the gateway and the orchestrator +avoids the host firewall (which drops container→host traffic), and the gateway +reaches the control plane by container name over docker DNS. The host CLI +reaches it via a published loopback port. -It is an **idempotent singleton**: `ensure_running` returns immediately if a -healthy control plane already answers on the port, and otherwise spawns one -and waits for it to come up. The control-plane port is the singleton key — -a second orchestrator can't bind it, so a stray double-start fails fast -rather than forking a rival. - -Host-process (not container) on purpose: the PRD sequences the orchestrator -as a plain-process dev-harness first (fast iteration, and it already has the -host user's docker access to broker launches), while the data-plane -*gateway* it manages runs as a container. Wrapping the orchestrator itself -in a backend-native unit is a later step. +The orchestrator runs with the **register-only broker** — the *backend* +launches agent containers (compose), so the orchestrator needs no docker +socket. That keeps this control-plane container unprivileged; the host manages +both containers. `ensure_running` is an idempotent singleton (fixed container +names + the published port). """ from __future__ import annotations -import subprocess -import sys import time import urllib.error import urllib.request +from pathlib import Path from .. import log +from ..docker_cmd import run_docker from ..paths import bot_bottle_root +from .gateway import GATEWAY_IMAGE, GATEWAY_NETWORK, DockerGateway + +DEFAULT_PORT = 8099 +ORCHESTRATOR_NAME = "bot-bottle-orchestrator" +ORCHESTRATOR_LABEL = "bot-bottle-orchestrator=1" + +# The repo root is bind-mounted into the control-plane container so +# `python -m bot_bottle.orchestrator` resolves the package (the orchestrator +# is stdlib-only, so the bundle image's python is enough). +_REPO_ROOT = Path(__file__).resolve().parents[2] +_APP_DIR = "/app" +_ROOT_IN_CONTAINER = "/bot-bottle-root" -DEFAULT_HOST = "127.0.0.1" -DEFAULT_PORT = 8080 -# Poll cadence + default ceiling while waiting for a freshly-spawned control -# plane to answer /health (the first start also builds/boots the gateway). _HEALTH_POLL_SECONDS = 0.25 DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0 _HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0 class OrchestratorStartError(RuntimeError): - """The orchestrator process did not become healthy within the timeout.""" + """The orchestrator container did not become healthy within the timeout.""" -class OrchestratorProcess: - """Manages the local orchestrator control-plane process for the docker - backend. Backend-neutral callers only need `ensure_running()` + `url`.""" +class OrchestratorService: + """Manages the orchestrator control-plane container + the shared gateway. + Callers only need `ensure_running()` + `url`.""" def __init__( self, - host: str = DEFAULT_HOST, - port: int = DEFAULT_PORT, *, - broker: str = "docker", - gateway: bool = True, + port: int = DEFAULT_PORT, + network: str = GATEWAY_NETWORK, + image: str = GATEWAY_IMAGE, + repo_root: Path = _REPO_ROOT, + host_root: Path | None = None, ) -> None: - self.host = host self.port = port - self._broker = broker - self._gateway = gateway + self.network = network + self.image = image + self._repo_root = repo_root + self._host_root = host_root or bot_bottle_root() @property def url(self) -> str: - """The control-plane base URL — also what the data plane's - BOT_BOTTLE_ORCHESTRATOR_URL points at.""" - return f"http://{self.host}:{self.port}" + """Host-side control-plane URL (published loopback port).""" + return f"http://127.0.0.1:{self.port}" + + @property + def internal_url(self) -> str: + """Control-plane URL as the gateway container reaches it — by name over + docker DNS on the shared network. This is the gateway's + BOT_BOTTLE_ORCHESTRATOR_URL.""" + return f"http://{ORCHESTRATOR_NAME}:{self.port}" def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool: - """True iff a control plane answers `GET /health` with 200 — the - singleton liveness check.""" try: with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp: return resp.status == 200 except (urllib.error.URLError, TimeoutError, OSError): return False + def _container_running(self, name: str) -> bool: + proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"]) + return name in proc.stdout.split() + + def _run_orchestrator_container(self) -> None: + """Start the control-plane container (idempotent: clears a stale + fixed-name container first). Register-only broker → no docker socket.""" + run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME]) + proc = run_docker([ + "docker", "run", "--detach", + "--name", ORCHESTRATOR_NAME, + "--label", ORCHESTRATOR_LABEL, + "--network", self.network, + # Host CLI reaches the control plane here; bound to loopback so it + # is not exposed on the host's external interfaces. + "--publish", f"127.0.0.1:{self.port}:{self.port}", + "--volume", f"{self._repo_root}:{_APP_DIR}:ro", + "--workdir", _APP_DIR, + # Persist the registry DB on the host (sole-owner: only the + # orchestrator opens bot-bottle.db). + "--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}", + "--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}", + "--entrypoint", "python3", + self.image, + "-m", "bot_bottle.orchestrator", + "--host", "0.0.0.0", "--port", str(self.port), "--broker", "stub", + ]) + if proc.returncode != 0: + raise OrchestratorStartError( + f"orchestrator container failed to start: {proc.stderr.strip()}" + ) + + def _gateway(self) -> DockerGateway: + return DockerGateway(self.image, network=self.network, orchestrator_url=self.internal_url) + def ensure_running( self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS, ) -> str: - """Return the control-plane URL, starting the orchestrator first if it - isn't already healthy. Idempotent — a healthy control plane is left - untouched. Raises `OrchestratorStartError` if a freshly-spawned one - doesn't answer within `startup_timeout`.""" + """Ensure the control plane + shared gateway are up; return the host + control-plane URL. Idempotent — a healthy control plane and a running + gateway are left untouched. Raises `OrchestratorStartError` on + timeout.""" + gateway = self._gateway() + gateway.ensure_built() # build the bundle image if missing (both use it) if self.is_healthy(): + gateway.ensure_running() # make sure the gateway is up too return self.url - log.info("starting orchestrator", context={"url": self.url, "broker": self._broker}) - self._spawn() + + log.info("starting orchestrator container", context={"name": ORCHESTRATOR_NAME}) + gateway.ensure_running() # creates the shared network + gateway + self._run_orchestrator_container() + deadline = time.monotonic() + startup_timeout while time.monotonic() < deadline: if self.is_healthy(): @@ -93,49 +143,19 @@ class OrchestratorProcess: return self.url time.sleep(_HEALTH_POLL_SECONDS) raise OrchestratorStartError( - f"orchestrator at {self.url} did not become healthy within " - f"{startup_timeout:g}s" + f"orchestrator at {self.url} did not become healthy within {startup_timeout:g}s" ) - def _argv(self) -> list[str]: - """`python -m bot_bottle.orchestrator ...` — static flags only.""" - argv = [ - sys.executable, "-m", "bot_bottle.orchestrator", - "--host", self.host, "--port", str(self.port), - "--broker", self._broker, - ] - if self._gateway: - argv.append("--gateway") - return argv - - def _log_path(self) -> str: - """Where the detached orchestrator's stdout/stderr goes so a failed - start is diagnosable after the CLI has moved on.""" - return str(bot_bottle_root() / "orchestrator.log") - - def _spawn(self) -> None: - """Launch the orchestrator detached so it outlives this CLI process, - with its output tee'd to a log file under the bot-bottle root.""" - root = bot_bottle_root() - root.mkdir(parents=True, exist_ok=True) - logfile = open(self._log_path(), "a", encoding="utf-8") # noqa: SIM115 # pylint: disable=consider-using-with - try: - subprocess.Popen( # noqa: S603 # pylint: disable=consider-using-with - self._argv(), - stdout=logfile, - stderr=subprocess.STDOUT, - stdin=subprocess.DEVNULL, - start_new_session=True, - ) - finally: - # The child inherits its own dup'd fd; this handle is ours to drop. - logfile.close() + def stop(self) -> None: + """Remove the orchestrator + gateway containers (idempotent).""" + run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME]) + self._gateway().stop() __all__ = [ - "OrchestratorProcess", + "OrchestratorService", "OrchestratorStartError", - "DEFAULT_HOST", + "ORCHESTRATOR_NAME", "DEFAULT_PORT", "DEFAULT_STARTUP_TIMEOUT_SECONDS", ] diff --git a/tests/unit/test_consolidated_launch.py b/tests/unit/test_consolidated_launch.py index 2209e33..cce64c7 100644 --- a/tests/unit/test_consolidated_launch.py +++ b/tests/unit/test_consolidated_launch.py @@ -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() diff --git a/tests/unit/test_orchestrator_lifecycle.py b/tests/unit/test_orchestrator_lifecycle.py index 8002321..359e18a 100644 --- a/tests/unit/test_orchestrator_lifecycle.py +++ b/tests/unit/test_orchestrator_lifecycle.py @@ -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__":