From 24df322c31f85618707b7337808dbab844aa2cdf Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 13 Jul 2026 20:33:14 -0400 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20slice=2013a=20=E2=80=94?= =?UTF-8?q?=20orchestrator=20process=20lifecycle=20(idempotent=20singleton?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First sub-slice of docker launch integration: the foundation the CLI needs to ensure exactly one orchestrator control plane + shared gateway is up before registering/launching bottles. Does NOT touch the real launch path yet — it's the lifecycle primitive the cut-over slices build on. - OrchestratorProcess.ensure_running(): idempotent singleton — returns the control-plane URL if a healthy one already answers /health, else spawns `python -m bot_bottle.orchestrator --broker docker --gateway` detached (start_new_session, output tee'd to /orchestrator.log) and polls /health until healthy or timeout (OrchestratorStartError). - 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) per the PRD's dev-harness sequencing — it already has the host user's docker access to broker launches; the data-plane gateway it manages is the container. pyright 0 errors; pylint 9.83/10; unit suite green (1714 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 --- bot_bottle/orchestrator/lifecycle.py | 141 ++++++++++++++++++++++ tests/unit/test_orchestrator_lifecycle.py | 87 +++++++++++++ 2 files changed, 228 insertions(+) create mode 100644 bot_bottle/orchestrator/lifecycle.py create mode 100644 tests/unit/test_orchestrator_lifecycle.py diff --git a/bot_bottle/orchestrator/lifecycle.py b/bot_bottle/orchestrator/lifecycle.py new file mode 100644 index 0000000..405b99a --- /dev/null +++ b/bot_bottle/orchestrator/lifecycle.py @@ -0,0 +1,141 @@ +"""Orchestrator process 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. + +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. +""" + +from __future__ import annotations + +import subprocess +import sys +import time +import urllib.error +import urllib.request + +from .. import log +from ..paths import 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.""" + + +class OrchestratorProcess: + """Manages the local orchestrator control-plane process for the docker + backend. Backend-neutral callers only need `ensure_running()` + `url`.""" + + def __init__( + self, + host: str = DEFAULT_HOST, + port: int = DEFAULT_PORT, + *, + broker: str = "docker", + gateway: bool = True, + ) -> None: + self.host = host + self.port = port + self._broker = broker + self._gateway = gateway + + @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}" + + 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 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`.""" + if self.is_healthy(): + return self.url + log.info("starting orchestrator", context={"url": self.url, "broker": self._broker}) + self._spawn() + deadline = time.monotonic() + startup_timeout + while time.monotonic() < deadline: + if self.is_healthy(): + log.info("orchestrator healthy", context={"url": self.url}) + 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" + ) + + 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() + + +__all__ = [ + "OrchestratorProcess", + "OrchestratorStartError", + "DEFAULT_HOST", + "DEFAULT_PORT", + "DEFAULT_STARTUP_TIMEOUT_SECONDS", +] diff --git a/tests/unit/test_orchestrator_lifecycle.py b/tests/unit/test_orchestrator_lifecycle.py new file mode 100644 index 0000000..8002321 --- /dev/null +++ b/tests/unit/test_orchestrator_lifecycle.py @@ -0,0 +1,87 @@ +"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070).""" + +from __future__ import annotations + +import tempfile +import unittest +import urllib.error +from pathlib import Path +from unittest.mock import MagicMock, patch + +from bot_bottle.orchestrator.lifecycle import ( + OrchestratorProcess, + OrchestratorStartError, +) +from tests.unit import use_bottle_root + +_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen" +_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen" +_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): + 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) + + def test_url(self) -> None: + self.assertEqual("http://127.0.0.1:8099", self.p.url) + + def test_is_healthy_true_on_200(self) -> None: + with patch(_URLOPEN, return_value=_health(200)): + self.assertTrue(self.p.is_healthy()) + + def test_is_healthy_false_on_error(self) -> None: + with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): + self.assertFalse(self.p.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_spawns_then_waits_for_health(self) -> None: + # First check (before spawn) fails; after spawn the poll succeeds. + 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) + self.assertIn("bot_bottle.orchestrator", argv) + self.assertEqual("docker", argv[argv.index("--broker") + 1]) + + def test_argv_omits_gateway_when_disabled(self) -> None: + self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv()) + + 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()) + + +if __name__ == "__main__": + unittest.main()