Compare commits

...

4 Commits

Author SHA1 Message Date
didericis 273f51223d fix(orchestrator): gateway image builds cache-aware, not build-if-missing
lint / lint (push) Successful in 2m13s
test / unit (pull_request) Successful in 1m8s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m22s
The cut-over dropped compose (whose `build:` rebuilt the sidecar image on
every up), and DockerGateway.ensure_built was build-if-missing — so a change
to the gateway's flat sources (egress addon / git-http / policy_resolver /
supervise) left a STALE image silently running the old single-tenant daemons
(this bit e2e validation). ensure_built now always `docker build`s
(cache-aware: a cache check when nothing changed, a real rebuild when sources
moved), matching the old compose behavior. BOT_BOTTLE_NO_CACHE forces a full
rebuild (parity with #354's `start --no-cache`).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-14 01:13:11 -04:00
didericis 9d85541b67 fix(tests): annotate on_network param — pyright strict (CI lint)
lint / lint (push) Successful in 2m22s
test / unit (pull_request) Successful in 1m2s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m15s
The cut-over's edit to test_consolidated_launch left the new on_network
kwarg untyped; pyright strict rejects it. No behavior change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-14 00:55:30 -04:00
didericis d65e97de73 feat(backend): slice 13d — cut docker launch over to the consolidated gateway (e2e green)
lint / lint (push) Failing after 2m9s
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m36s
Rewire DockerBottleBackend.launch to the consolidated model, replacing the
per-bottle sidecar bundle. VALIDATED END-TO-END on real docker:
test_sandbox_escape passes all 5 attacks (egress DLP + git-gate gitleaks)
through the shared gateway.

launch now:
  - mints git-gate dynamic keys (if any), then launch_consolidated() to
    register the bottle + provision its git-gate state into the gateway and
    get the agent's attach context (pinned source IP, gateway address);
  - installs the SHARED gateway CA (gateway.ca_cert_pem) into the agent;
  - renders the agent-only consolidated compose on the gateway network at the
    pinned IP, proxied through the gateway;
  - points the agent's git-gate insteadOf (http://<gw>:9420) and supervise
    MCP (http://<gw>:9100) at the gateway (DockerBottlePlan.agent_git_gate_url
    / agent_supervise_url);
  - teardown = compose down + teardown_consolidated (dereg + deprovision).
Dropped the per-bottle networks, per-bottle egress CA, and bundle service.

Fixes surfaced by the real e2e (couldn't be caught by unit mocks):
  - provision_git_gate installs the bottle-agnostic pre-receive/access hooks
    into the gateway (were cp'd per-bundle before);
  - source-IP allocation reads the *actual* container IPs on the network
    (gateway + orchestrator + agents), not just gateway + registry — the
    orchestrator container sits on the network and was colliding.

Unit tests for the old bundle launch rewritten against the new collaborators.

NOTE for reviewers: the gateway/bundle image (bot-bottle-sidecars) must be
rebuilt when its flat sources change — `ensure_built` only builds-if-missing,
so a stale image silently runs the OLD single-tenant daemons. A content-hash
/ --rebuild path is a follow-up.

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);
test_sandbox_escape green on real docker.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 23:28:36 -04:00
didericis a1acbc00ed feat(orchestrator): slice 13d(i) — containerize the orchestrator (validated on docker)
lint / lint (push) Successful in 2m5s
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 21s
test / coverage (pull_request) Successful in 1m13s
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
2026-07-13 23:07:49 -04:00
15 changed files with 413 additions and 385 deletions
+5
View File
@@ -111,6 +111,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
plumbing needed; the alias resolves inside the bridge.""" plumbing needed; the alias resolves inside the bridge."""
if plan.supervise_plan is None: if plan.supervise_plan is None:
return "" return ""
# Consolidated: the supervise daemon lives on the shared gateway, so
# the agent registers the gateway address (NO_PROXY bypasses egress),
# not the per-bottle `supervise` alias.
if plan.agent_supervise_url:
return plan.agent_supervise_url
return f"http://{SUPERVISE_HOSTNAME}:{SUPERVISE_PORT}/" return f"http://{SUPERVISE_HOSTNAME}:{SUPERVISE_PORT}/"
def prepare_cleanup(self) -> DockerBottleCleanupPlan: def prepare_cleanup(self) -> DockerBottleCleanupPlan:
+19
View File
@@ -28,11 +28,30 @@ class DockerBottlePlan(BottlePlan):
# accidental log of the plan dataclass. # accidental log of the plan dataclass.
forwarded_env: dict[str, str] = field(repr=False) forwarded_env: dict[str, str] = field(repr=False)
use_runsc: bool use_runsc: bool
# Consolidated mode (PRD 0070): the agent reaches git-gate over HTTP at the
# shared gateway's address (`http://<gateway_ip>:9420`), set at launch.
# Empty → single-tenant defaults (the `git-gate` alias over git://).
agent_git_gate_url: str = ""
# Likewise the supervise MCP endpoint at the gateway (`http://<gw>:9100/`);
# empty → the single-tenant `supervise` alias.
agent_supervise_url: str = ""
@property @property
def container_name(self) -> str: def container_name(self) -> str:
return self.agent_provision.instance_name return self.agent_provision.instance_name
@property
def git_gate_insteadof_host(self) -> str:
if self.agent_git_gate_url.startswith("http://"):
return self.agent_git_gate_url.removeprefix("http://").rstrip("/")
return super().git_gate_insteadof_host
@property
def git_gate_insteadof_scheme(self) -> str:
if self.agent_git_gate_url.startswith("http://"):
return "http"
return super().git_gate_insteadof_scheme
@property @property
def image(self) -> str: def image(self) -> str:
return self.agent_provision.image return self.agent_provision.image
@@ -26,7 +26,7 @@ from ...egress import EgressPlan
from ...git_gate import GitGatePlan from ...git_gate import GitGatePlan
from ...orchestrator.client import OrchestratorClient from ...orchestrator.client import OrchestratorClient
from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK 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 ...orchestrator.registration import registration_inputs
from .gateway_net import next_free_ip from .gateway_net import next_free_ip
from .gateway_provision import deprovision_git_gate, provision_git_gate from .gateway_provision import deprovision_git_gate, provision_git_gate
@@ -76,15 +76,21 @@ def _container_ip(name: str, network: str) -> str:
return ip return ip
def _taken_ips(client: OrchestratorClient, gateway_ip: str) -> list[str]: def _network_container_ips(network: str) -> list[str]:
"""Every address already in use on the gateway network: the gateway """Every address currently assigned on the gateway network the ground
container plus every live bottle the registry knows.""" truth for "in use": the gateway + orchestrator infrastructure containers
taken = [gateway_ip] and every live agent. Read from the network so a new bottle can't collide
for rec in client.list_bottles(): with anything actually attached (a registry-only view would miss the
src = rec.get("source_ip") orchestrator/gateway containers)."""
if isinstance(src, str) and src: proc = run_docker([
taken.append(src) "docker", "network", "inspect", "--format",
return taken "{{range .Containers}}{{.IPv4Address}} {{end}}", network,
])
ips: list[str] = []
for entry in proc.stdout.split():
# entries look like "172.20.0.2/16" — keep the address.
ips.append(entry.split("/", 1)[0])
return ips
def launch_consolidated( def launch_consolidated(
@@ -92,7 +98,7 @@ def launch_consolidated(
git_gate_plan: GitGatePlan, git_gate_plan: GitGatePlan,
*, *,
image_ref: str = "", image_ref: str = "",
process: OrchestratorProcess | None = None, service: OrchestratorService | None = None,
gateway_name: str = GATEWAY_NAME, gateway_name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK, network: str = GATEWAY_NETWORK,
) -> LaunchContext: ) -> LaunchContext:
@@ -100,13 +106,13 @@ def launch_consolidated(
bottle, and provision its git-gate state. Returns the agent's attach bottle, and provision its git-gate state. Returns the agent's attach
context. Raises `ConsolidatedLaunchError` (or the primitives' own errors) context. Raises `ConsolidatedLaunchError` (or the primitives' own errors)
if any step fails — the caller tears down on failure.""" if any step fails — the caller tears down on failure."""
process = process or OrchestratorProcess() service = service or OrchestratorService()
url = process.ensure_running() url = service.ensure_running()
client = OrchestratorClient(url) client = OrchestratorClient(url)
cidr = _network_cidr(network) cidr = _network_cidr(network)
gateway_ip = _container_ip(gateway_name, network) gateway_ip = _container_ip(gateway_name, network)
source_ip = next_free_ip(cidr, _taken_ips(client, gateway_ip)) source_ip = next_free_ip(cidr, _network_container_ips(network))
inputs = registration_inputs(egress_plan) inputs = registration_inputs(egress_plan)
reg = client.register_bottle( reg = client.register_bottle(
@@ -67,6 +67,12 @@ def provision_git_gate(gateway: str, bottle_id: str, plan: GitGatePlan) -> None:
_require_safe(bottle_id) _require_safe(bottle_id)
if not plan.upstreams: if not plan.upstreams:
return return
# The pre-receive + access hooks are bottle-agnostic and shared by every
# bottle's repos; install them into the gateway (idempotent — same content
# each time). The per-bottle model cp'd these into each bundle at start.
_exec(gateway, ["mkdir", "-p", "/etc/git-gate"])
_cp_into(gateway, str(plan.hook_script), "/etc/git-gate/pre-receive")
_cp_into(gateway, str(plan.access_hook_script), "/etc/git-gate/access-hook")
creds = _creds_dir(bottle_id) creds = _creds_dir(bottle_id)
_exec(gateway, ["mkdir", "-p", creds]) _exec(gateway, ["mkdir", "-p", creds])
for u in plan.upstreams: for u in plan.upstreams:
+47 -62
View File
@@ -36,13 +36,11 @@ from contextlib import ExitStack, contextmanager
from pathlib import Path from pathlib import Path
from typing import Callable, Generator from typing import Callable, Generator
from ...egress import egress_resolve_token_values
from ...git_gate import ( from ...git_gate import (
provision_git_gate_dynamic_keys, provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys, revoke_git_gate_provisioned_keys,
) )
from ...log import info, warn from ...log import info, warn
from . import network as network_mod
from . import util as docker_mod from . import util as docker_mod
from .bottle import DockerBottle from .bottle import DockerBottle
from .bottle_plan import DockerBottlePlan from .bottle_plan import DockerBottlePlan
@@ -53,7 +51,6 @@ from ...bottle_state import (
read_committed_image, read_committed_image,
) )
from .compose import ( from .compose import (
bottle_plan_to_compose,
compose_down, compose_down,
compose_dump_logs, compose_dump_logs,
compose_file_path, compose_file_path,
@@ -62,7 +59,9 @@ from .compose import (
compose_up, compose_up,
write_compose_file, write_compose_file,
) )
from .egress import egress_tls_init from .consolidated_compose import consolidated_agent_compose
from .consolidated_launch import launch_consolidated, teardown_consolidated
from ...orchestrator.gateway import DockerGateway
# Where the repo root lives, for `docker build` context. Computed once. # Where the repo root lives, for `docker build` context. Computed once.
@@ -97,8 +96,7 @@ def launch(
try: try:
# Step 1: agent image. Use a committed snapshot when one exists # Step 1: agent image. Use a committed snapshot when one exists
# and is present in the local daemon; otherwise build from the # and is present in the local daemon; otherwise build from the
# Dockerfile. Sidecar images get built lazily by `docker compose # Dockerfile. (The gateway image is built by the orchestrator.)
# up` via the renderer's `build:` directives.
committed = read_committed_image(plan.slug) committed = read_committed_image(plan.slug)
if committed and docker_mod.image_exists(committed): if committed and docker_mod.image_exists(committed):
info(f"using committed image {committed!r}") info(f"using committed image {committed!r}")
@@ -112,82 +110,73 @@ def launch(
dockerfile=plan.dockerfile_path, dockerfile=plan.dockerfile_path,
) )
internal_network = network_mod.network_name_for_slug(plan.slug) # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
egress_network = network_mod.network_egress_name_for_slug(plan.slug) # provisioning the bottle's repos into the shared gateway.
egress_ca_host, egress_ca_cert_only = egress_tls_init(
egress_state_dir(plan.slug),
)
git_gate_plan = plan.git_gate_plan git_gate_plan = plan.git_gate_plan
if git_gate_plan.upstreams: if git_gate_plan.upstreams:
git_gate_plan = provision_git_gate_dynamic_keys( git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle, plan.manifest.bottle, git_gate_plan, git_gate_state_dir(plan.slug),
git_gate_plan,
git_gate_state_dir(plan.slug),
)
git_gate_plan = dataclasses.replace(
git_gate_plan,
internal_network=internal_network,
egress_network=egress_network,
) )
# Step 3: register on the orchestrator + provision this bottle's
# git-gate state into the shared gateway; get the agent's attach
# context (pinned source IP, gateway address, shared network).
ctx = launch_consolidated(plan.egress_plan, git_gate_plan, image_ref=plan.image)
stack.callback(
teardown_consolidated, ctx.bottle_id, orchestrator_url=ctx.orchestrator_url,
)
# Step 4: install the SHARED gateway CA into the agent (replaces the
# per-bottle CA) — read it out of the running gateway.
ca_dir = egress_state_dir(plan.slug) / "gateway-ca"
ca_dir.mkdir(parents=True, exist_ok=True)
ca_file = ca_dir / "gateway-ca.pem"
ca_file.write_text(DockerGateway(network=ctx.network).ca_cert_pem())
egress_plan = dataclasses.replace( egress_plan = dataclasses.replace(
plan.egress_plan, plan.egress_plan,
internal_network=internal_network, mitmproxy_ca_host_path=ca_file,
egress_network=egress_network, mitmproxy_ca_cert_only_host_path=ca_file,
mitmproxy_ca_host_path=egress_ca_host, )
mitmproxy_ca_cert_only_host_path=egress_ca_cert_only, # Point the agent's git-gate insteadOf rewrites at the shared gateway's
# HTTP git endpoint (9420) instead of the dead per-bottle `git-gate`
# alias. git-http + supervise on the gateway bypass the egress proxy
# (NO_PROXY includes the gateway address).
git_gate_url = (
f"http://{ctx.gateway_ip}:9420" if git_gate_plan.upstreams else ""
)
supervise_url = (
f"http://{ctx.gateway_ip}:9100/" if plan.supervise_plan is not None else ""
) )
supervise_plan = plan.supervise_plan
if supervise_plan is not None:
supervise_plan = dataclasses.replace(
supervise_plan,
internal_network=internal_network,
)
plan = dataclasses.replace( plan = dataclasses.replace(
plan, plan,
git_gate_plan=git_gate_plan, git_gate_plan=git_gate_plan,
egress_plan=egress_plan, egress_plan=egress_plan,
supervise_plan=supervise_plan, agent_git_gate_url=git_gate_url,
agent_supervise_url=supervise_url,
) )
# Step 6: render + write the compose file. metadata.json # Step 5: render + up the agent-only compose, pinned on the shared
# was written at prepare time and already carries # gateway network and proxied through the gateway's address.
# compose_project; nothing to update here.
state_dir = bottle_state_dir(plan.slug) state_dir = bottle_state_dir(plan.slug)
spec = bottle_plan_to_compose(plan) spec = consolidated_agent_compose(
plan, gateway_ip=ctx.gateway_ip, source_ip=ctx.source_ip, network=ctx.network,
)
compose_file = write_compose_file(spec, compose_file_path(state_dir)) compose_file = write_compose_file(spec, compose_file_path(state_dir))
project = compose_project_name(plan.slug) project = compose_project_name(plan.slug)
# Forwarded vars (OAuth token, host interpolations) flow through the
# Step 7: compose up. Token values + the OAuth placeholder # subprocess env as bare names so values never land in the file.
# flow through subprocess env; the compose file holds only compose_env: dict[str, str] = {**os.environ, **plan.forwarded_env}
# bare names for the secret-carrying entries.
effective_env = {**dict(os.environ), **plan.agent_provision.provisioned_env}
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, effective_env,
)
compose_env: dict[str, str] = {
**os.environ,
**plan.forwarded_env,
**token_values,
}
info( info(
f"docker compose up -d (project {project}, " f"docker compose up -d (project {project}, agent on shared "
f"{len(spec['services'])} services)" f"gateway {ctx.gateway_ip}, ip {ctx.source_ip})"
) )
compose_up(project, compose_file, env=compose_env) compose_up(project, compose_file, env=compose_env)
# Register teardown in reverse order: log dump first, then
# `compose down`. Networks come down last via callbacks
# registered in step 2.
stack.callback(compose_down, project, compose_file) stack.callback(compose_down, project, compose_file)
stack.callback( stack.callback(
compose_dump_logs, project, compose_file, compose_log_path(state_dir), compose_dump_logs, project, compose_file, compose_log_path(state_dir),
) )
# Step 8: provision. Create the bottle first so provisioners # Step 6: provision (CA install now uses the gateway CA) + yield.
# can use bottle.exec / bottle.cp_in; set the prompt path
# returned by provision_prompt after the fact.
bottle = DockerBottle( bottle = DockerBottle(
plan.container_name, plan.container_name,
teardown, teardown,
@@ -200,10 +189,6 @@ def launch(
agent_workdir=plan.workspace_plan.workdir, agent_workdir=plan.workspace_plan.workdir,
) )
bottle.prompt_path = provision(plan, bottle) bottle.prompt_path = provision(plan, bottle)
# Step 9: yield. exec_agent continues to use `docker exec -it`
# — the agent runs `sleep infinity` per the renderer's
# service spec.
yield bottle yield bottle
finally: finally:
teardown() teardown()
+7 -1
View File
@@ -51,7 +51,13 @@ def main(argv: list[str] | None = None) -> int:
# anything; 'docker' runs real containers (firecracker drops in later). # anything; 'docker' runs real containers (firecracker drops in later).
secret = secrets.token_bytes(32) secret = secrets.token_bytes(32)
broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret) 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) orchestrator = Orchestrator(registry, broker, secret, gateway)
# One persistent per-host gateway, shared by every bottle: build the # One persistent per-host gateway, shared by every bottle: build the
+11 -2
View File
@@ -34,6 +34,7 @@ import http.server
import json import json
import os import os
import socketserver import socketserver
import sys
import typing import typing
from urllib.parse import urlsplit from urllib.parse import urlsplit
@@ -151,12 +152,20 @@ class Handler(http.server.BaseHTTPRequestHandler):
super().log_message(format, *args) super().log_message(format, *args)
def _serve(self, method: str) -> None: 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 server = self.server
assert isinstance(server, ControlPlaneServer) assert isinstance(server, ControlPlaneServer)
length = int(self.headers.get("Content-Length") or 0) length = int(self.headers.get("Content-Length") or 0)
body = self.rfile.read(length) if length > 0 else b"" 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() data = json.dumps(payload).encode()
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
+29 -13
View File
@@ -91,12 +91,17 @@ class DockerGateway(Gateway):
*, *,
name: str = GATEWAY_NAME, name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK, network: str = GATEWAY_NETWORK,
orchestrator_url: str = "",
build_context: Path | None = None, build_context: Path | None = None,
dockerfile: str | None = GATEWAY_DOCKERFILE, dockerfile: str | None = GATEWAY_DOCKERFILE,
) -> None: ) -> None:
self.image_ref = image_ref self.image_ref = image_ref
self.name = name self.name = name
self.network = network 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._build_context = build_context or _REPO_ROOT
self._dockerfile = dockerfile self._dockerfile = dockerfile
@@ -104,17 +109,23 @@ class DockerGateway(Gateway):
return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0 return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0
def ensure_built(self) -> None: def ensure_built(self) -> None:
"""Build the bundle image from its Dockerfile when it's missing. """Build the bundle image from its Dockerfile, **cache-aware** — cheap
No-op when the image is already present, or when no dockerfile is (a cache check) when nothing changed, a real rebuild when the flat
configured (e.g. a pre-pulled image).""" sources (egress addon / git-http / policy_resolver / supervise) moved.
if self._dockerfile is None or self.image_exists():
This deliberately builds every time rather than build-if-missing: the
per-bottle model kept the image fresh via compose's `build:` on up, and
a stale image silently runs the OLD single-tenant daemons. No-op only
when no dockerfile is configured (a pre-pulled image). BOT_BOTTLE_NO_CACHE
forces a full rebuild (parity with `start --no-cache`)."""
if self._dockerfile is None:
return return
proc = run_docker([ argv = ["docker", "build", "-t", self.image_ref,
"docker", "build", "-f", str(self._build_context / self._dockerfile),
"-t", self.image_ref, str(self._build_context)]
"-f", str(self._build_context / self._dockerfile), if os.environ.get("BOT_BOTTLE_NO_CACHE"):
str(self._build_context), argv.insert(2, "--no-cache")
]) proc = run_docker(argv)
if proc.returncode != 0: if proc.returncode != 0:
raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}") raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}")
@@ -146,7 +157,7 @@ class DockerGateway(Gateway):
# Clear any stale (stopped) container holding the fixed name, then # Clear any stale (stopped) container holding the fixed name, then
# start fresh. `rm --force` on an absent name is a tolerated no-op. # start fresh. `rm --force` on an absent name is a tolerated no-op.
run_docker(["docker", "rm", "--force", self.name]) run_docker(["docker", "rm", "--force", self.name])
proc = run_docker([ argv = [
"docker", "run", "--detach", "docker", "run", "--detach",
"--name", self.name, "--name", self.name,
"--label", GATEWAY_LABEL, "--label", GATEWAY_LABEL,
@@ -154,8 +165,13 @@ class DockerGateway(Gateway):
# Persist the self-generated CA so it survives restarts (agents # Persist the self-generated CA so it survives restarts (agents
# trust it) — see GATEWAY_CA_VOLUME. # trust it) — see GATEWAY_CA_VOLUME.
"--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}", "--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: if proc.returncode != 0:
raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}") raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
+102 -82
View File
@@ -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 Runs the orchestrator control plane **as a container** on the shared gateway
model, exactly one orchestrator control plane — and the single per-host network, alongside the gateway container. This is the PRD's "virtualize the
gateway it manages — must be running. This starts the orchestrator orchestrator": container↔container between the gateway and the orchestrator
dev-harness (`python -m bot_bottle.orchestrator`) as a background host avoids the host firewall (which drops container→host traffic), and the gateway
process and health-checks it. 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 The orchestrator runs with the **register-only broker** — the *backend*
healthy control plane already answers on the port, and otherwise spawns one launches agent containers (compose), so the orchestrator needs no docker
and waits for it to come up. The control-plane port is the singleton key — socket. That keeps this control-plane container unprivileged; the host manages
a second orchestrator can't bind it, so a stray double-start fails fast both containers. `ensure_running` is an idempotent singleton (fixed container
rather than forking a rival. names + the published port).
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 from __future__ import annotations
import subprocess
import sys
import time import time
import urllib.error import urllib.error
import urllib.request import urllib.request
from pathlib import Path
from .. import log from .. import log
from ..docker_cmd import run_docker
from ..paths import bot_bottle_root 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 _HEALTH_POLL_SECONDS = 0.25
DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0 DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0 _HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
class OrchestratorStartError(RuntimeError): 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: class OrchestratorService:
"""Manages the local orchestrator control-plane process for the docker """Manages the orchestrator control-plane container + the shared gateway.
backend. Backend-neutral callers only need `ensure_running()` + `url`.""" Callers only need `ensure_running()` + `url`."""
def __init__( def __init__(
self, self,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
*, *,
broker: str = "docker", port: int = DEFAULT_PORT,
gateway: bool = True, network: str = GATEWAY_NETWORK,
image: str = GATEWAY_IMAGE,
repo_root: Path = _REPO_ROOT,
host_root: Path | None = None,
) -> None: ) -> None:
self.host = host
self.port = port self.port = port
self._broker = broker self.network = network
self._gateway = gateway self.image = image
self._repo_root = repo_root
self._host_root = host_root or bot_bottle_root()
@property @property
def url(self) -> str: def url(self) -> str:
"""The control-plane base URL — also what the data plane's """Host-side control-plane URL (published loopback port)."""
BOT_BOTTLE_ORCHESTRATOR_URL points at.""" return f"http://127.0.0.1:{self.port}"
return f"http://{self.host}:{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: 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: try:
with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp: with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
return resp.status == 200 return resp.status == 200
except (urllib.error.URLError, TimeoutError, OSError): except (urllib.error.URLError, TimeoutError, OSError):
return False 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( def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS, self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> str: ) -> str:
"""Return the control-plane URL, starting the orchestrator first if it """Ensure the control plane + shared gateway are up; return the host
isn't already healthy. Idempotent — a healthy control plane is left control-plane URL. Idempotent — a healthy control plane and a running
untouched. Raises `OrchestratorStartError` if a freshly-spawned one gateway are left untouched. Raises `OrchestratorStartError` on
doesn't answer within `startup_timeout`.""" timeout."""
gateway = self._gateway()
gateway.ensure_built() # build the bundle image if missing (both use it)
if self.is_healthy(): if self.is_healthy():
gateway.ensure_running() # make sure the gateway is up too
return self.url 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 deadline = time.monotonic() + startup_timeout
while time.monotonic() < deadline: while time.monotonic() < deadline:
if self.is_healthy(): if self.is_healthy():
@@ -93,49 +143,19 @@ class OrchestratorProcess:
return self.url return self.url
time.sleep(_HEALTH_POLL_SECONDS) time.sleep(_HEALTH_POLL_SECONDS)
raise OrchestratorStartError( raise OrchestratorStartError(
f"orchestrator at {self.url} did not become healthy within " f"orchestrator at {self.url} did not become healthy within {startup_timeout:g}s"
f"{startup_timeout:g}s"
) )
def _argv(self) -> list[str]: def stop(self) -> None:
"""`python -m bot_bottle.orchestrator ...` — static flags only.""" """Remove the orchestrator + gateway containers (idempotent)."""
argv = [ run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME])
sys.executable, "-m", "bot_bottle.orchestrator", self._gateway().stop()
"--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__ = [ __all__ = [
"OrchestratorProcess", "OrchestratorService",
"OrchestratorStartError", "OrchestratorStartError",
"DEFAULT_HOST", "ORCHESTRATOR_NAME",
"DEFAULT_PORT", "DEFAULT_PORT",
"DEFAULT_STARTUP_TIMEOUT_SECONDS", "DEFAULT_STARTUP_TIMEOUT_SECONDS",
] ]
+12 -8
View File
@@ -39,14 +39,18 @@ def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
class TestLaunchConsolidated(unittest.TestCase): class TestLaunchConsolidated(unittest.TestCase):
def _run(self, client: Mock, provision: Mock | None = None): def _run(
process = MagicMock() self, client: Mock, provision: Mock | None = None,
process.ensure_running.return_value = "http://orch:8080" *, on_network: tuple[str, ...] = ("172.18.0.2",),
):
service = MagicMock()
service.ensure_running.return_value = "http://orch:8080"
with patch(f"{_MOD}._network_cidr", return_value="172.18.0.0/16"), \ 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}._container_ip", return_value="172.18.0.2"), \
patch(f"{_MOD}._network_container_ips", return_value=list(on_network)), \
patch(f"{_MOD}.OrchestratorClient", return_value=client), \ patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()): 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: def test_allocates_ip_registers_and_provisions(self) -> None:
client = _client() client = _client()
@@ -64,10 +68,10 @@ class TestLaunchConsolidated(unittest.TestCase):
self.assertIn("api.example.com", kwargs.kwargs["policy"]) self.assertIn("api.example.com", kwargs.kwargs["policy"])
provision.assert_called_once() provision.assert_called_once()
def test_skips_gateway_and_live_bottle_addresses(self) -> None: def test_skips_all_addresses_on_the_network(self) -> None:
client = _client(bottles=[{"source_ip": "172.18.0.3"}]) # Gateway .2 + orchestrator .3 already attached -> agent gets .4.
ctx = self._run(client) ctx = self._run(_client(), on_network=("172.18.0.2", "172.18.0.3"))
self.assertEqual("172.18.0.4", ctx.source_ip) # .2 gw, .3 taken → .4 self.assertEqual("172.18.0.4", ctx.source_ip)
def test_provision_failure_rolls_back_registration(self) -> None: def test_provision_failure_rolls_back_registration(self) -> None:
client = _client() client = _client()
+64 -113
View File
@@ -1,4 +1,4 @@
"""Unit: Docker launch step uses committed image when available.""" """Unit: Docker launch step uses committed image when available (consolidated)."""
from __future__ import annotations from __future__ import annotations
@@ -6,6 +6,7 @@ import contextlib
import io import io
import tempfile import tempfile
import unittest import unittest
from collections.abc import Callable, Generator
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from unittest import mock from unittest import mock
@@ -14,9 +15,11 @@ from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import BottleSpec from bot_bottle.backend import BottleSpec
from bot_bottle.backend.docker import launch as launch_mod from bot_bottle.backend.docker import launch as launch_mod
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.docker.consolidated_launch import LaunchContext
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
from tests.unit import use_bottle_root
_SLUG = "dev-abc12" _SLUG = "dev-abc12"
@@ -28,163 +31,111 @@ _IDX = ManifestIndex.from_json_obj({
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}) })
_CTX = LaunchContext(
bottle_id="b1", identity_token="t", source_ip="172.20.0.4",
network="bot-bottle-gateway", gateway_ip="172.20.0.2",
orchestrator_url="http://orch:8099",
)
def _plan(tmp: str) -> DockerBottlePlan: def _plan(tmp: str) -> DockerBottlePlan:
stage = Path(tmp) stage = Path(tmp)
spec = BottleSpec( spec = BottleSpec(
manifest=_IDX, manifest=_IDX, agent_name="demo", copy_cwd=False, user_cwd=tmp, identity=_SLUG,
agent_name="demo",
copy_cwd=False,
user_cwd=tmp,
identity=_SLUG,
) )
return DockerBottlePlan( return DockerBottlePlan(
spec=spec, spec=spec,
manifest=_IDX.load_for_agent("demo"), manifest=_IDX.load_for_agent("demo"),
stage_dir=stage, stage_dir=stage,
git_gate_plan=GitGatePlan( git_gate_plan=GitGatePlan(
slug=_SLUG, slug=_SLUG, entrypoint_script=stage / "e.sh", hook_script=stage / "h.sh",
entrypoint_script=stage / "entrypoint.sh", access_hook_script=stage / "a.sh", upstreams=(),
hook_script=stage / "hook.sh",
access_hook_script=stage / "access-hook.sh",
upstreams=(),
), ),
egress_plan=EgressPlan( egress_plan=EgressPlan(
slug=_SLUG, slug=_SLUG, routes_path=stage / "egress.yaml", routes=(), token_env_map={},
routes_path=stage / "egress.yaml",
routes=(),
token_env_map={},
), ),
supervise_plan=None, supervise_plan=None,
agent_provision=AgentProvisionPlan( agent_provision=AgentProvisionPlan(
template="claude", template="claude", command="claude", prompt_mode="append_file",
command="claude", image=_DEFAULT_IMAGE, dockerfile="", guest_home="/home/node",
prompt_mode="append_file", instance_name=f"bot-bottle-{_SLUG}", prompt_file=stage / "prompt.txt",
image=_DEFAULT_IMAGE,
dockerfile="",
guest_home="/home/node",
instance_name=f"bot-bottle-{_SLUG}",
prompt_file=stage / "prompt.txt",
guest_env={}, guest_env={},
), ),
slug=_SLUG, slug=_SLUG, forwarded_env={}, use_runsc=False,
forwarded_env={},
use_runsc=False,
) )
class TestLaunchCommittedImage(unittest.TestCase): class TestLaunchCommittedImage(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self._tmp = tempfile.mkdtemp(prefix="launch-committed-test.") self._tmp = tempfile.mkdtemp(prefix="launch-committed-test.")
self.addCleanup(lambda: __import__("shutil").rmtree(self._tmp, ignore_errors=True))
# The launch writes the gateway CA under the bottle root — sandbox it.
self.addCleanup(use_bottle_root(Path(self._tmp)))
def tearDown(self) -> None: @contextlib.contextmanager
import shutil def _patched(
shutil.rmtree(self._tmp, ignore_errors=True)
def _run_launch(
self, self,
plan: DockerBottlePlan,
*, *,
committed_tag: str | None = None, committed_tag: str | None,
image_present: bool = True, image_present: bool,
) -> list[str]: compose: Callable[..., dict[str, Any]],
"""Drive launch() through its full sequence with the committed-image ) -> Generator[list[str], None, None]:
behaviour controlled by the arguments. Returns the images that were
passed to `build_image` (empty list if it was never called)."""
built: list[str] = [] built: list[str] = []
gw = mock.Mock()
gw.ca_cert_pem.return_value = "-----BEGIN CERTIFICATE-----\nx\n-----END CERTIFICATE-----\n"
def fake_build(image: str, ctx: str, *, dockerfile: str = "") -> None: def _build(image: str, ctx: str, *, dockerfile: str = "") -> None:
del ctx, dockerfile del ctx, dockerfile
built.append(image) built.append(image)
with mock.patch.object( with mock.patch.object(launch_mod, "read_committed_image", return_value=committed_tag), \
launch_mod, "read_committed_image", return_value=committed_tag, mock.patch.object(launch_mod.docker_mod, "image_exists", return_value=image_present), \
), mock.patch.object( mock.patch.object(launch_mod.docker_mod, "build_image", side_effect=_build), \
launch_mod.docker_mod, "image_exists", return_value=image_present, mock.patch.object(launch_mod, "launch_consolidated", return_value=_CTX), \
), mock.patch.object( mock.patch.object(launch_mod, "teardown_consolidated"), \
launch_mod.docker_mod, "build_image", side_effect=fake_build, mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
), mock.patch.object( mock.patch.object(launch_mod, "consolidated_agent_compose", side_effect=compose), \
launch_mod, "egress_tls_init", mock.patch.object(launch_mod, "write_compose_file", return_value=Path("/tmp/c.yml")), \
return_value=(Path("/egress_ca"), Path("/egress_cert")), mock.patch.object(launch_mod, "compose_up"), \
), mock.patch.object( mock.patch.object(launch_mod, "compose_dump_logs"), \
launch_mod.network_mod, "network_name_for_slug", mock.patch.object(launch_mod, "compose_down"), \
return_value="bb-internal", contextlib.redirect_stderr(io.StringIO()):
), mock.patch.object( yield built
launch_mod.network_mod, "network_egress_name_for_slug",
return_value="bb-egress",
), mock.patch.object(
launch_mod, "bottle_plan_to_compose",
return_value={"services": {"agent": {}}},
), mock.patch.object(
launch_mod, "write_compose_file",
return_value=Path("/tmp/compose.yml"),
), mock.patch.object(launch_mod, "compose_up"), \
mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()):
provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, provision=provision):
pass
def _run_launch(
self, plan: DockerBottlePlan, *,
committed_tag: str | None = None, image_present: bool = True,
) -> list[str]:
def compose(_p: DockerBottlePlan, **_kw: Any) -> dict[str, Any]:
return {"services": {"agent": {}}}
with self._patched(
committed_tag=committed_tag, image_present=image_present, compose=compose,
) as built:
with launch_mod.launch(plan, provision=mock.Mock(return_value=None)):
pass
return built return built
def test_skips_build_when_committed_image_present(self) -> None: def test_skips_build_when_committed_image_present(self) -> None:
plan = _plan(self._tmp) built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=True)
built = self._run_launch(plan, committed_tag=_COMMITTED_TAG, image_present=True) self.assertEqual([], built) # committed image reused, no build
self.assertEqual([], built, "build_image should not be called when committed image exists")
def test_uses_committed_image_in_compose_spec(self) -> None: def test_uses_committed_image_in_compose_spec(self) -> None:
"""The compose spec renderer receives the committed image tag via captured: list[DockerBottlePlan] = []
plan.image — captured here by checking what bottle_plan_to_compose
was called with."""
plan = _plan(self._tmp)
captured_plans: list[DockerBottlePlan] = []
def fake_compose(p: DockerBottlePlan) -> dict[str, Any]: def compose(p: DockerBottlePlan, **_kw: Any) -> dict[str, Any]:
captured_plans.append(p) captured.append(p)
return {"services": {"agent": {}}} return {"services": {"agent": {}}}
with mock.patch.object( with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose):
launch_mod, "read_committed_image", return_value=_COMMITTED_TAG, with launch_mod.launch(_plan(self._tmp), provision=mock.Mock(return_value=None)):
), mock.patch.object(
launch_mod.docker_mod, "image_exists", return_value=True,
), mock.patch.object(
launch_mod.docker_mod, "build_image",
), mock.patch.object(
launch_mod, "egress_tls_init",
return_value=(Path("/egress_ca"), Path("/egress_cert")),
), mock.patch.object(
launch_mod.network_mod, "network_name_for_slug",
return_value="bb-internal",
), mock.patch.object(
launch_mod.network_mod, "network_egress_name_for_slug",
return_value="bb-egress",
), mock.patch.object(
launch_mod, "bottle_plan_to_compose", side_effect=fake_compose,
), mock.patch.object(
launch_mod, "write_compose_file",
return_value=Path("/tmp/compose.yml"),
), mock.patch.object(launch_mod, "compose_up"), \
mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()):
provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, provision=provision):
pass pass
self.assertEqual(_COMMITTED_TAG, captured[0].image)
self.assertEqual(1, len(captured_plans))
self.assertEqual(_COMMITTED_TAG, captured_plans[0].image)
def test_falls_back_to_build_when_no_committed_image(self) -> None: def test_falls_back_to_build_when_no_committed_image(self) -> None:
plan = _plan(self._tmp) self.assertEqual([_DEFAULT_IMAGE], self._run_launch(_plan(self._tmp), committed_tag=None))
built = self._run_launch(plan, committed_tag=None)
self.assertEqual([_DEFAULT_IMAGE], built)
def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None: def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
plan = _plan(self._tmp) built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=False)
built = self._run_launch(
plan, committed_tag=_COMMITTED_TAG, image_present=False,
)
self.assertEqual([_DEFAULT_IMAGE], built) self.assertEqual([_DEFAULT_IMAGE], built)
+17 -20
View File
@@ -19,9 +19,11 @@ from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import BottleSpec from bot_bottle.backend import BottleSpec
from bot_bottle.backend.docker import launch as launch_mod from bot_bottle.backend.docker import launch as launch_mod
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.docker.consolidated_launch import LaunchContext
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
from tests.unit import use_bottle_root
_INDEX = ManifestIndex.from_json_obj({ _INDEX = ManifestIndex.from_json_obj({
"bottles": {"dev": {}}, "bottles": {"dev": {}},
@@ -77,41 +79,36 @@ def _plan(tmp: str) -> DockerBottlePlan:
class TestTeardownWarning(unittest.TestCase): class TestTeardownWarning(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self._tmp = tempfile.mkdtemp(prefix="docker-launch-teardown-test.") self._tmp = tempfile.mkdtemp(prefix="docker-launch-teardown-test.")
self.addCleanup(lambda: __import__("shutil").rmtree(self._tmp, ignore_errors=True))
def tearDown(self) -> None: self.addCleanup(use_bottle_root(Path(self._tmp))) # sandbox the gateway CA write
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
def test_teardown_failure_emits_warning_with_container_and_operation(self): def test_teardown_failure_emits_warning_with_container_and_operation(self):
plan = _plan(self._tmp) plan = _plan(self._tmp)
buf = io.StringIO() buf = io.StringIO()
gw = mock.Mock()
gw.ca_cert_pem.return_value = "-----BEGIN CERTIFICATE-----\nx\n-----END CERTIFICATE-----\n"
ctx = LaunchContext(
bottle_id="b1", identity_token="t", source_ip="172.20.0.4",
network="bot-bottle-gateway", gateway_ip="172.20.0.2",
orchestrator_url="http://orch:8099",
)
with mock.patch.object(launch_mod.docker_mod, "build_image"), \ with mock.patch.object(launch_mod.docker_mod, "build_image"), \
mock.patch.object(launch_mod, "launch_consolidated", return_value=ctx), \
mock.patch.object(launch_mod, "teardown_consolidated"), \
mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
mock.patch.object( mock.patch.object(
launch_mod, "egress_tls_init", launch_mod, "consolidated_agent_compose",
return_value=(Path("/egress_ca"), Path("/egress_cert")),
), \
mock.patch.object(
launch_mod.network_mod, "network_name_for_slug",
return_value="bb-internal-test",
), \
mock.patch.object(
launch_mod.network_mod, "network_egress_name_for_slug",
return_value="bb-egress-test",
), \
mock.patch.object(
launch_mod, "bottle_plan_to_compose",
return_value={"services": {"agent": {}}}, return_value={"services": {"agent": {}}},
), \ ), \
mock.patch.object( mock.patch.object(
launch_mod, "write_compose_file", launch_mod, "write_compose_file", return_value=Path("/tmp/compose.yml"),
return_value=Path("/tmp/compose.yml"),
), \ ), \
mock.patch.object(launch_mod, "compose_up"), \ mock.patch.object(launch_mod, "compose_up"), \
mock.patch.object(launch_mod, "compose_dump_logs"), \ mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object( mock.patch.object(
launch_mod, "compose_down", launch_mod, "compose_down",
side_effect=RuntimeError("network remove failed"), side_effect=RuntimeError("compose down failed"),
), \ ), \
contextlib.redirect_stderr(buf): contextlib.redirect_stderr(buf):
provision = mock.Mock(return_value=None) provision = mock.Mock(return_value=None)
+8 -6
View File
@@ -31,9 +31,9 @@ def _recorder(calls: list[list[str]]):
def _plan(*upstreams: GitGateUpstream) -> GitGatePlan: def _plan(*upstreams: GitGateUpstream) -> GitGatePlan:
return GitGatePlan( return GitGatePlan(
slug="demo", slug="demo",
entrypoint_script=Path(), entrypoint_script=Path("/stage/entrypoint.sh"),
hook_script=Path(), hook_script=Path("/stage/pre-receive"),
access_hook_script=Path(), access_hook_script=Path("/stage/access-hook"),
upstreams=tuple(upstreams), upstreams=tuple(upstreams),
) )
@@ -61,6 +61,8 @@ class TestProvisionGitGate(unittest.TestCase):
self.assertIn( self.assertIn(
["docker", "cp", "/host/kh", "gw:/git-gate/creds/bottle1/foo-known_hosts"], cps, ["docker", "cp", "/host/kh", "gw:/git-gate/creds/bottle1/foo-known_hosts"], cps,
) )
# The shared (bottle-agnostic) hooks are installed into the gateway.
self.assertIn(["docker", "cp", "/stage/pre-receive", "gw:/etc/git-gate/pre-receive"], cps)
# The init script runs in the gateway, namespaced under the bottle id. # The init script runs in the gateway, namespaced under the bottle id.
exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"] exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"]
self.assertEqual(1, len(exec_scripts)) self.assertEqual(1, len(exec_scripts))
@@ -70,9 +72,9 @@ class TestProvisionGitGate(unittest.TestCase):
calls: list[list[str]] = [] calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)): with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "b1", _plan(_up("foo"))) # no known_hosts provision_git_gate("gw", "b1", _plan(_up("foo"))) # no known_hosts
cps = [c for c in calls if c[:2] == ["docker", "cp"]] creds_cps = [c for c in calls if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]]
self.assertEqual(1, len(cps)) # only the key, not known_hosts self.assertEqual(1, len(creds_cps)) # only the key, not known_hosts
self.assertTrue(cps[0][3].endswith("/foo-key")) self.assertTrue(creds_cps[0][3].endswith("/foo-key"))
def test_no_upstreams_is_noop(self) -> None: def test_no_upstreams_is_noop(self) -> None:
with patch(_RUN) as m: with patch(_RUN) as m:
+21 -19
View File
@@ -129,28 +129,35 @@ class TestDockerGatewayBuild(unittest.TestCase):
with patch(_RUN_DOCKER, return_value=_proc(returncode=1)): with patch(_RUN_DOCKER, return_value=_proc(returncode=1)):
self.assertFalse(self.sc.image_exists()) self.assertFalse(self.sc.image_exists())
def test_ensure_built_is_noop_when_image_present(self) -> None: def test_ensure_built_builds_even_when_image_present(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=0)) as m: # Always build (cache-aware) so a flat-source change rebuilds; the old
self.sc.ensure_built() # build-if-missing silently ran a stale single-tenant image.
self.assertEqual(1, m.call_count) # only the image-inspect probe
def test_ensure_built_builds_from_dockerfile_when_missing(self) -> None:
calls: list[list[str]] = [] calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock: def rec(argv: list[str]) -> Mock:
calls.append(argv) calls.append(argv)
if argv[:3] == ["docker", "image", "inspect"]: return _proc() # image present, build succeeds
return _proc(returncode=1) # missing
return _proc()
with patch(_RUN_DOCKER, side_effect=fake): with patch(_RUN_DOCKER, side_effect=rec):
self.sc.ensure_built() self.sc.ensure_built()
builds = [c for c in calls if c[:2] == ["docker", "build"]] builds = [c for c in calls if c[:2] == ["docker", "build"]]
self.assertEqual(1, len(builds)) self.assertEqual(1, len(builds))
self.assertIn(self.sc.image_ref, builds[0]) self.assertIn(self.sc.image_ref, builds[0])
self.assertIn("-f", builds[0])
self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0])) self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0]))
self.assertNotIn("--no-cache", builds[0])
def test_ensure_built_no_cache_env_forces_full_rebuild(self) -> None:
calls: list[list[str]] = []
def rec(argv: list[str]) -> Mock:
calls.append(argv)
return _proc()
with patch(_RUN_DOCKER, side_effect=rec), \
patch.dict("os.environ", {"BOT_BOTTLE_NO_CACHE": "1"}):
self.sc.ensure_built()
builds = [c for c in calls if c[:2] == ["docker", "build"]]
self.assertIn("--no-cache", builds[0])
def test_ensure_built_noop_when_no_dockerfile(self) -> None: def test_ensure_built_noop_when_no_dockerfile(self) -> None:
sc = DockerGateway("busybox", dockerfile=None) sc = DockerGateway("busybox", dockerfile=None)
@@ -159,12 +166,7 @@ class TestDockerGatewayBuild(unittest.TestCase):
m.assert_not_called() m.assert_not_called()
def test_ensure_built_raises_on_build_failure(self) -> None: def test_ensure_built_raises_on_build_failure(self) -> None:
def fake(argv: list[str]) -> Mock: with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="build boom")):
if argv[:3] == ["docker", "image", "inspect"]:
return _proc(returncode=1)
return _proc(returncode=1, stderr="build boom")
with patch(_RUN_DOCKER, side_effect=fake):
with self.assertRaises(GatewayError): with self.assertRaises(GatewayError):
self.sc.ensure_built() self.sc.ensure_built()
+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 from __future__ import annotations
@@ -6,81 +6,81 @@ import tempfile
import unittest import unittest
import urllib.error import urllib.error
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, Mock, patch
from bot_bottle.orchestrator.lifecycle import ( from bot_bottle.orchestrator.lifecycle import (
OrchestratorProcess, ORCHESTRATOR_NAME,
OrchestratorService,
OrchestratorStartError, OrchestratorStartError,
) )
from tests.unit import use_bottle_root from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen" _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" _SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
_MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic" _MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
def _health(status: int) -> MagicMock: def _health(status: int) -> MagicMock:
"""A urlopen() context-manager whose `.status` is `status`."""
m = MagicMock() m = MagicMock()
m.__enter__.return_value.status = status m.__enter__.return_value.status = status
return m return m
class TestOrchestratorProcess(unittest.TestCase): class TestOrchestratorService(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory() self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup) self.addCleanup(self._tmp.cleanup)
self.addCleanup(use_bottle_root(Path(self._tmp.name))) self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.p = OrchestratorProcess(port=8099) self.svc = OrchestratorService(port=8099)
def test_url(self) -> None: def test_urls(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.p.url) 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)): with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.p.is_healthy()) self.assertTrue(self.svc.is_healthy())
def test_is_healthy_false_on_error(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): 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: def test_ensure_running_healthy_still_ensures_gateway_but_not_orchestrator(self) -> None:
with patch(_URLOPEN, return_value=_health(200)), patch(_POPEN) as popen: with patch(_URLOPEN, return_value=_health(200)), \
self.assertEqual(self.p.url, self.p.ensure_running()) patch(_GATEWAY) as gw_cls, patch(_RUN) as run:
popen.assert_not_called() # a live control plane is left untouched 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: def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
# First check (before spawn) fails; after spawn the poll succeeds. run = Mock(return_value=Mock(returncode=0, stderr=""))
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \ with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_POPEN) as popen, patch(_SLEEP): patch(_GATEWAY), patch(_RUN, run), patch(_SLEEP):
url = self.p.ensure_running() self.assertEqual(self.svc.url, self.svc.ensure_running())
self.assertEqual(self.p.url, url) runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
popen.assert_called_once() self.assertEqual(1, len(runs))
argv = runs[0]
def test_ensure_running_raises_on_startup_timeout(self) -> None: self.assertIn(ORCHESTRATOR_NAME, argv)
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \ self.assertIn("--broker", argv)
patch(_POPEN), patch(_SLEEP), \ self.assertEqual("stub", argv[argv.index("--broker") + 1]) # register-only, no socket
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.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: def test_ensure_running_raises_on_timeout(self) -> None:
self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv()) 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: def test_stop_removes_orchestrator_and_gateway(self) -> None:
with patch(_POPEN) as popen: with patch(_RUN) as run, patch(_GATEWAY) as gw_cls:
self.p._spawn() self.svc.stop()
popen.assert_called_once() rms = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "rm", "--force"]]
kwargs = popen.call_args.kwargs self.assertTrue(any(ORCHESTRATOR_NAME in a for a in rms))
self.assertTrue(kwargs["start_new_session"]) # outlives the CLI gw_cls.return_value.stop.assert_called_once()
self.assertTrue((Path(self._tmp.name) / "orchestrator.log").exists())
if __name__ == "__main__": if __name__ == "__main__":