Compare commits

...

5 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
didericis eab7d6ccbc feat(orchestrator): slice 13c(viii) — agent-only consolidated compose render
test / unit (pull_request) Successful in 1m2s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m18s
lint / lint (push) Successful in 2m8s
The per-bottle model rendered a compose project with the agent + a sidecar
bundle on two per-bottle networks. Consolidated has no sidecars — one shared
gateway serves everyone — so this renders JUST the agent, attached to the
external shared gateway network at the orchestrator-allocated pinned IP, and
pointed at the gateway's address for egress.

- consolidated_agent_compose(plan, *, gateway_ip, source_ip, network):
  agent service only; networks {<gateway>: {ipv4_address: source_ip}} on the
  external gateway network; HTTPS_PROXY -> http://<gateway_ip>:9099; NO_PROXY
  includes the gateway address so git-http + supervise (also on the gateway)
  bypass the proxy and are reached directly. No depends_on/sidecars.
- Pure (takes the launch-time LaunchContext values), so it's unit-testable
  without docker.

pyright 0 errors; pylint 9.83/10; unit suite green (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 22:17:11 -04:00
17 changed files with 542 additions and 385 deletions
+5
View File
@@ -111,6 +111,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
plumbing needed; the alias resolves inside the bridge."""
if plan.supervise_plan is None:
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}/"
def prepare_cleanup(self) -> DockerBottleCleanupPlan:
+19
View File
@@ -28,11 +28,30 @@ class DockerBottlePlan(BottlePlan):
# accidental log of the plan dataclass.
forwarded_env: dict[str, str] = field(repr=False)
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
def container_name(self) -> str:
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
def image(self) -> str:
return self.agent_provision.image
@@ -0,0 +1,78 @@
"""Agent-only compose for the consolidated docker backend (PRD 0070).
The per-bottle model rendered a compose project with the agent *and* a
sidecar bundle on two per-bottle networks. In the consolidated model the
sidecars are gone — one shared gateway serves every bottle — so this renders
just the agent, attached to the **external shared gateway network** with the
pinned source IP the orchestrator allocated, and pointed at the gateway's
address for egress (and, around the proxy, for git-http / supervise).
Pure: it takes the launch-time `LaunchContext` values (gateway address,
source IP, network) and the prepared plan, and returns a compose dict — no
docker, so it's testable in isolation.
"""
from __future__ import annotations
from typing import Any
from ...egress import egress_agent_env_entries
from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
from .bottle_plan import DockerBottlePlan
from .egress import EGRESS_PORT
def consolidated_agent_compose(
plan: DockerBottlePlan,
*,
gateway_ip: str,
source_ip: str,
network: str,
) -> dict[str, Any]:
"""A compose spec with only the agent service, on the external gateway
network at `source_ip`, proxying egress through `gateway_ip`."""
proxy_url = f"http://{gateway_ip}:{EGRESS_PORT}"
# git-http + supervise live on the gateway too and must NOT go through the
# egress proxy — the agent reaches them directly by the gateway address.
no_proxy = f"localhost,127.0.0.1,{gateway_ip}"
env: list[str] = [
f"HTTPS_PROXY={proxy_url}",
f"HTTP_PROXY={proxy_url}",
f"https_proxy={proxy_url}",
f"http_proxy={proxy_url}",
f"NO_PROXY={no_proxy}",
f"no_proxy={no_proxy}",
f"NODE_EXTRA_CA_CERTS={AGENT_CA_PATH}",
f"SSL_CERT_FILE={AGENT_CA_BUNDLE}",
f"REQUESTS_CA_BUNDLE={AGENT_CA_BUNDLE}",
]
for name, value in sorted(plan.agent_provision.guest_env.items()):
env.append(f"{name}={value}")
# Forwarded vars: bare name → inherits from the compose-up process env so
# the secret value never lands on argv or in the compose file.
for name in sorted(plan.forwarded_env.keys()):
env.append(name)
env.extend(egress_agent_env_entries(plan.egress_plan))
service: dict[str, Any] = {
"image": plan.image,
"container_name": plan.container_name,
"command": ["sleep", "infinity"],
# Pinned address on the shared gateway network — the orchestrator
# registered this IP, and the gateway attributes the bottle by it.
"networks": {network: {"ipv4_address": source_ip}},
"environment": env,
}
if plan.use_runsc:
service["runtime"] = "runsc"
return {
"name": f"bot-bottle-{plan.slug}",
"services": {"agent": service},
# The gateway network is created + owned by the orchestrator; compose
# attaches to it (external) and must not create or destroy it.
"networks": {network: {"external": True}},
}
__all__ = ["consolidated_agent_compose"]
@@ -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
@@ -76,15 +76,21 @@ def _container_ip(name: str, network: str) -> str:
return ip
def _taken_ips(client: OrchestratorClient, gateway_ip: str) -> list[str]:
"""Every address already in use on the gateway network: the gateway
container plus every live bottle the registry knows."""
taken = [gateway_ip]
for rec in client.list_bottles():
src = rec.get("source_ip")
if isinstance(src, str) and src:
taken.append(src)
return taken
def _network_container_ips(network: str) -> list[str]:
"""Every address currently assigned on the gateway network the ground
truth for "in use": the gateway + orchestrator infrastructure containers
and every live agent. Read from the network so a new bottle can't collide
with anything actually attached (a registry-only view would miss the
orchestrator/gateway containers)."""
proc = run_docker([
"docker", "network", "inspect", "--format",
"{{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(
@@ -92,7 +98,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,13 +106,13 @@ 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)
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)
reg = client.register_bottle(
@@ -67,6 +67,12 @@ def provision_git_gate(gateway: str, bottle_id: str, plan: GitGatePlan) -> None:
_require_safe(bottle_id)
if not plan.upstreams:
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)
_exec(gateway, ["mkdir", "-p", creds])
for u in plan.upstreams:
+47 -62
View File
@@ -36,13 +36,11 @@ from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import Callable, Generator
from ...egress import egress_resolve_token_values
from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import info, warn
from . import network as network_mod
from . import util as docker_mod
from .bottle import DockerBottle
from .bottle_plan import DockerBottlePlan
@@ -53,7 +51,6 @@ from ...bottle_state import (
read_committed_image,
)
from .compose import (
bottle_plan_to_compose,
compose_down,
compose_dump_logs,
compose_file_path,
@@ -62,7 +59,9 @@ from .compose import (
compose_up,
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.
@@ -97,8 +96,7 @@ def launch(
try:
# Step 1: agent image. Use a committed snapshot when one exists
# and is present in the local daemon; otherwise build from the
# Dockerfile. Sidecar images get built lazily by `docker compose
# up` via the renderer's `build:` directives.
# Dockerfile. (The gateway image is built by the orchestrator.)
committed = read_committed_image(plan.slug)
if committed and docker_mod.image_exists(committed):
info(f"using committed image {committed!r}")
@@ -112,82 +110,73 @@ def launch(
dockerfile=plan.dockerfile_path,
)
internal_network = network_mod.network_name_for_slug(plan.slug)
egress_network = network_mod.network_egress_name_for_slug(plan.slug)
egress_ca_host, egress_ca_cert_only = egress_tls_init(
egress_state_dir(plan.slug),
)
# Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
# provisioning the bottle's repos into the shared gateway.
git_gate_plan = plan.git_gate_plan
if git_gate_plan.upstreams:
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
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,
plan.manifest.bottle, git_gate_plan, git_gate_state_dir(plan.slug),
)
# 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(
plan.egress_plan,
internal_network=internal_network,
egress_network=egress_network,
mitmproxy_ca_host_path=egress_ca_host,
mitmproxy_ca_cert_only_host_path=egress_ca_cert_only,
mitmproxy_ca_host_path=ca_file,
mitmproxy_ca_cert_only_host_path=ca_file,
)
# 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,
git_gate_plan=git_gate_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
# was written at prepare time and already carries
# compose_project; nothing to update here.
# Step 5: render + up the agent-only compose, pinned on the shared
# gateway network and proxied through the gateway's address.
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))
project = compose_project_name(plan.slug)
# Step 7: compose up. Token values + the OAuth placeholder
# flow through subprocess env; the compose file holds only
# 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,
}
# Forwarded vars (OAuth token, host interpolations) flow through the
# subprocess env as bare names so values never land in the file.
compose_env: dict[str, str] = {**os.environ, **plan.forwarded_env}
info(
f"docker compose up -d (project {project}, "
f"{len(spec['services'])} services)"
f"docker compose up -d (project {project}, agent on shared "
f"gateway {ctx.gateway_ip}, ip {ctx.source_ip})"
)
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_dump_logs, project, compose_file, compose_log_path(state_dir),
)
# Step 8: provision. Create the bottle first so provisioners
# can use bottle.exec / bottle.cp_in; set the prompt path
# returned by provision_prompt after the fact.
# Step 6: provision (CA install now uses the gateway CA) + yield.
bottle = DockerBottle(
plan.container_name,
teardown,
@@ -200,10 +189,6 @@ def launch(
agent_workdir=plan.workspace_plan.workdir,
)
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
finally:
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).
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
+11 -2
View File
@@ -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")
+29 -13
View File
@@ -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
@@ -104,17 +109,23 @@ class DockerGateway(Gateway):
return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0
def ensure_built(self) -> None:
"""Build the bundle image from its Dockerfile when it's missing.
No-op when the image is already present, or when no dockerfile is
configured (e.g. a pre-pulled image)."""
if self._dockerfile is None or self.image_exists():
"""Build the bundle image from its Dockerfile, **cache-aware** — cheap
(a cache check) when nothing changed, a real rebuild when the flat
sources (egress addon / git-http / policy_resolver / supervise) moved.
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
proc = run_docker([
"docker", "build",
"-t", self.image_ref,
"-f", str(self._build_context / self._dockerfile),
str(self._build_context),
])
argv = ["docker", "build", "-t", self.image_ref,
"-f", str(self._build_context / self._dockerfile),
str(self._build_context)]
if os.environ.get("BOT_BOTTLE_NO_CACHE"):
argv.insert(2, "--no-cache")
proc = run_docker(argv)
if proc.returncode != 0:
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
# 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 +165,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()}")
+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
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",
]
+51
View File
@@ -0,0 +1,51 @@
"""Unit: agent-only consolidated compose render (PRD 0070)."""
from __future__ import annotations
import unittest
from bot_bottle.backend.docker.consolidated_compose import consolidated_agent_compose
from tests.unit.test_compose import _plan
_GW = "172.18.0.2"
_IP = "172.18.0.5"
_NET = "bot-bottle-gateway"
class TestConsolidatedAgentCompose(unittest.TestCase):
def _spec(self, *, runsc: bool = False):
plan = _plan(with_egress=True, supervise=True, with_git=True)
if runsc:
plan = type(plan)(**{**vars(plan), "use_runsc": True}) # type: ignore[arg-type]
return consolidated_agent_compose(plan, gateway_ip=_GW, source_ip=_IP, network=_NET)
def test_only_agent_service_no_sidecars(self) -> None:
# The whole point of consolidation: no per-bottle sidecar bundle.
self.assertEqual(["agent"], list(self._spec()["services"]))
def test_agent_pinned_on_external_gateway_network(self) -> None:
spec = self._spec()
self.assertEqual({"external": True}, spec["networks"][_NET])
agent_net = spec["services"]["agent"]["networks"][_NET]
self.assertEqual(_IP, agent_net["ipv4_address"])
def test_proxy_and_ca_point_at_gateway(self) -> None:
env = self._spec()["services"]["agent"]["environment"]
self.assertIn(f"HTTPS_PROXY=http://{_GW}:9099", env)
# git-http + supervise on the gateway must bypass the egress proxy.
self.assertTrue(any(e.startswith("NO_PROXY=") and _GW in e for e in env))
def test_no_sidecar_dependency(self) -> None:
self.assertNotIn("depends_on", self._spec()["services"]["agent"])
def test_runsc_runtime_when_enabled(self) -> None:
self.assertEqual("runsc", self._spec(runsc=True)["services"]["agent"]["runtime"])
def test_forwarded_env_stays_bare_names(self) -> None:
env = self._spec()["services"]["agent"]["environment"]
# forwarded secrets are bare names (value inherited from process env).
self.assertIn("CLAUDE_CODE_OAUTH_TOKEN", env)
if __name__ == "__main__":
unittest.main()
+12 -8
View File
@@ -39,14 +39,18 @@ 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"
def _run(
self, client: Mock, provision: Mock | None = None,
*, 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"), \
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}.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()
@@ -64,10 +68,10 @@ class TestLaunchConsolidated(unittest.TestCase):
self.assertIn("api.example.com", kwargs.kwargs["policy"])
provision.assert_called_once()
def test_skips_gateway_and_live_bottle_addresses(self) -> None:
client = _client(bottles=[{"source_ip": "172.18.0.3"}])
ctx = self._run(client)
self.assertEqual("172.18.0.4", ctx.source_ip) # .2 gw, .3 taken → .4
def test_skips_all_addresses_on_the_network(self) -> None:
# Gateway .2 + orchestrator .3 already attached -> agent gets .4.
ctx = self._run(_client(), on_network=("172.18.0.2", "172.18.0.3"))
self.assertEqual("172.18.0.4", ctx.source_ip)
def test_provision_failure_rolls_back_registration(self) -> None:
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
@@ -6,6 +6,7 @@ import contextlib
import io
import tempfile
import unittest
from collections.abc import Callable, Generator
from pathlib import Path
from typing import Any
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.docker import launch as launch_mod
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.git_gate import GitGatePlan
from bot_bottle.manifest import ManifestIndex
from tests.unit import use_bottle_root
_SLUG = "dev-abc12"
@@ -28,163 +31,111 @@ _IDX = ManifestIndex.from_json_obj({
"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:
stage = Path(tmp)
spec = BottleSpec(
manifest=_IDX,
agent_name="demo",
copy_cwd=False,
user_cwd=tmp,
identity=_SLUG,
manifest=_IDX, agent_name="demo", copy_cwd=False, user_cwd=tmp, identity=_SLUG,
)
return DockerBottlePlan(
spec=spec,
manifest=_IDX.load_for_agent("demo"),
stage_dir=stage,
git_gate_plan=GitGatePlan(
slug=_SLUG,
entrypoint_script=stage / "entrypoint.sh",
hook_script=stage / "hook.sh",
access_hook_script=stage / "access-hook.sh",
upstreams=(),
slug=_SLUG, entrypoint_script=stage / "e.sh", hook_script=stage / "h.sh",
access_hook_script=stage / "a.sh", upstreams=(),
),
egress_plan=EgressPlan(
slug=_SLUG,
routes_path=stage / "egress.yaml",
routes=(),
token_env_map={},
slug=_SLUG, routes_path=stage / "egress.yaml", routes=(), token_env_map={},
),
supervise_plan=None,
agent_provision=AgentProvisionPlan(
template="claude",
command="claude",
prompt_mode="append_file",
image=_DEFAULT_IMAGE,
dockerfile="",
guest_home="/home/node",
instance_name=f"bot-bottle-{_SLUG}",
prompt_file=stage / "prompt.txt",
template="claude", command="claude", prompt_mode="append_file",
image=_DEFAULT_IMAGE, dockerfile="", guest_home="/home/node",
instance_name=f"bot-bottle-{_SLUG}", prompt_file=stage / "prompt.txt",
guest_env={},
),
slug=_SLUG,
forwarded_env={},
use_runsc=False,
slug=_SLUG, forwarded_env={}, use_runsc=False,
)
class TestLaunchCommittedImage(unittest.TestCase):
def setUp(self) -> None:
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:
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
def _run_launch(
@contextlib.contextmanager
def _patched(
self,
plan: DockerBottlePlan,
*,
committed_tag: str | None = None,
image_present: bool = True,
) -> list[str]:
"""Drive launch() through its full sequence with the committed-image
behaviour controlled by the arguments. Returns the images that were
passed to `build_image` (empty list if it was never called)."""
committed_tag: str | None,
image_present: bool,
compose: Callable[..., dict[str, Any]],
) -> Generator[list[str], None, None]:
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
built.append(image)
with mock.patch.object(
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(
launch_mod.docker_mod, "build_image", side_effect=fake_build,
), 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",
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
with mock.patch.object(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(launch_mod.docker_mod, "build_image", side_effect=_build), \
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(launch_mod, "consolidated_agent_compose", side_effect=compose), \
mock.patch.object(launch_mod, "write_compose_file", return_value=Path("/tmp/c.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()):
yield built
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
def test_skips_build_when_committed_image_present(self) -> None:
plan = _plan(self._tmp)
built = self._run_launch(plan, committed_tag=_COMMITTED_TAG, image_present=True)
self.assertEqual([], built, "build_image should not be called when committed image exists")
built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=True)
self.assertEqual([], built) # committed image reused, no build
def test_uses_committed_image_in_compose_spec(self) -> None:
"""The compose spec renderer receives the committed image tag via
plan.image — captured here by checking what bottle_plan_to_compose
was called with."""
plan = _plan(self._tmp)
captured_plans: list[DockerBottlePlan] = []
captured: list[DockerBottlePlan] = []
def fake_compose(p: DockerBottlePlan) -> dict[str, Any]:
captured_plans.append(p)
def compose(p: DockerBottlePlan, **_kw: Any) -> dict[str, Any]:
captured.append(p)
return {"services": {"agent": {}}}
with mock.patch.object(
launch_mod, "read_committed_image", return_value=_COMMITTED_TAG,
), 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):
with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose):
with launch_mod.launch(_plan(self._tmp), provision=mock.Mock(return_value=None)):
pass
self.assertEqual(1, len(captured_plans))
self.assertEqual(_COMMITTED_TAG, captured_plans[0].image)
self.assertEqual(_COMMITTED_TAG, captured[0].image)
def test_falls_back_to_build_when_no_committed_image(self) -> None:
plan = _plan(self._tmp)
built = self._run_launch(plan, committed_tag=None)
self.assertEqual([_DEFAULT_IMAGE], built)
self.assertEqual([_DEFAULT_IMAGE], self._run_launch(_plan(self._tmp), committed_tag=None))
def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
plan = _plan(self._tmp)
built = self._run_launch(
plan, committed_tag=_COMMITTED_TAG, image_present=False,
)
built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=False)
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.docker import launch as launch_mod
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.git_gate import GitGatePlan
from bot_bottle.manifest import ManifestIndex
from tests.unit import use_bottle_root
_INDEX = ManifestIndex.from_json_obj({
"bottles": {"dev": {}},
@@ -77,41 +79,36 @@ def _plan(tmp: str) -> DockerBottlePlan:
class TestTeardownWarning(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.mkdtemp(prefix="docker-launch-teardown-test.")
def tearDown(self) -> None:
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
self.addCleanup(lambda: __import__("shutil").rmtree(self._tmp, ignore_errors=True))
self.addCleanup(use_bottle_root(Path(self._tmp))) # sandbox the gateway CA write
def test_teardown_failure_emits_warning_with_container_and_operation(self):
plan = _plan(self._tmp)
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"), \
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(
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-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",
launch_mod, "consolidated_agent_compose",
return_value={"services": {"agent": {}}},
), \
mock.patch.object(
launch_mod, "write_compose_file",
return_value=Path("/tmp/compose.yml"),
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",
side_effect=RuntimeError("network remove failed"),
side_effect=RuntimeError("compose down failed"),
), \
contextlib.redirect_stderr(buf):
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:
return GitGatePlan(
slug="demo",
entrypoint_script=Path(),
hook_script=Path(),
access_hook_script=Path(),
entrypoint_script=Path("/stage/entrypoint.sh"),
hook_script=Path("/stage/pre-receive"),
access_hook_script=Path("/stage/access-hook"),
upstreams=tuple(upstreams),
)
@@ -61,6 +61,8 @@ class TestProvisionGitGate(unittest.TestCase):
self.assertIn(
["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.
exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"]
self.assertEqual(1, len(exec_scripts))
@@ -70,9 +72,9 @@ class TestProvisionGitGate(unittest.TestCase):
calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "b1", _plan(_up("foo"))) # no known_hosts
cps = [c for c in calls if c[:2] == ["docker", "cp"]]
self.assertEqual(1, len(cps)) # only the key, not known_hosts
self.assertTrue(cps[0][3].endswith("/foo-key"))
creds_cps = [c for c in calls if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]]
self.assertEqual(1, len(creds_cps)) # only the key, not known_hosts
self.assertTrue(creds_cps[0][3].endswith("/foo-key"))
def test_no_upstreams_is_noop(self) -> None:
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)):
self.assertFalse(self.sc.image_exists())
def test_ensure_built_is_noop_when_image_present(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=0)) as m:
self.sc.ensure_built()
self.assertEqual(1, m.call_count) # only the image-inspect probe
def test_ensure_built_builds_from_dockerfile_when_missing(self) -> None:
def test_ensure_built_builds_even_when_image_present(self) -> None:
# Always build (cache-aware) so a flat-source change rebuilds; the old
# build-if-missing silently ran a stale single-tenant image.
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def rec(argv: list[str]) -> Mock:
calls.append(argv)
if argv[:3] == ["docker", "image", "inspect"]:
return _proc(returncode=1) # missing
return _proc()
return _proc() # image present, build succeeds
with patch(_RUN_DOCKER, side_effect=fake):
with patch(_RUN_DOCKER, side_effect=rec):
self.sc.ensure_built()
builds = [c for c in calls if c[:2] == ["docker", "build"]]
self.assertEqual(1, len(builds))
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.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:
sc = DockerGateway("busybox", dockerfile=None)
@@ -159,12 +166,7 @@ class TestDockerGatewayBuild(unittest.TestCase):
m.assert_not_called()
def test_ensure_built_raises_on_build_failure(self) -> None:
def fake(argv: list[str]) -> Mock:
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 patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="build boom")):
with self.assertRaises(GatewayError):
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
@@ -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__":