Compare commits

...

9 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
didericis 389a2c10ce feat(orchestrator): slice 13c(vii) — stable shared gateway CA
lint / lint (push) Successful in 2m4s
test / unit (pull_request) Successful in 1m2s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m13s
The consolidated gateway TLS-intercepts every bottle, so all agents must
trust ONE CA. Rather than host-generate + mount a CA, let the gateway's own
mitmproxy generate its CA into a persistent named volume and extract the
cert for agents — no host openssl, keeps gateway.py lean.

- DockerGateway mounts GATEWAY_CA_VOLUME at /home/mitmproxy/.mitmproxy so the
  self-generated CA survives container recreation (it must not rotate, or
  already-launched agents would stop trusting the gateway).
- ca_cert_pem(): read the CA cert (PEM) out of the running gateway
  (docker exec cat mitmproxy-ca-cert.pem) — the agent installs this into its
  trust store to accept the gateway's bumped certs.

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:12:18 -04:00
didericis 9896986810 feat(orchestrator): slice 13c(vi) — consolidated launch sequence (composition)
test / unit (pull_request) Successful in 1m3s
test / integration (pull_request) Successful in 21s
test / coverage (pull_request) Successful in 1m12s
lint / lint (push) Successful in 2m5s
Composes the orchestrator primitives into the register/teardown sequence
that replaces the per-bottle bundle:

  launch_consolidated(egress_plan, git_gate_plan): ensure orchestrator +
    gateway up -> read the gateway network's subnet + the gateway's own
    address -> allocate the bottle a pinned source IP (skip the gateway +
    live bottles) -> register (egress policy blob + slug metadata) ->
    provision its git-gate repos/creds into the gateway. Returns a
    LaunchContext (bottle_id, identity_token, source_ip, network, gateway_ip,
    orchestrator_url) — everything the agent container needs to attach.
    Rolls the registration back if provisioning fails, so no orphan.

  teardown_consolidated(bottle_id): deregister + deprovision (idempotent).

The agent `docker run` itself stays the backend's job (provider
provisioning); this owns the orchestrator-facing wiring so the sequence is
unit-testable — collaborators mocked, next_free_ip + registration_inputs run
for real (IP allocation + policy blob asserted end to end).

pyright 0 errors; pylint 9.83/10; unit suite green (1755 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 22:03:21 -04:00
didericis 20e83258bf feat(git-gate+orchestrator): slice 13c(v) — provision git-gate into the running gateway
lint / lint (push) Successful in 2m16s
test / unit (pull_request) Successful in 1m8s
test / integration (pull_request) Successful in 29s
test / coverage (pull_request) Successful in 1m21s
Places a bottle's git-gate credentials + bare repos into the live shared
gateway container when it registers — the execution side of the 13c(i)
provisioning render.

- provision_git_gate(gateway, bottle_id, plan): mkdir the per-bottle creds
  dir, docker cp each upstream's identity key (+ known_hosts when present)
  into /git-gate/creds/<bottle_id>/, then docker exec the namespaced,
  init-only script from git_gate_render_provision. No-op with no upstreams.
- deprovision_git_gate(gateway, bottle_id): rm the bottle's /git/<id> +
  creds on teardown (idempotent).
- bottle_id is validated (shell/path-safe alphabet) BEFORE any docker cp/rm,
  so a traversal id can never reach a path arg — the creds/repo dirs land in
  docker cp/rm arguments. Registry ids are token_hex; defense in depth.

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 21:59:12 -04:00
didericis c5ba33bf0a feat(orchestrator): slice 13c(iv) — gateway joins a shared network
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m15s
lint / lint (push) Successful in 2m18s
The consolidated gateway ran on the default bridge; the model needs it on a
dedicated user-defined network that every agent bottle also joins, reaching
the gateway's egress / git-http / supervise ports by its address (no host
port publishing) — and the source IP the gateway attributes by is the
bottle's address on this network.

- GATEWAY_NETWORK = "bot-bottle-gateway"; DockerGateway gains a `network`
  param.
