feat(orchestrator): slice 13b — consolidated registration inputs (egress policy)

Bridges the per-bottle `prepare` output to the consolidated registry:
`registration_inputs(plan)` turns a prepared egress plan into the
backend-neutral inputs `Orchestrator.launch_bottle` takes.

- egress_policy(plan): the policy blob = the exact routes YAML the
  per-bottle sidecar read from a file; the multi-tenant gateway's
  PolicyResolver now fetches it from the registry per request instead.
  Same egress_render_routes, so a bottle's allow-list is byte-identical
  moving onto the shared gateway.
- RegistrationInputs bundles policy + metadata; metadata carries the human
  slug so the shared registry can map a minted bottle id back to a name
  (console / supervise display).

Host-side glue (imports bot_bottle.egress) used by the launch path — not
the lean orchestrator control-plane process. Not yet wired into launch
(that's 13c/13d), consistent with the slice-6/10 pattern of landing the
primitive before the cut-over.

Key test: the policy round-trips through load_config back to the same
routes + log level — the resolver-served policy reconstructs the per-bottle
egress config exactly.

pyright 0 errors; pylint 9.83/10; unit suite green (1718 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

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",
+]

bot_bottle/orchestrator/registration.py

@@ -0,0 +1,57 @@
+"""Consolidated registration inputs (PRD 0070, docker slice).
+
+Bridges the existing per-bottle `prepare` output to the consolidated
+registry: turns a prepared bottle's egress plan into the backend-neutral
+inputs `Orchestrator.launch_bottle` takes — the egress **policy** blob and
+launch **metadata**.
+
+The policy blob is the exact routes YAML the per-bottle egress sidecar used
+to read from a file; in the consolidated model the multi-tenant gateway's
+`PolicyResolver` fetches it from the registry per request (keyed by source
+IP) instead. Same render, so consolidated and single-tenant egress apply
+byte-identical policy — a bottle's allow-list doesn't change when it moves
+onto the shared gateway.
+
+Host-side glue (imports `bot_bottle.egress`), used by the launch path — not
+by the lean orchestrator control-plane process itself.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+
+from ..egress import EgressPlan, egress_render_routes
+
+
+@dataclass(frozen=True)
+class RegistrationInputs:
+    """What `Orchestrator.launch_bottle` needs to register a bottle, derived
+    from its prepared plan. `policy` is served verbatim by the gateway's
+    `/resolve`; `metadata` is opaque forward-compat state — it carries the
+    human slug so the console / supervise can show a name, not just the
+    minted bottle id."""
+
+    policy: str
+    metadata: str
+
+
+def egress_policy(plan: EgressPlan) -> str:
+    """The bottle's egress policy blob: the routes YAML the gateway serves
+    and the addon parses with `load_config`. Identical to the per-bottle
+    `routes.yaml` render, so the consolidated path applies the same
+    allow-list."""
+    return egress_render_routes(plan.routes, log=plan.log)
+
+
+def registration_inputs(plan: EgressPlan) -> RegistrationInputs:
+    """Assemble the orchestrator registration inputs from a prepared egress
+    plan. `metadata` records the slug so the shared registry can map a minted
+    bottle id back to its human name."""
+    return RegistrationInputs(
+        policy=egress_policy(plan),
+        metadata=json.dumps({"slug": plan.slug}),
+    )
+
+
+__all__ = ["RegistrationInputs", "egress_policy", "registration_inputs"]

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()

tests/unit/test_orchestrator_registration.py

@@ -0,0 +1,56 @@
+"""Unit: consolidated registration inputs — egress policy round-trip (PRD 0070)."""
+
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+
+from bot_bottle.egress import EgressPlan, EgressRoute
+from bot_bottle.egress_addon_core import LOG_BLOCKS, load_config
+from bot_bottle.orchestrator.registration import (
+    RegistrationInputs,
+    egress_policy,
+    registration_inputs,
+)
+
+
+def _plan(routes: tuple[EgressRoute, ...], *, slug: str = "demo", log: int = 0) -> EgressPlan:
+    return EgressPlan(
+        slug=slug,
+        routes_path=Path("/unused/routes.yaml"),
+        routes=routes,
+        token_env_map={},
+        log=log,
+    )
+
+
+class TestEgressPolicy(unittest.TestCase):
+    def test_policy_round_trips_through_load_config(self) -> None:
+        # The policy the gateway serves must parse back to the same allow-list
+        # the per-bottle sidecar applied — moving onto the shared gateway must
+        # not change a bottle's egress.
+        routes = (EgressRoute(host="api.example.com"), EgressRoute(host="pypi.org"))
+        cfg = load_config(egress_policy(_plan(routes)))
+        self.assertEqual(("api.example.com", "pypi.org"), tuple(r.host for r in cfg.routes))
+
+    def test_policy_preserves_log_level(self) -> None:
+        plan = _plan((EgressRoute(host="x.example.com"),), log=LOG_BLOCKS)
+        self.assertEqual(LOG_BLOCKS, load_config(egress_policy(plan)).log)
+
+    def test_empty_routes_yield_deny_all(self) -> None:
+        cfg = load_config(egress_policy(_plan(())))
+        self.assertEqual((), cfg.routes)  # no routes → default-deny
+
+
+class TestRegistrationInputs(unittest.TestCase):
+    def test_bundles_policy_and_slug_metadata(self) -> None:
+        plan = _plan((EgressRoute(host="api.example.com"),), slug="my-bot")
+        inputs = registration_inputs(plan)
+        self.assertIsInstance(inputs, RegistrationInputs)
+        self.assertEqual("my-bot", json.loads(inputs.metadata)["slug"])
+        self.assertEqual(egress_policy(plan), inputs.policy)  # same blob egress_policy renders
+
+
+if __name__ == "__main__":
+    unittest.main()