- ensure_running now ensures the network exists (idempotent create,
  tolerating a concurrent 'already exists') then runs with `--network`.
- Docker picks the subnet; the launcher reads it back (13c source-IP
  allocator) to hand each bottle a pinned address.

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 21:55:10 -04:00
17 changed files with 1045 additions and 358 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
@@ -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"]
@@ -0,0 +1,151 @@
"""Consolidated bottle launch sequence for the docker backend (PRD 0070).
Composes the orchestrator primitives into the register/teardown sequence that
replaces the per-bottle sidecar bundle:
1. ensure the orchestrator control plane + shared gateway are up;
2. allocate the bottle a pinned source IP on the gateway network (the
attribution key), skipping the gateway's own address + live bottles;
3. register it (egress policy blob + slug metadata) → bottle id + identity
token;
4. provision its git-gate repos/creds into the running gateway.
It returns a `LaunchContext` with everything the agent container needs to
attach — network, pinned IP, the gateway's address (its proxy target), the
orchestrator URL, and the identity token. The agent `docker run` itself is
the backend's job (it owns provider provisioning); this owns the
orchestrator-facing wiring so that sequence stays testable in isolation.
"""
from __future__ import annotations
from dataclasses import dataclass
from ...docker_cmd import run_docker
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 OrchestratorService
from ...orchestrator.registration import registration_inputs
from .gateway_net import next_free_ip
from .gateway_provision import deprovision_git_gate, provision_git_gate
class ConsolidatedLaunchError(RuntimeError):
"""The consolidated register/provision sequence could not complete."""
@dataclass(frozen=True)
class LaunchContext:
"""What the agent container needs to join the shared gateway."""
bottle_id: str
identity_token: str
source_ip: str # the agent's pinned address (attribution key)
network: str # the shared gateway network to attach to
gateway_ip: str # the gateway's address — the agent's proxy target
orchestrator_url: str
def _network_cidr(network: str) -> str:
"""The gateway network's IPv4 subnet, or raise."""
proc = run_docker([
"docker", "network", "inspect",
"--format", "{{range .IPAM.Config}}{{.Subnet}}{{end}}", network,
])
cidr = proc.stdout.strip()
if proc.returncode != 0 or not cidr:
raise ConsolidatedLaunchError(
f"gateway network {network} has no subnet: {proc.stderr.strip()}"
)
return cidr
def _container_ip(name: str, network: str) -> str:
"""A container's IPv4 address on `network`, or raise."""
proc = run_docker([
"docker", "inspect", "--format",
f'{{{{(index .NetworkSettings.Networks "{network}").IPAddress}}}}', name,
])
ip = proc.stdout.strip()
if proc.returncode != 0 or not ip:
raise ConsolidatedLaunchError(
f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
)
return ip
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(
egress_plan: EgressPlan,
git_gate_plan: GitGatePlan,
*,
image_ref: str = "",
service: OrchestratorService | None = None,
gateway_name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK,
) -> LaunchContext:
"""Ensure the orchestrator + gateway are up, allocate + register the
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."""
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, _network_container_ips(network))
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy, metadata=inputs.metadata,
)
try:
provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
except Exception:
# Roll the registration back so a provisioning failure leaves no orphan.
client.teardown_bottle(reg.bottle_id)
raise
return LaunchContext(
bottle_id=reg.bottle_id,
identity_token=reg.identity_token,
source_ip=source_ip,
network=network,
gateway_ip=gateway_ip,
orchestrator_url=url,
)
def teardown_consolidated(
bottle_id: str, *, orchestrator_url: str, gateway_name: str = GATEWAY_NAME,
) -> None:
"""Deregister the bottle and remove its git-gate state from the gateway.
Both steps are idempotent so this is safe from a cleanup trap."""
OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
deprovision_git_gate(gateway_name, bottle_id)
__all__ = [
"LaunchContext",
"launch_consolidated",
"teardown_consolidated",
"ConsolidatedLaunchError",
]
@@ -0,0 +1,100 @@
"""Provision one bottle's git-gate state into the running shared gateway
(PRD 0070, docker slice).
The consolidated gateway serves every bottle's repos under `/git/<bottle_id>/`
with per-repo credentials under `/git-gate/creds/<bottle_id>/`. When a bottle
is registered the launcher must place *its* deploy keys + known_hosts into
that per-bottle creds dir and init its bare repos there — so this copies the
credential files into the live gateway container and runs the (namespaced,
init-only) provisioning script produced by `git_gate_render_provision`.
Isolating each bottle's creds dir + repo root by id is what keeps one
bottle's push credentials out of another's repos on the shared gateway.
"""
from __future__ import annotations
import re
from ...docker_cmd import run_docker
from ...git_gate import GitGatePlan, git_gate_render_provision
# bottle ids index the gateway's per-bottle repo + creds dirs; they land in
# `docker cp`/`rm` path arguments, so validate before any path is built (a
# traversal id like "../etc" must never reach the container). Registry ids are
# token_hex — this is defense in depth at the docker boundary.
_SAFE_BOTTLE_ID = re.compile(r"[A-Za-z0-9_-]+")
class GatewayProvisionError(RuntimeError):
"""A git-gate provisioning step against the running gateway failed."""
def _require_safe(bottle_id: str) -> None:
if not _SAFE_BOTTLE_ID.fullmatch(bottle_id):
raise GatewayProvisionError(f"unsafe bottle id {bottle_id!r}")
def _creds_dir(bottle_id: str) -> str:
return f"/git-gate/creds/{bottle_id}"
def _exec(gateway: str, argv: list[str]) -> None:
"""`docker exec` a command in the gateway, raising on non-zero exit."""
proc = run_docker(["docker", "exec", gateway, *argv])
if proc.returncode != 0:
raise GatewayProvisionError(
f"gateway exec {argv!r} failed: {proc.stderr.strip()}"
)
def _cp_into(gateway: str, src: str, dest: str) -> None:
"""`docker cp` a host file into the gateway, raising on non-zero exit."""
proc = run_docker(["docker", "cp", src, f"{gateway}:{dest}"])
if proc.returncode != 0:
raise GatewayProvisionError(
f"gateway cp {src} -> {dest} failed: {proc.stderr.strip()}"
)
def provision_git_gate(gateway: str, bottle_id: str, plan: GitGatePlan) -> None:
"""Place `bottle_id`'s git-gate credentials into the running `gateway`
container and init its bare repos under `/git/<bottle_id>/`.
Copies each upstream's identity key (and known_hosts, when present) into
`/git-gate/creds/<bottle_id>/`, then runs the namespaced provisioning
script. No-op for a bottle with no git upstreams."""
_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:
if u.identity_file:
_cp_into(gateway, u.identity_file, f"{creds}/{u.name}-key")
known_hosts = str(u.known_hosts_file)
if known_hosts and known_hosts != ".":
_cp_into(gateway, known_hosts, f"{creds}/{u.name}-known_hosts")
# Init the bare repos + per-repo credential config for this namespace.
script = git_gate_render_provision(bottle_id, plan.upstreams)
_exec(gateway, ["sh", "-c", script])
def deprovision_git_gate(gateway: str, bottle_id: str) -> None:
"""Remove a bottle's repos + creds from the gateway on teardown. Idempotent
— an already-absent namespace is a clean no-op (best effort; a stray dir
can't leak, since attribution is by source IP and the bottle is gone)."""
_require_safe(bottle_id)
run_docker([
"docker", "exec", gateway, "rm", "-rf",
f"/git/{bottle_id}", _creds_dir(bottle_id),
])
__all__ = ["provision_git_gate", "deprovision_git_gate", "GatewayProvisionError"]
+45 -60
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, # Step 3: register on the orchestrator + provision this bottle's
internal_network=internal_network, # git-gate state into the shared gateway; get the agent's attach
egress_network=egress_network, # 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,
) )
supervise_plan = plan.supervise_plan # Point the agent's git-gate insteadOf rewrites at the shared gateway's
if supervise_plan is not None: # HTTP git endpoint (9420) instead of the dead per-bottle `git-gate`
supervise_plan = dataclasses.replace( # alias. git-http + supervise on the gateway bypass the egress proxy
supervise_plan, # (NO_PROXY includes the gateway address).
internal_network=internal_network, 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 ""
) )
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
+10 -1
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""
try:
status, payload = dispatch(server.orchestrator, method, self.path, body) 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")
+74 -13
View File
@@ -25,6 +25,19 @@ from ..docker_cmd import run_docker
GATEWAY_NAME = "bot-bottle-orch-gateway" GATEWAY_NAME = "bot-bottle-orch-gateway"
GATEWAY_LABEL = "bot-bottle-orch-gateway=1" GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
# The single user-defined network the gateway and every agent bottle share.
# Agents attach here with a pinned IP and reach the gateway's egress /
# git-http / supervise ports by its address — no host port publishing, and
# the source IP the gateway attributes by is the address on this network.
GATEWAY_NETWORK = "bot-bottle-gateway"
# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
# gateway's self-generated CA STABLE across container recreation — every agent
# installs this one CA to trust the shared gateway's TLS interception, so it
# must not rotate when the gateway restarts.
MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
# The real sidecar-bundle image + its Dockerfile. Kept as a local constant # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
# rather than imported from backend.docker.sidecar_bundle, which would drag # rather than imported from backend.docker.sidecar_bundle, which would drag
@@ -77,11 +90,18 @@ class DockerGateway(Gateway):
image_ref: str = GATEWAY_IMAGE, image_ref: str = GATEWAY_IMAGE,
*, *,
name: str = GATEWAY_NAME, name: str = GATEWAY_NAME,
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
# 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
@@ -89,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",
"-t", self.image_ref,
"-f", str(self._build_context / self._dockerfile), "-f", str(self._build_context / self._dockerfile),
str(self._build_context), 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: if proc.returncode != 0:
raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}") raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}")
@@ -112,21 +138,55 @@ class DockerGateway(Gateway):
]) ])
return self.name in proc.stdout.split() return self.name in proc.stdout.split()
def _ensure_network(self) -> None:
"""Create the shared gateway network if it doesn't exist. Idempotent —
a concurrent create loses harmlessly (the loser sees 'already exists').
Docker picks the subnet; the launcher reads it back to allocate IPs."""
if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
return
proc = run_docker(["docker", "network", "create", self.network])
if proc.returncode != 0 and "already exists" not in proc.stderr:
raise GatewayError(
f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
)
def ensure_running(self) -> None: def ensure_running(self) -> None:
if self.is_running(): if self.is_running():
return return
self._ensure_network()
# 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,
self.image_ref, "--network", self.network,
]) # Persist the self-generated CA so it survives restarts (agents
# trust it) — see GATEWAY_CA_VOLUME.
"--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
]
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()}")
def ca_cert_pem(self) -> str:
"""The gateway's CA certificate (PEM) that agents install to trust its
TLS interception. Read from the running container; raises if the
gateway hasn't generated it yet (it's created on first mitmproxy
start)."""
proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
if proc.returncode != 0 or not proc.stdout.strip():
raise GatewayError(
f"gateway CA cert not available: {proc.stderr.strip() or 'empty'}"
)
return proc.stdout
def stop(self) -> None: def stop(self) -> None:
proc = run_docker(["docker", "rm", "--force", self.name]) proc = run_docker(["docker", "rm", "--force", self.name])
if proc.returncode != 0 and "No such container" not in proc.stderr: if proc.returncode != 0 and "No such container" not in proc.stderr:
@@ -135,5 +195,6 @@ class DockerGateway(Gateway):
__all__ = [ __all__ = [
"Gateway", "DockerGateway", "GatewayError", "Gateway", "DockerGateway", "GatewayError",
"GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
"GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
] ]
+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",
] ]
+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()
+95
View File
@@ -0,0 +1,95 @@
"""Unit: consolidated launch sequence — compose the orchestrator primitives (PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from bot_bottle.backend.docker.consolidated_launch import (
launch_consolidated,
teardown_consolidated,
)
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.orchestrator.client import RegisteredBottle
_MOD = "bot_bottle.backend.docker.consolidated_launch"
def _egress_plan() -> EgressPlan:
return EgressPlan(
slug="demo", routes_path=Path("/x"), routes=(EgressRoute(host="api.example.com"),),
token_env_map={},
)
def _git_plan() -> GitGatePlan:
return GitGatePlan(
slug="demo", entrypoint_script=Path(), hook_script=Path(),
access_hook_script=Path(), upstreams=(),
)
def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
c = Mock()
c.list_bottles.return_value = bottles or []
c.register_bottle.return_value = RegisteredBottle("b1", "tok")
return c
class TestLaunchConsolidated(unittest.TestCase):
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(), service=service)
def test_allocates_ip_registers_and_provisions(self) -> None:
client = _client()
provision = Mock()
ctx = self._run(client, provision)
# .1 is the router, .2 is the gateway → first bottle gets .3.
self.assertEqual("172.18.0.3", ctx.source_ip)
self.assertEqual("172.18.0.2", ctx.gateway_ip)
self.assertEqual("b1", ctx.bottle_id)
self.assertEqual("tok", ctx.identity_token)
self.assertEqual("http://orch:8080", ctx.orchestrator_url)
# Registered with the source IP + the egress policy blob.
kwargs = client.register_bottle.call_args
self.assertEqual("172.18.0.3", kwargs.args[0])
self.assertIn("api.example.com", kwargs.kwargs["policy"])
provision.assert_called_once()
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()
provision = Mock(side_effect=RuntimeError("provision boom"))
with self.assertRaises(RuntimeError):
self._run(client, provision)
client.teardown_bottle.assert_called_once_with("b1") # no orphan left
class TestTeardownConsolidated(unittest.TestCase):
def test_deregisters_and_deprovisions(self) -> None:
client = Mock()
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.deprovision_git_gate") as deprov:
teardown_consolidated("b1", orchestrator_url="http://orch:8080")
client.teardown_bottle.assert_called_once_with("b1")
deprov.assert_called_once()
if __name__ == "__main__":
unittest.main()
+61 -110
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(
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_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \ mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()): contextlib.redirect_stderr(io.StringIO()):
provision = mock.Mock(return_value=None) yield built
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)
+113
View File
@@ -0,0 +1,113 @@
"""Unit: git-gate provisioning into the running shared gateway (PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from bot_bottle.backend.docker.gateway_provision import (
GatewayProvisionError,
deprovision_git_gate,
provision_git_gate,
)
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
_RUN = "bot_bottle.backend.docker.gateway_provision.run_docker"
def _proc(returncode: int = 0, stderr: str = "") -> Mock:
return Mock(returncode=returncode, stderr=stderr)
def _recorder(calls: list[list[str]]):
"""A run_docker side_effect that records argv and returns success."""
def fake(argv: list[str]) -> Mock:
calls.append(argv)
return _proc()
return fake
def _plan(*upstreams: GitGateUpstream) -> GitGatePlan:
return GitGatePlan(
slug="demo",
entrypoint_script=Path("/stage/entrypoint.sh"),
hook_script=Path("/stage/pre-receive"),
access_hook_script=Path("/stage/access-hook"),
upstreams=tuple(upstreams),
)
def _up(name: str, *, key: str = "/host/keys/id", known_hosts: str = "") -> GitGateUpstream:
return GitGateUpstream(
name=name,
upstream_url=f"ssh://git@github.com/x/{name}.git",
upstream_host="github.com",
upstream_port="22",
identity_file=key,
known_host_key="",
known_hosts_file=Path(known_hosts) if known_hosts else Path(),
)
class TestProvisionGitGate(unittest.TestCase):
def test_copies_creds_and_runs_namespaced_init(self) -> None:
calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "bottle1", _plan(_up("foo", known_hosts="/host/kh")))
cps = [c for c in calls if c[:2] == ["docker", "cp"]]
self.assertIn(["docker", "cp", "/host/keys/id", "gw:/git-gate/creds/bottle1/foo-key"], cps)
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))
self.assertIn("repo=/git/bottle1/${name}.git", exec_scripts[0][-1])
def test_omits_known_hosts_copy_when_absent(self) -> None:
calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "b1", _plan(_up("foo"))) # no known_hosts
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:
provision_git_gate("gw", "b1", _plan())
m.assert_not_called()
def test_raises_on_docker_failure(self) -> None:
with patch(_RUN, return_value=_proc(returncode=1, stderr="boom")):
with self.assertRaises(GatewayProvisionError):
provision_git_gate("gw", "b1", _plan(_up("foo")))
def test_rejects_unsafe_bottle_id_before_any_docker(self) -> None:
with patch(_RUN) as m:
with self.assertRaises(GatewayProvisionError):
provision_git_gate("gw", "../etc", _plan(_up("foo")))
m.assert_not_called() # rejected before a single docker call
class TestDeprovision(unittest.TestCase):
def test_removes_repo_and_creds(self) -> None:
with patch(_RUN, return_value=_proc()) as m:
deprovision_git_gate("gw", "b1")
argv = m.call_args.args[0]
self.assertEqual(["docker", "exec", "gw", "rm", "-rf"], argv[:5])
self.assertIn("/git/b1", argv)
self.assertIn("/git-gate/creds/b1", argv)
def test_rejects_unsafe_bottle_id(self) -> None:
with patch(_RUN) as m:
with self.assertRaises(GatewayProvisionError):
deprovision_git_gate("gw", "a/b")
m.assert_not_called()
if __name__ == "__main__":
unittest.main()
+65 -19
View File
@@ -6,11 +6,15 @@ import unittest
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from bot_bottle.orchestrator.gateway import ( from bot_bottle.orchestrator.gateway import (
GATEWAY_CA_CERT,
GATEWAY_NAME, GATEWAY_NAME,
DockerGateway, DockerGateway,
GatewayError, GatewayError,
) )
_CA_PEM = "-----BEGIN CERTIFICATE-----\nMII...\n-----END CERTIFICATE-----\n"
_RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker" _RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
@@ -51,6 +55,46 @@ class TestDockerGateway(unittest.TestCase):
self.assertEqual(1, len(runs)) self.assertEqual(1, len(runs))
self.assertIn(self.sc.name, runs[0]) self.assertIn(self.sc.name, runs[0])
self.assertIn("bot-bottle-sidecars:latest", runs[0]) self.assertIn("bot-bottle-sidecars:latest", runs[0])
# Runs on the shared gateway network so agents can reach it by IP.
self.assertEqual(self.sc.network, runs[0][runs[0].index("--network") + 1])
# Persists its CA on a named volume so agents keep trusting it.
self.assertTrue(any("mitmproxy" in a for a in runs[0]))
def test_ensure_running_creates_network_when_missing(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
calls.append(argv)
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="No such network")
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
with patch(_RUN_DOCKER, side_effect=fake):
self.sc.ensure_running()
creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
def test_ca_cert_pem_reads_from_container(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(stdout=_CA_PEM)) as m:
self.assertEqual(_CA_PEM, self.sc.ca_cert_pem())
argv = m.call_args.args[0]
self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
def test_ca_cert_pem_raises_when_absent(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
with self.assertRaises(GatewayError):
self.sc.ca_cert_pem()
def test_ensure_running_reuses_existing_network(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
calls.append(argv)
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
with patch(_RUN_DOCKER, side_effect=fake):
self.sc.ensure_running() # network inspect returns 0 → exists
self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
def test_ensure_running_raises_on_docker_failure(self) -> None: def test_ensure_running_raises_on_docker_failure(self) -> None:
def fake(argv: list[str]) -> Mock: def fake(argv: list[str]) -> Mock:
@@ -85,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)
@@ -115,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__":