Compare commits

...

11 Commits

Author SHA1 Message Date
didericis 49c2ed0b93 feat(smolmachines): run backend on Linux
lint / lint (push) Failing after 1m52s
test / unit (pull_request) Successful in 45s
test / integration (pull_request) Successful in 17s
Port the smolmachines backend so BOT_BOTTLE_BACKEND=smolmachines
works on Linux (KVM), not just macOS:

- Preflight gates /dev/kvm presence + accessibility on Linux with
  actionable remediation (kvm module, kvm group).
- smolvm state-DB path is platform-derived (XDG on Linux).
- force_allowlist runs on both platforms and is fail-closed: it
  verifies the persisted TSI allowlist and dies rather than booting
  a VM whose egress confinement it can't confirm. Previously it
  no-oped on Linux, failing OPEN.
- allocate() does per-bottle 127.0.0.<N> scoping on Linux too (no
  ifconfig needed — all of 127/8 is already loopback); only
  ensure_pool's lo0 aliasing stays macOS-only.
- README documents Linux + NixOS host setup.

Linux/KVM integration (the sandbox-escape acceptance gate) is
pending verification on a NixOS host; unit tests cover the new
platform branches.

Issue: #283

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NkwFXLFff9PYPy4wgVBJp9
2026-06-25 17:08:22 -04:00
didericis a666f9fe54 docs(prd): add PRD for smolmachines backend on Linux
Design for porting the smolmachines backend off macOS-only: KVM
preflight, platform-aware smolvm state-DB path, fail-closed TSI
allowlist enforcement, and per-bottle loopback scoping on Linux.
NixOS is the primary validation target.

Issue: #283

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NkwFXLFff9PYPy4wgVBJp9
2026-06-25 17:08:22 -04:00
github-actions[bot] eaf6b1f72e ci(prd): assign sequential numbers to new PRDs 2026-06-25 20:43:06 +00:00
didericis ca910f8f4f fix(start): show bottle lineage root-first with -> arrows
lint / lint (push) Successful in 1m51s
test / unit (push) Successful in 43s
test / integration (push) Successful in 17s
test / unit (pull_request) Successful in 43s
test / integration (pull_request) Successful in 18s
prd-number / assign-numbers (push) Successful in 21s
Update Quality Badges / update-badges (push) Failing after 1m47s
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-25 16:13:19 -04:00
didericis-codex 338c08a243 test: fix cli selector typing 2026-06-25 16:13:19 -04:00
didericis-claude 6faa6f67aa feat(tui,start): space/enter split, bottle lineage, YAML preflight
Three UX improvements requested in #270 review:

- filter_multiselect: Space toggles selection, Enter confirms (was both)
- bottle picker: bottles with extends chains show ancestry labels
  (e.g. 'claude-dev <- bot-bottle-dev <- dev') for at-a-glance lineage
- preflight: replaces key-value summary with YAML of the resolved manifest
2026-06-25 16:13:19 -04:00
didericis-claude b6ae6af63a fix(types): resolve pyright errors introduced in #269 changes
- manifest.py: remove unused load_bottle_chain_from_dir import
- manifest_extends.py: drop redundant ManifestEgressRoute annotation
- test_cli_start_selector.py: remove unused call import
- test_cli_tui.py: move Optional/constants to top, annotate FakeScreen,
  remove unused curses import
- test_manifest_bottle_merge.py: add type args to dict, annotate **kwargs
2026-06-25 16:13:19 -04:00
didericis-claude ad72eeddc1 feat(tui): add reordering to filter_multiselect
Tab switches focus to the selected-order panel; K/J shift the
highlighted item up/down; Space/Enter removes it. The filter list dims
while the order panel is active. Help line updates per focus mode.
2026-06-25 16:13:19 -04:00
didericis-claude 61f89de2da docs(prd): activate PRD for separate agent/bottle selection 2026-06-25 16:13:19 -04:00
didericis-claude 1ba185d1e0 feat(#269): separate agent and bottle selection at launch time
- `bottle:` in agent frontmatter is now optional; agents without it
  are portable and require bottles to be selected at launch.
- Adds `filter_multiselect` to `tui.py`: multi-select picker with
  ordered selection list, Space/Enter to toggle, Ctrl-D to confirm.
- `ManifestIndex` gains `all_bottle_names` and `load_for_agent` accepts
  `bottle_names: tuple[str, ...]` to merge bottles in order at runtime.
- `merge_bottles_runtime` in `manifest_extends.py` applies the same
  field-merge rules as `extends:` to pre-resolved bottle objects.
- `BottleSpec` gains `bottle_names`; `_validate` and `write_launch_metadata`
  thread it through so `resume` replays the same bottle configuration.
- `cmd_start` shows the bottle multiselect after agent selection,
  pre-populated from the agent's `bottle:` field when present.
- Existing agents with `bottle:` declared continue to work unchanged.
2026-06-25 16:13:19 -04:00
didericis-claude e82dbaba09 docs(prd): draft PRD for separate agent/bottle selection
Closes #269.
2026-06-25 16:13:19 -04:00
22 changed files with 2010 additions and 186 deletions
+18 -2
View File
@@ -26,7 +26,7 @@
- **Provider templates (Claude, Codex)** — `Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile. Claude auth via long-lived OAuth token; Codex via opt-in host device-auth forwarding. - **Provider templates (Claude, Codex)** — `Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile. Claude auth via long-lived OAuth token; Codex via opt-in host device-auth forwarding.
- **gVisor auto-detect** — on Linux hosts where `runsc` is registered with Docker, every bottle launches under it for a userspace syscall barrier; no manifest config required. - **gVisor auto-detect** — on Linux hosts where `runsc` is registered with Docker, every bottle launches under it for a userspace syscall barrier; no manifest config required.
- **Apple Container backend (macOS default when available)** — runs the agent and sidecar bundle with Apple's `container` CLI, using a host-only agent network plus a separate sidecar egress network. - **Apple Container backend (macOS default when available)** — runs the agent and sidecar bundle with Apple's `container` CLI, using a host-only agent network plus a separate sidecar egress network.
- **Smolmachines backend** — runs the agent in a libkrun micro-VM while the sidecar bundle stays in Docker. TSI and smolmachines DNS filtering close the raw DNS exfiltration gap that exists in the legacy Docker backend. - **Smolmachines backend** — runs the agent in a libkrun micro-VM while the sidecar bundle stays in Docker. TSI and smolmachines DNS filtering close the raw DNS exfiltration gap that exists in the legacy Docker backend. Runs on macOS (Hypervisor.framework) and Linux (KVM, `/dev/kvm`).
- **Legacy Docker backend** — still available for examples, CI, and hosts without Apple Container via `BOT_BOTTLE_BACKEND=docker` or `--backend=docker`. - **Legacy Docker backend** — still available for examples, CI, and hosts without Apple Container via `BOT_BOTTLE_BACKEND=docker` or `--backend=docker`.
## Architecture ## Architecture
@@ -72,10 +72,26 @@ When the agent exits, `cli.py` tears down every sidecar and both networks; nothi
## Quickstart ## Quickstart
On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The smolmachines backend requires Docker on the host for the sidecar bundle plus smolvm. The legacy Docker backend requires Docker. Claude bottles also need a long-lived Claude Code OAuth token (`claude setup-token`) exported as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`. On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The smolmachines backend requires Docker on the host for the sidecar bundle plus `smolvm` (macOS or Linux). The legacy Docker backend requires Docker. Claude bottles also need a long-lived Claude Code OAuth token (`claude setup-token`) exported as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`.
Use `BOT_BOTTLE_BACKEND=docker ./cli.py start <agent>` on hosts where Apple Container is not installed and Docker is the desired backend. Use `BOT_BOTTLE_BACKEND=docker ./cli.py start <agent>` on hosts where Apple Container is not installed and Docker is the desired backend.
### smolmachines on Linux
The smolmachines backend runs on Linux as well as macOS. On Linux, `smolvm`/libkrun use KVM, so the host needs:
- **`/dev/kvm`** present and accessible. Load `kvm-intel` or `kvm-amd` (and enable virtualization in BIOS/firmware). The invoking user must be in the `kvm` group: `sudo usermod -aG kvm "$USER"` then re-login. bot-bottle preflights this and reports exactly what's missing.
- **`smolvm`** on `PATH`: `curl -sSL https://smolmachines.com/install.sh | sh`.
- **Docker** for the sidecar bundle and image build, same as macOS.
Per-bottle isolation works the same as macOS without any `ifconfig`/sudo step — all of `127.0.0.0/8` is already loopback on Linux, so each bottle's sidecar bundle is published on its own `127.0.0.<N>` and TSI's allowlist is scoped to that `/32`.
```sh
BOT_BOTTLE_BACKEND=smolmachines ./cli.py start <agent>
```
> **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. If you run bottles from a Gitea Actions runner, use a `host`-label runner so Docker, `smolvm`, and `/dev/kvm` are all reachable from the job. `smolvm` isn't in nixpkgs — install the release binary (pin the version) and put it on the runner's `PATH`.
```sh ```sh
./cli.py start <agent> # builds the image on first run, drops you into claude ./cli.py start <agent> # builds the image on first run, drops you into claude
``` ```
+13 -3
View File
@@ -72,6 +72,9 @@ class BottleSpec:
identity: str = "" identity: str = ""
label: str = "" label: str = ""
color: str = "" color: str = ""
# Ordered bottle names selected at launch (issue #269). When non-empty
# they are merged in order and replace the agent's `bottle:` field.
bottle_names: tuple[str, ...] = ()
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -129,7 +132,11 @@ class BottlePlan(ABC):
info(f"provider : {self.agent_provision.template}") info(f"provider : {self.agent_provision.template}")
print_multi("env ", env_names) print_multi("env ", env_names)
print_multi("skills ", list(agent.skills)) print_multi("skills ", list(agent.skills))
info(f"bottle : {agent.bottle}") effective_bottles = (
list(spec.bottle_names) if spec.bottle_names
else ([agent.bottle] if agent.bottle else [])
)
print_multi("bottle ", effective_bottles)
identity = manifest.git_identity_summary() identity = manifest.git_identity_summary()
if identity: if identity:
@@ -363,7 +370,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
Returns the loaded Manifest for the selected agent. Subclasses with Returns the loaded Manifest for the selected agent. Subclasses with
additional preconditions should override and call additional preconditions should override and call
`super()._validate(spec)` first.""" `super()._validate(spec)` first."""
manifest = spec.manifest.load_for_agent(spec.agent_name) manifest = spec.manifest.load_for_agent(spec.agent_name, spec.bottle_names)
self._validate_skills(manifest.agent.skills) self._validate_skills(manifest.agent.skills)
self._validate_agent_provider_dockerfile(spec, manifest) self._validate_agent_provider_dockerfile(spec, manifest)
return manifest return manifest
@@ -389,9 +396,12 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
if not path.is_absolute(): if not path.is_absolute():
path = Path(spec.user_cwd) / path path = Path(spec.user_cwd) / path
if not path.is_file(): if not path.is_file():
effective = (
", ".join(spec.bottle_names) if spec.bottle_names else manifest.agent.bottle
)
die( die(
f"agent_provider.dockerfile for bottle " f"agent_provider.dockerfile for bottle "
f"'{manifest.agent.bottle}' not found: {path}" f"'{effective}' not found: {path}"
) )
@abstractmethod @abstractmethod
+1
View File
@@ -63,6 +63,7 @@ def write_launch_metadata(
backend=backend, backend=backend,
label=spec.label, label=spec.label,
color=spec.color, color=spec.color,
bottle_names=spec.bottle_names,
)) ))
+23 -15
View File
@@ -141,10 +141,12 @@ def _allocate_resources(
) -> tuple[str, str]: ) -> tuple[str, str]:
"""Reserve a loopback alias and create the per-bottle docker bridge. """Reserve a loopback alias and create the per-bottle docker bridge.
macOS only routes 127.0.0.1 by default; the per-bottle alias The per-bottle alias scopes TSI's allowlist to this bottle's
scopes TSI's allowlist to this bottle's published ports so the published ports so the agent can't reach other bottles' or host
agent can't reach other bottles' or host services' ports on services' ports on loopback. On macOS `ensure_pool` first
loopback. No-op on Linux.""" sudo-aliases the pool on `lo0`; on Linux that's a no-op since
all of 127.0.0.0/8 is already loopback, but the per-bottle
allocation runs on both."""
_loopback.ensure_pool() _loopback.ensure_pool()
loopback_ip = _loopback.allocate(plan.slug) loopback_ip = _loopback.allocate(plan.slug)
network = _bundle.bundle_network_name(plan.slug) network = _bundle.bundle_network_name(plan.slug)
@@ -190,9 +192,11 @@ def _discover_urls(
return the plan with URLs + guest_env stamped in. return the plan with URLs + guest_env stamped in.
Docker container IPs (192.168.x.x in the daemon's bridge) Docker container IPs (192.168.x.x in the daemon's bridge)
aren't reachable from the smolvm guest on macOS — TSI uses aren't reachable from the smolvm guest — TSI proxies the
macOS networking, and macOS sees the daemon's bridge via the guest's connects through the host, and the host reaches the
published-port loopback forward only. bundle only via its published-port loopback forward (the
daemon's bridge isn't on the TSI allowlist). The agent dials
the published port on the per-bottle loopback alias.
NO_PROXY includes the per-bottle loopback alias so the NO_PROXY includes the per-bottle loopback alias so the
supervise + git-gate URLs bypass HTTPS_PROXY.""" supervise + git-gate URLs bypass HTTPS_PROXY."""
@@ -252,10 +256,11 @@ def _launch_vm(
"""Create, patch, and start the smolvm VM; register teardown. """Create, patch, and start the smolvm VM; register teardown.
--allow-cidr is the per-bottle loopback alias so the guest can --allow-cidr is the per-bottle loopback alias so the guest can
only reach this bottle's bundle ports. force_allowlist patches only reach this bottle's bundle ports. force_allowlist then
smolvm 0.8.0's silent-drop of --allow-cidr when combined with confirms the allowlist persisted (patching smolvm 0.8.0's
--from. Smolfile isn't usable here — smolvm 0.8.0 makes --from silent-drop of --allow-cidr when combined with --from) and
and --smolfile mutually exclusive.""" fails closed if it can't. Smolfile isn't usable here — smolvm
0.8.0 makes --from and --smolfile mutually exclusive."""
_smolvm.machine_create( _smolvm.machine_create(
plan.machine_name, plan.machine_name,
from_path=agent_from_path, from_path=agent_from_path,
@@ -263,9 +268,10 @@ def _launch_vm(
env=plan.guest_env, env=plan.guest_env,
) )
stack.callback(_smolvm.machine_delete, plan.machine_name) stack.callback(_smolvm.machine_delete, plan.machine_name)
# Workaround smolvm 0.8.0: `--allow-cidr` is silently dropped # Confirm the booted VM's TSI allowlist will actually enforce the
# when combined with `--from`. Patch the persisted state DB # /32 before start (smolvm 0.8.0 silently drops `--allow-cidr`
# before start so the booted VM's TSI actually enforces. # with `--from`, so the persisted state DB is patched if needed).
# Fails closed if enforcement can't be confirmed.
_loopback.force_allowlist(plan.machine_name, [f"{loopback_ip}/32"]) _loopback.force_allowlist(plan.machine_name, [f"{loopback_ip}/32"])
_smolvm.machine_start(plan.machine_name) _smolvm.machine_start(plan.machine_name)
stack.callback(_smolvm.machine_stop, plan.machine_name) stack.callback(_smolvm.machine_stop, plan.machine_name)
@@ -275,7 +281,9 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
"""Repair filesystem ownership and wait for exec channel readiness. """Repair filesystem ownership and wait for exec channel readiness.
Ownership repair: smolvm's pack process remaps files to the host Ownership repair: smolvm's pack process remaps files to the host
invoker's uid (501 on macOS). /home/node must be node:node so invoker's uid (e.g. 501 on macOS, 1000 on Linux). The chowns use
names not numbers so they're correct on either. /home/node must
be node:node so
Claude Code can write ~/.claude.json; /tmp + /var/tmp need root Claude Code can write ~/.claude.json; /tmp + /var/tmp need root
mode 1777 so non-root processes can create per-uid scratch dirs. mode 1777 so non-root processes can create per-uid scratch dirs.
All folded into one sh -c to avoid back-to-back exec calls All folded into one sh -c to avoid back-to-back exec calls
@@ -33,10 +33,13 @@ sudo-add the missing pool on first use per boot — the aliases
persist on `lo0` until reboot, so subsequent launches don't persist on `lo0` until reboot, so subsequent launches don't
prompt. prompt.
Linux native daemons share the host's network namespace; the On Linux the whole `127.0.0.0/8` is already routed to `lo`, so
whole `127.0.0.0/8` is reachable by default and aliases are docker can publish a bundle's ports directly on `127.0.0.<N>`
unnecessary. The pool logic detects native-Linux and skips sudo with no `ifconfig`/sudo step. `ensure_pool` is therefore a no-op
entirely; the DB patch is also gated on macOS. on Linux, but per-bottle alias *allocation* and the TSI allowlist
DB patch run on both platforms the isolation property is
identical, it's just cheaper to set up on Linux. The state-DB
path differs per platform (see `_smolvm_db_path`).
Allocation is coordinated by inspecting running bundle Allocation is coordinated by inspecting running bundle
containers' published host IPs — each bottle's bundle owns the containers' published host IPs — each bottle's bundle owns the
@@ -47,6 +50,7 @@ from __future__ import annotations
import fcntl import fcntl
import json import json
import os
import platform import platform
import re import re
import sqlite3 import sqlite3
@@ -57,20 +61,34 @@ from typing import Iterable
from ...log import die, info from ...log import die, info
# smolvm's persistent VM state on macOS — a SQLite DB whose `vms` def _smolvm_db_path() -> Path:
# table holds one JSON BLOB per machine. The Linux path is """smolvm's persistent VM state — a SQLite DB whose `vms` table
# different, but smolmachines is macOS-only in v1 (PRD 0023) so holds one JSON BLOB per machine. macOS stores it under
# we hard-code this. If the file moves under us we'll see a `Application Support`; Linux follows the XDG base-dir spec
# clear FileNotFoundError; not worth defensive cross-platform (`$XDG_DATA_HOME`, default `~/.local/share`).
# detection until the backend actually needs Linux.
_SMOLVM_DB_PATH = ( NOTE: the Linux location is inferred from smolvm's documented
Path.home() `~/.local/share` install layout and must be confirmed against a
/ "Library" real Linux smolvm install. If it's wrong, `force_allowlist`'s
/ "Application Support" fail-closed check turns it into a clear launch-time error rather
/ "smolvm" than a silent escape."""
/ "server" if platform.system() == "Darwin":
/ "smolvm.db" return (
) Path.home()
/ "Library"
/ "Application Support"
/ "smolvm"
/ "server"
/ "smolvm.db"
)
xdg_data = os.environ.get("XDG_DATA_HOME")
base = Path(xdg_data) if xdg_data else Path.home() / ".local" / "share"
return base / "smolvm" / "server" / "smolvm.db"
# Resolved once at import: the host platform doesn't change within a
# process. Tests patch this attribute directly.
_SMOLVM_DB_PATH = _smolvm_db_path()
# Sixteen aliases by default. Tunable for hosts that want more # Sixteen aliases by default. Tunable for hosts that want more
@@ -131,51 +149,74 @@ def ensure_pool() -> None:
def force_allowlist(machine_name: str, allowed_cidrs: list[str]) -> None: def force_allowlist(machine_name: str, allowed_cidrs: list[str]) -> None:
"""Patch smolvm's persistent VM-state DB to set the machine's """Ensure the machine's persisted TSI allowlist equals
`allowed_cidrs` to the given list. Workaround for smolvm `allowed_cidrs`, failing **closed** if that can't be confirmed.
0.8.0's silent-drop of `--allow-cidr` when used with `--from`.
Must run AFTER `smolvm machine create` (the row has to Runs on both macOS and Linux. It exists because smolvm 0.8.0
exist) and BEFORE `smolvm machine start` (smolvm reads the silently drops `--allow-cidr` when combined with `--from`, so
row on start; in-flight VMs don't pick up changes). Once the allowlist has to be written into smolvm's persistent state
smolvm honors the CLI flag upstream this whole function is DB before `machine start`. Rather than assume the flag was
redundant flag-respecting create + remove this call from dropped, we read the persisted row and only patch when it
launch. doesn't already match — so a newer smolvm that honors the flag
is left untouched.
No-op on non-macOS the DB path differs and the Linux Must run AFTER `smolvm machine create` (the row has to exist)
smolmachines code path isn't exercised in v1.""" and BEFORE `smolvm machine start` (smolvm reads the row on
if not _is_macos(): start; in-flight VMs don't pick up changes).
return
Fail-closed: if the state DB is missing, the row is missing, or
the allowlist still doesn't match after patching, we `die()`
rather than boot a VM whose egress confinement we can't verify
an unconfirmed allowlist is a sandbox-escape risk (the agent
VM could reach all of host loopback)."""
want = list(allowed_cidrs)
if not _SMOLVM_DB_PATH.is_file(): if not _SMOLVM_DB_PATH.is_file():
die( die(
f"smolvm state DB not found at {_SMOLVM_DB_PATH}. " f"smolvm state DB not found at {_SMOLVM_DB_PATH}; cannot "
f"smolvm 0.8.0 expected? `smolvm --version` to check." f"confirm the TSI allowlist is enforced. Refusing to launch "
f"(fail-closed). Check `smolvm --version` and the DB "
f"location for your platform."
) )
con = sqlite3.connect(str(_SMOLVM_DB_PATH)) con = sqlite3.connect(str(_SMOLVM_DB_PATH))
try: try:
cur = con.cursor() cfg = _read_machine_cfg(con, machine_name)
row = cur.execute( if cfg.get("allowed_cidrs") != want:
"SELECT data FROM vms WHERE name = ?", (machine_name,), cfg["allowed_cidrs"] = want
).fetchone() # Write as BLOB (the column type smolvm uses) — passing a
if row is None: # plain str makes sqlite store it as Text and smolvm then
die( # fails to read it.
f"smolvm DB has no row for machine {machine_name!r}" con.execute(
f"machine_create must run before force_allowlist." "UPDATE vms SET data = ? WHERE name = ?",
(sqlite3.Binary(json.dumps(cfg).encode()), machine_name),
)
con.commit()
cfg = _read_machine_cfg(con, machine_name)
if cfg.get("allowed_cidrs") != want:
die(
f"could not enforce TSI allowlist {want!r} for machine "
f"{machine_name!r} (persisted value is "
f"{cfg.get('allowed_cidrs')!r}). Refusing to launch "
f"(fail-closed)."
) )
cfg = json.loads(row[0])
cfg["allowed_cidrs"] = list(allowed_cidrs)
# Write as BLOB (the column type smolvm uses) — passing a
# plain str makes sqlite store it as Text and smolvm then
# fails to read it.
cur.execute(
"UPDATE vms SET data = ? WHERE name = ?",
(sqlite3.Binary(json.dumps(cfg).encode()), machine_name),
)
con.commit()
finally: finally:
con.close() con.close()
def _read_machine_cfg(con: sqlite3.Connection, machine_name: str) -> dict[str, object]:
"""Read + JSON-decode a machine's `data` BLOB from the smolvm
state DB. Dies (fail-closed) if the row is missing the caller
can't confirm enforcement without it."""
row = con.execute(
"SELECT data FROM vms WHERE name = ?", (machine_name,),
).fetchone()
if row is None:
die(
f"smolvm DB has no row for machine {machine_name!r}"
f"machine_create must run before force_allowlist."
)
return json.loads(row[0])
def allocate(_slug: str) -> str: def allocate(_slug: str) -> str:
"""Pick the lowest-numbered alias from the pool not already """Pick the lowest-numbered alias from the pool not already
in use by a running smolmachines bundle. Bails when the pool in use by a running smolmachines bundle. Bails when the pool
@@ -184,16 +225,17 @@ def allocate(_slug: str) -> str:
used (no on-disk reservation, allocation is purely used (no on-disk reservation, allocation is purely
docker-state-driven). docker-state-driven).
On non-macOS the whole `127.0.0.0/8` is loopback by default; Runs on both platforms: the allocation logic (docker-state
`127.0.0.1` is fine to share and we skip the alias dance. inspection + the file lock) is platform-independent. macOS
This still returns a deterministic address so launch.py's needs `ensure_pool` to have aliased the addresses on `lo0`
callers don't have to branch on platform. first; on Linux all of `127.0.0.0/8` is already loopback, so
docker can publish on the chosen `127.0.0.<N>` with no setup.
Per-bottle scoping (so the agent can't reach other bottles' or
host services' loopback ports) therefore holds on both.
An exclusive file lock serialises concurrent calls so two An exclusive file lock serialises concurrent calls so two
simultaneous launches don't read the same docker state and simultaneous launches don't read the same docker state and
claim the same alias.""" claim the same alias."""
if not _is_macos():
return "127.0.0.1"
_ALLOC_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) _ALLOC_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(_ALLOC_LOCK_PATH, "w", encoding="utf-8") as lf: with open(_ALLOC_LOCK_PATH, "w", encoding="utf-8") as lf:
fcntl.flock(lf, fcntl.LOCK_EX) fcntl.flock(lf, fcntl.LOCK_EX)
+46 -14
View File
@@ -5,26 +5,58 @@ unit-tested without importing the docker subprocess paths."""
from __future__ import annotations from __future__ import annotations
import hashlib import hashlib
import os
import platform
import shutil import shutil
from ...log import die from ...log import die
# libkrun's Linux backend drives the guest through KVM, so the host
# must expose `/dev/kvm` and the invoking user must be able to open
# it. macOS uses Hypervisor.framework and needs no device node.
_KVM_DEVICE = "/dev/kvm"
def smolmachines_preflight() -> None: def smolmachines_preflight() -> None:
"""Ensure `smolvm` is on PATH before the launch flow runs. """Ensure the host can run the smolmachines backend before the
Called from `_resolve_plan`; gives the operator a clear launch flow starts. Called from `_resolve_plan`; surfaces a
install pointer rather than a cryptic FileNotFoundError clear, actionable error instead of a cryptic `smolvm` failure
later. `gvproxy` is no longer required see the PRD's design deep in launch.
pivot section."""
if shutil.which("smolvm") is not None: Checks `smolvm` is on PATH (both platforms) and, on Linux,
return that `/dev/kvm` exists and is accessible. `gvproxy` is no
die( longer required see the PRD's design pivot section."""
"BOT_BOTTLE_BACKEND=smolmachines requires `smolvm` on " if shutil.which("smolvm") is None:
"PATH. Install with: " die(
"curl -sSL https://smolmachines.com/install.sh | sh. " "BOT_BOTTLE_BACKEND=smolmachines requires `smolvm` on "
"To use the legacy Docker backend instead, set " "PATH. Install with: "
"BOT_BOTTLE_BACKEND=docker or pass --backend=docker." "curl -sSL https://smolmachines.com/install.sh | sh. "
) "To use the legacy Docker backend instead, set "
"BOT_BOTTLE_BACKEND=docker or pass --backend=docker."
)
if platform.system() == "Linux":
_preflight_kvm()
def _preflight_kvm() -> None:
"""Linux-only: libkrun needs `/dev/kvm`. Distinguish 'KVM not
enabled' from 'no permission' so the operator knows which to
fix."""
if not os.path.exists(_KVM_DEVICE):
die(
f"BOT_BOTTLE_BACKEND=smolmachines needs {_KVM_DEVICE} on "
"Linux but it is missing. Enable KVM: load the kvm-intel "
"or kvm-amd kernel module (and confirm virtualization is "
"enabled in BIOS/firmware). To use the legacy Docker "
"backend instead, set BOT_BOTTLE_BACKEND=docker."
)
if not os.access(_KVM_DEVICE, os.R_OK | os.W_OK):
die(
f"{_KVM_DEVICE} exists but is not readable/writable by the "
"current user. Add your user to the `kvm` group "
"(`sudo usermod -aG kvm \"$USER\"`) and re-login, or run "
"with access to the device."
)
def smolmachines_bundle_subnet(slug: str) -> tuple[str, str, str]: def smolmachines_bundle_subnet(slug: str) -> tuple[str, str, str]:
+9
View File
@@ -111,6 +111,10 @@ class BottleMetadata:
backend: str = "" backend: str = ""
label: str = "" label: str = ""
color: str = "" color: str = ""
# Ordered bottle names selected at launch (issue #269). Empty tuple
# for state dirs written before this change; resume falls back to
# the agent's `bottle:` field in that case.
bottle_names: tuple[str, ...] = ()
def metadata_path(identity: str) -> Path: def metadata_path(identity: str) -> Path:
@@ -138,6 +142,10 @@ def read_metadata(identity: str) -> BottleMetadata | None:
if not isinstance(raw, dict): if not isinstance(raw, dict):
return None return None
raw_typed = cast(dict[str, object], raw) raw_typed = cast(dict[str, object], raw)
raw_bottle_names = raw_typed.get("bottle_names", [])
bottle_names: tuple[str, ...] = ()
if isinstance(raw_bottle_names, list):
bottle_names = tuple(str(n) for n in raw_bottle_names if isinstance(n, str))
return BottleMetadata( return BottleMetadata(
identity=str(raw_typed.get("identity", identity)), identity=str(raw_typed.get("identity", identity)),
agent_name=str(raw_typed.get("agent_name", "")), agent_name=str(raw_typed.get("agent_name", "")),
@@ -148,6 +156,7 @@ def read_metadata(identity: str) -> BottleMetadata | None:
backend=str(raw_typed.get("backend", "")), backend=str(raw_typed.get("backend", "")),
label=str(raw_typed.get("label", "")), label=str(raw_typed.get("label", "")),
color=str(raw_typed.get("color", "")), color=str(raw_typed.get("color", "")),
bottle_names=bottle_names,
) )
+1
View File
@@ -49,6 +49,7 @@ def cmd_resume(argv: list[str]) -> int:
copy_cwd=metadata.copy_cwd, copy_cwd=metadata.copy_cwd,
user_cwd=metadata.cwd or USER_CWD, user_cwd=metadata.cwd or USER_CWD,
identity=metadata.identity, identity=metadata.identity,
bottle_names=tuple(metadata.bottle_names),
) )
backend_name = metadata.backend or None backend_name = metadata.backend or None
return _launch_bottle( return _launch_bottle(
+154 -2
View File
@@ -32,7 +32,7 @@ from ..bottle_state import (
mark_preserved, mark_preserved,
) )
from ..log import info from ..log import info
from ..manifest import ManifestIndex from ..manifest import Manifest, ManifestIndex
from ._common import PROG, USER_CWD, read_tty_line from ._common import PROG, USER_CWD, read_tty_line
from . import tui from . import tui
@@ -73,6 +73,23 @@ def cmd_start(argv: list[str]) -> int:
backend_name: str | None = args.backend backend_name: str | None = args.backend
# Bottle multiselect: always show after agent selection so operators
# can compose bottles at launch time without editing agent manifests.
available_bottles = manifest.all_bottle_names
lineage_map = _bottle_lineage(manifest)
display_labels = [lineage_map.get(n, n) for n in available_bottles]
label_to_name = {lineage_map.get(n, n): n for n in available_bottles}
initial_bottle = _peek_agent_bottle(manifest, agent_name)
initial_labels = [lineage_map.get(initial_bottle, initial_bottle)] if initial_bottle else []
selected_labels = tui.filter_multiselect(
display_labels,
title="Select bottles",
initial=initial_labels,
)
if selected_labels is None:
return 0
bottle_names = tuple(label_to_name.get(lbl, lbl) for lbl in selected_labels)
label, color = tui.name_color_modal(default_label=agent_name) label, color = tui.name_color_modal(default_label=agent_name)
label, color = _resolve_unique_label(label, color) label, color = _resolve_unique_label(label, color)
@@ -83,6 +100,7 @@ def cmd_start(argv: list[str]) -> int:
user_cwd=USER_CWD, user_cwd=USER_CWD,
label=label, label=label,
color=color, color=color,
bottle_names=bottle_names,
) )
return _launch_bottle( return _launch_bottle(
spec, spec,
@@ -189,6 +207,38 @@ def _identity_from_plan(plan: object) -> str:
return getattr(plan, "slug", "") return getattr(plan, "slug", "")
def _peek_agent_bottle(manifest: ManifestIndex, agent_name: str) -> str:
"""Return the `bottle:` value from the named agent's frontmatter without
fully parsing the agent file, or "" when absent or unreadable.
Used to pre-populate the bottle multiselect with the agent's default
bottle so operators who haven't removed `bottle:` from their manifests
don't need to re-select it every time."""
if manifest.home_md is None:
# Eager mode (from_json_obj): agent is pre-parsed.
if agent_name in manifest.agents:
return manifest.agents[agent_name].bottle
return ""
from ..manifest_loader import scan_agent_names
from ..yaml_subset import YamlSubsetError, parse_frontmatter
home_agents = scan_agent_names(manifest.home_md / "agents")
cwd_agents: dict[str, Path] = {}
if manifest.cwd_md is not None:
cwd_agents = scan_agent_names(manifest.cwd_md / "agents")
merged = {**home_agents, **cwd_agents}
path = merged.get(agent_name)
if path is None:
return ""
try:
fm, _ = parse_frontmatter(path.read_text())
bottle = fm.get("bottle", "")
return str(bottle) if isinstance(bottle, str) else ""
except (OSError, YamlSubsetError):
return ""
def _resolve_unique_label(label: str, color: str) -> tuple[str, str]: def _resolve_unique_label(label: str, color: str) -> tuple[str, str]:
"""Re-prompt with a disclaimer until the label's slug is not already """Re-prompt with a disclaimer until the label's slug is not already
in use among running bottles. Passes through unchanged when no in use among running bottles. Passes through unchanged when no
@@ -215,10 +265,112 @@ def _text_prompt_yes() -> bool:
def _text_render_preflight(): def _text_render_preflight():
def _render(plan: DockerBottlePlan) -> None: def _render(plan: DockerBottlePlan) -> None:
plan.print() print(file=sys.stderr)
print(_manifest_to_yaml(plan.manifest), file=sys.stderr)
return _render return _render
def _bottle_lineage(manifest: ManifestIndex) -> dict[str, str]:
"""Return {bottle_name: lineage_label} for bottles that have an extends chain.
Bottles without a parent are omitted (the caller falls back to the bare name).
Labels show the chain root-first: e.g. 'dev -> bot-bottle-dev -> claude-dev'."""
if manifest.home_md is None:
return {}
bottles_dir = manifest.home_md / "bottles"
if not bottles_dir.is_dir():
return {}
from ..yaml_subset import YamlSubsetError, parse_frontmatter
extends_of: dict[str, str] = {}
for path in bottles_dir.glob("*.md"):
try:
fm, _ = parse_frontmatter(path.read_text())
parent = fm.get("extends", "")
if isinstance(parent, str) and parent:
extends_of[path.stem] = parent
except (OSError, YamlSubsetError):
pass
labels: dict[str, str] = {}
for name in extends_of:
chain = [name]
seen = {name}
cur = name
while cur in extends_of:
par = extends_of[cur]
if par in seen:
break
chain.append(par)
seen.add(par)
cur = par
labels[name] = " -> ".join(reversed(chain))
return labels
def _manifest_to_yaml(manifest: Manifest) -> str:
"""Serialize the resolved Manifest to a YAML string for preflight display."""
lines: list[str] = []
agent = manifest.agent
lines.append("agent:")
if agent.skills:
lines.append(" skills:")
for s in agent.skills:
lines.append(f" - {s}")
if not agent.git_user.is_empty():
lines.append(" git-gate:")
lines.append(" user:")
if agent.git_user.name:
lines.append(f" name: {agent.git_user.name}")
if agent.git_user.email:
lines.append(f" email: {agent.git_user.email}")
bottle = manifest.bottle
lines.append("bottle:")
if bottle.agent_provider.template != "claude" or bottle.agent_provider.dockerfile:
lines.append(" agent_provider:")
lines.append(f" template: {bottle.agent_provider.template}")
if bottle.agent_provider.dockerfile:
lines.append(f" dockerfile: {bottle.agent_provider.dockerfile}")
if bottle.env:
lines.append(" env:")
for k, v in sorted(bottle.env.items()):
lines.append(f" {k}: {v}")
has_git_gate = not bottle.git_user.is_empty() or bottle.git
if has_git_gate:
lines.append(" git-gate:")
if not bottle.git_user.is_empty():
lines.append(" user:")
if bottle.git_user.name:
lines.append(f" name: {bottle.git_user.name}")
if bottle.git_user.email:
lines.append(f" email: {bottle.git_user.email}")
if bottle.git:
lines.append(" repos:")
for entry in bottle.git:
lines.append(f" {entry.Name}:")
lines.append(f" url: {entry.Upstream}")
if bottle.egress.routes:
lines.append(" egress:")
lines.append(" routes:")
for r in bottle.egress.routes:
lines.append(f" - host: {r.Host}")
if r.AuthScheme:
lines.append(f" auth:")
lines.append(f" scheme: {r.AuthScheme}")
lines.append(f" supervise: {'true' if bottle.supervise else 'false'}")
return "\n".join(lines)
def _launch_bottle( def _launch_bottle(
spec: BottleSpec, spec: BottleSpec,
*, *,
+292
View File
@@ -17,6 +17,43 @@ import sys
from typing import Any, Optional from typing import Any, Optional
def filter_multiselect(
items: list[str],
*,
title: str = "",
initial: Optional[list[str]] = None,
tty_path: str = "/dev/tty",
) -> Optional[list[str]]:
"""Render a multi-select picker over *items*.
Returns the ordered list of selected items, or ``None`` if the user
cancelled (Esc / ``q`` / Ctrl-C / Ctrl-D with no items).
Press Space to toggle the item under the cursor.
Press Enter to confirm the current selection.
Press Ctrl-D to confirm the current selection (returns even if empty).
Press Esc/q to cancel (returns None).
*initial* pre-populates the selection in insertion order. Items
added are appended; removed items leave the remaining order unchanged.
"""
if not items:
return []
try:
tty_fd = open(tty_path, "r+b", buffering=0)
except OSError:
return None
try:
fd_dup = os.dup(tty_fd.fileno())
return _run_multiselect(
items, title=title, initial=list(initial or []), tty_fd=fd_dup
)
finally:
tty_fd.close()
def filter_select( def filter_select(
items: list[str], items: list[str],
*, *,
@@ -221,6 +258,261 @@ def _addstr_safe(screen: Any, row: int, col: int, text: str, attr: int = curses.
pass pass
# ---------------------------------------------------------------------------
# filter_multiselect internals
# ---------------------------------------------------------------------------
_KEY_SPACE = 32
def _run_multiselect(
items: list[str], *, title: str, initial: list[str], tty_fd: int
) -> Optional[list[str]]:
"""Drive a curses multi-select session on *tty_fd*."""
os.environ.setdefault("TERM", "xterm-256color")
orig_stdin = sys.__stdin__
orig_stdout = sys.__stdout__
try:
import io
tty_text = io.TextIOWrapper(io.FileIO(tty_fd, mode='r+'), write_through=True)
sys.__stdin__ = tty_text # type: ignore[assignment]
sys.__stdout__ = tty_text # type: ignore[assignment]
screen = curses.initscr()
curses.noecho()
curses.cbreak()
screen.keypad(True)
try:
result = _multiselect_loop(screen, items, title=title, initial=initial)
finally:
screen.keypad(False)
curses.nocbreak()
curses.echo()
curses.endwin()
except Exception: # noqa: W0718
return None
finally:
sys.__stdin__ = orig_stdin # type: ignore[assignment]
sys.__stdout__ = orig_stdout # type: ignore[assignment]
return result
def _multiselect_loop(
screen: Any, items: list[str], *, title: str, initial: list[str]
) -> Optional[list[str]]:
query = ""
cursor = 0
selected: list[str] = [s for s in initial if s in items]
# focus = "filter": navigate + toggle items in the filterable list
# focus = "order": navigate + reorder items in the selected list
focus = "filter"
order_cursor = 0
while True:
filtered = _filter_items(items, query)
if not filtered:
cursor = 0
elif cursor >= len(filtered):
cursor = len(filtered) - 1
if not selected:
order_cursor = 0
if focus == "order":
focus = "filter"
elif order_cursor >= len(selected):
order_cursor = len(selected) - 1
try:
_render_multiselect(
screen, filtered, cursor,
query=query, title=title, selected=selected,
focus=focus, order_cursor=order_cursor,
)
except curses.error:
return None
try:
key = screen.getch()
except KeyboardInterrupt:
return None
if key in (_KEY_ESC, _KEY_CTRL_C, ord("q")):
return None
if key == _KEY_CTRL_D:
return list(selected)
# Tab toggles between filter and order focus.
if key == ord("\t"):
if focus == "filter" and selected:
focus = "order"
order_cursor = 0
else:
focus = "filter"
continue
if focus == "filter":
if key in (curses.KEY_ENTER, _KEY_ENTER_ALT, ord("\r")):
return list(selected)
elif key == _KEY_SPACE:
if filtered:
item = filtered[cursor]
if item in selected:
selected.remove(item)
else:
selected.append(item)
elif key in (curses.KEY_UP, ord("k")):
if cursor > 0:
cursor -= 1
elif key in (curses.KEY_DOWN, ord("j")):
if cursor < len(filtered) - 1:
cursor += 1
elif key in (curses.KEY_BACKSPACE, _KEY_BACKSPACE_WIN, 127):
query = query[:-1]
new_filtered = _filter_items(items, query)
if cursor >= len(new_filtered):
cursor = max(0, len(new_filtered) - 1)
elif 32 <= key <= 126 and key != _KEY_SPACE:
query += chr(key)
cursor = 0
else: # focus == "order"
if key in (curses.KEY_UP, ord("k")):
if order_cursor > 0:
order_cursor -= 1
elif key in (curses.KEY_DOWN, ord("j")):
if order_cursor < len(selected) - 1:
order_cursor += 1
elif key == ord("K"):
# Move selected item up (earlier in order).
if order_cursor > 0:
i = order_cursor
selected[i - 1], selected[i] = selected[i], selected[i - 1]
order_cursor -= 1
elif key == ord("J"):
# Move selected item down (later in order).
if order_cursor < len(selected) - 1:
i = order_cursor
selected[i], selected[i + 1] = selected[i + 1], selected[i]
order_cursor += 1
elif key in (curses.KEY_ENTER, _KEY_ENTER_ALT, ord("\r"), _KEY_SPACE):
# Remove item from selection while in order mode.
del selected[order_cursor]
if order_cursor >= len(selected) and order_cursor > 0:
order_cursor -= 1
def _render_multiselect(
screen: Any,
filtered: list[str],
cursor: int,
*,
query: str,
title: str,
selected: list[str],
focus: str = "filter",
order_cursor: int = 0,
) -> None:
screen.erase()
rows, cols = screen.getmaxyx()
min_rows = 7
if rows < min_rows:
raise curses.error("terminal too small")
sep = "" * min(cols - 1, 40)
row = 0
if title and row < rows - 1:
_addstr_safe(screen, row, 0, title[:cols - 1], curses.A_BOLD)
row += 1
# Filter line — dim when focus is on the order panel.
filter_label = f"Filter: {query}"
filter_hint = " [Tab: reorder]" if focus == "filter" and selected else ""
filter_attr = curses.A_DIM if focus == "order" else curses.A_NORMAL
if row < rows - 1:
_addstr_safe(screen, row, 0, (filter_label + filter_hint)[:cols - 1], filter_attr)
row += 1
if row < rows - 1:
_addstr_safe(screen, row, 0, sep)
row += 1
# Compute how many rows the bottom order panel needs.
# Cap the visible selected list to keep the filter list legible.
order_rows = min(len(selected), max(1, (rows - row) // 3)) if selected else 0
# Bottom reserved: sep + order_rows + sep + help = order_rows + 3
bottom_reserved = order_rows + 3
list_start = row
list_rows = rows - list_start - bottom_reserved
if list_rows < 1:
list_rows = 1
selected_set = set(selected)
filter_dim = focus == "order"
scroll = max(0, cursor - list_rows + 1)
visible = filtered[scroll: scroll + list_rows]
for idx, item in enumerate(visible):
abs_idx = scroll + idx
mark = "[*]" if item in selected_set else "[ ]"
prefix = "> " if (abs_idx == cursor and focus == "filter") else " "
line = (prefix + mark + " " + item)[:cols - 1]
item_attr = curses.A_DIM if filter_dim else (
curses.A_REVERSE if abs_idx == cursor else curses.A_NORMAL
)
if row < rows - bottom_reserved:
_addstr_safe(screen, row, 0, line, item_attr)
row += 1
# Separator before the order panel.
if row < rows - (order_rows + 2):
_addstr_safe(screen, row, 0, sep)
row += 1
# Order panel.
order_scroll = max(0, order_cursor - order_rows + 1)
order_visible = selected[order_scroll: order_scroll + order_rows]
for idx, item in enumerate(order_visible):
abs_idx = order_scroll + idx
is_active = focus == "order" and abs_idx == order_cursor
prefix = "> " if is_active else " "
line = (prefix + item)[:cols - 1]
attr = curses.A_REVERSE if is_active else curses.A_NORMAL
if row < rows - 2:
_addstr_safe(screen, row, 0, line, attr)
row += 1
if row < rows - 1:
_addstr_safe(screen, row, 0, sep)
row += 1
if focus == "filter":
help_line = "[↑↓/jk] move [Space] toggle [Enter] confirm [Tab] reorder [Esc/q] cancel"
else:
help_line = "[↑↓/jk] cursor [K/J] reorder [Space/Enter] remove [Tab] back [Ctrl-D] done"
if row < rows:
_addstr_safe(screen, min(rows - 1, row), 0, help_line[:cols - 1])
screen.refresh()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# name_color_modal — two-step label + color picker # name_color_modal — two-step label + color picker
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+103 -14
View File
@@ -213,6 +213,65 @@ def _merge_git_user(
) )
def _resolve_effective_bottle_eager(
agent_name: str,
agent: "ManifestAgent",
bottle_names: "tuple[str, ...]",
bottles: "Mapping[str, ManifestBottle]",
) -> "ManifestBottle":
"""Return the effective ManifestBottle for the eager (from_json_obj) path.
When bottle_names is non-empty they are merged in order. When empty, falls
back to agent.bottle. Raises ManifestError when neither is set."""
from .manifest_extends import merge_bottles_runtime
if bottle_names:
resolved: list[ManifestBottle] = []
for bn in bottle_names:
if bn not in bottles:
available = ", ".join(sorted(bottles.keys())) or "(none)"
raise ManifestError(
f"bottle '{bn}' not defined. Available: {available}"
)
resolved.append(bottles[bn])
return merge_bottles_runtime(resolved)
if not agent.bottle:
raise ManifestError(
f"agent '{agent_name}' has no 'bottle' field and no bottles were "
f"selected at launch. Select at least one bottle or add "
f"'bottle: <name>' to the agent manifest."
)
return bottles[agent.bottle]
def _resolve_effective_bottle_lazy(
agent_name: str,
agent_bottle: str,
bottle_names: "tuple[str, ...]",
bottles_dir: "Path",
) -> "ManifestBottle":
"""Return the effective ManifestBottle for the lazy (from_md_dirs) path.
When bottle_names is non-empty they are resolved from disk and merged in
order. When empty, falls back to agent_bottle. Raises ManifestError when
neither is set."""
from .manifest_extends import merge_bottles_runtime
from .manifest_loader import load_bottle_chain_from_dir
if bottle_names:
resolved = [load_bottle_chain_from_dir(bn, bottles_dir) for bn in bottle_names]
return merge_bottles_runtime(resolved)
if not agent_bottle:
raise ManifestError(
f"agent '{agent_name}' has no 'bottle' field and no bottles were "
f"selected at launch. Select at least one bottle or add "
f"'bottle: <name>' to the agent manifest."
)
return load_bottle_chain_from_dir(agent_bottle, bottles_dir)
@dataclass(frozen=True) @dataclass(frozen=True)
class Manifest: class Manifest:
"""Single-agent/bottle value type. Returned by ManifestIndex.load_for_agent(). """Single-agent/bottle value type. Returned by ManifestIndex.load_for_agent().
@@ -358,6 +417,18 @@ class ManifestIndex:
} }
return cls(bottles=bottles, agents=agents) return cls(bottles=bottles, agents=agents)
@property
def all_bottle_names(self) -> list[str]:
"""Sorted list of all discoverable bottle names.
In names-only mode (from resolve/from_md_dirs) this scans bottle
filenames without reading their content. In eager mode (from
from_json_obj) it returns the pre-parsed bottles' names."""
if self.home_md is not None:
from .manifest_loader import scan_bottle_names
return scan_bottle_names(self.home_md / "bottles")
return sorted(self.bottles.keys())
@property @property
def all_agent_names(self) -> list[str]: def all_agent_names(self) -> list[str]:
"""Sorted list of all discoverable agent names. """Sorted list of all discoverable agent names.
@@ -374,9 +445,18 @@ class ManifestIndex:
return sorted(home_names | cwd_names) return sorted(home_names | cwd_names)
return sorted(self.agents.keys()) return sorted(self.agents.keys())
def load_for_agent(self, agent_name: str) -> "Manifest": def load_for_agent(
self,
agent_name: str,
bottle_names: "tuple[str, ...] | None" = None,
) -> "Manifest":
"""Parse the named agent and its bottle; return a single-value Manifest. """Parse the named agent and its bottle; return a single-value Manifest.
`bottle_names` is an ordered list of bottles selected at launch time.
When non-empty they are resolved and merged in order (index 0 = base;
later entries override). When empty or None, falls back to the agent's
own `bottle:` field. Raises ManifestError when neither is set.
In lazy mode (from resolve/from_md_dirs) the agent file and its In lazy mode (from resolve/from_md_dirs) the agent file and its
bottle chain are read from disk for the first time here. In eager bottle chain are read from disk for the first time here. In eager
mode (from_json_obj) the data is already parsed; this just filters mode (from_json_obj) the data is already parsed; this just filters
@@ -387,6 +467,8 @@ class ManifestIndex:
Always raises ManifestError if the agent is unknown or invalid. Always raises ManifestError if the agent is unknown or invalid.
Backends call this at preflight inside _validate.""" Backends call this at preflight inside _validate."""
effective_bottle_names: tuple[str, ...] = bottle_names or ()
if self.home_md is None: if self.home_md is None:
# Eager manifest (from_json_obj): data already parsed; filter to # Eager manifest (from_json_obj): data already parsed; filter to
# the one requested agent and its bottle so the returned Manifest # the one requested agent and its bottle so the returned Manifest
@@ -397,12 +479,14 @@ class ManifestIndex:
f"agent '{agent_name}' not defined. Available: {available}" f"agent '{agent_name}' not defined. Available: {available}"
) )
agent = self.agents[agent_name] agent = self.agents[agent_name]
raw_bottle = self.bottles[agent.bottle] raw_bottle = _resolve_effective_bottle_eager(
agent_name, agent, effective_bottle_names, self.bottles
)
merged = _merge_git_user(agent.git_user, raw_bottle.git_user) merged = _merge_git_user(agent.git_user, raw_bottle.git_user)
bottle = raw_bottle if merged == raw_bottle.git_user else replace(raw_bottle, git_user=merged) bottle = raw_bottle if merged == raw_bottle.git_user else replace(raw_bottle, git_user=merged)
return Manifest(agent=agent, bottle=bottle) return Manifest(agent=agent, bottle=bottle)
from .manifest_loader import load_bottle_chain_from_dir, scan_agent_names from .manifest_loader import scan_agent_names
from .manifest_schema import validate_agent_frontmatter_keys from .manifest_schema import validate_agent_frontmatter_keys
from .yaml_subset import YamlSubsetError, parse_frontmatter from .yaml_subset import YamlSubsetError, parse_frontmatter
@@ -429,26 +513,31 @@ class ManifestIndex:
validate_agent_frontmatter_keys(agent_path, fm.keys()) validate_agent_frontmatter_keys(agent_path, fm.keys())
bottle_name = fm.get("bottle") # Determine the effective bottle name(s).
if not isinstance(bottle_name, str) or not bottle_name: agent_bottle = fm.get("bottle") or ""
raise ManifestError(
f"agent '{agent_name}' must declare a 'bottle' field "
f"naming a defined bottle"
)
# Load the bottle chain (may raise ManifestError).
bottles_dir = self.home_md / "bottles" bottles_dir = self.home_md / "bottles"
raw_bottle = load_bottle_chain_from_dir(bottle_name, bottles_dir) raw_bottle = _resolve_effective_bottle_lazy(
agent_name, str(agent_bottle), effective_bottle_names, bottles_dir
)
effective_bottle_name = (
effective_bottle_names[-1] if effective_bottle_names
else str(agent_bottle)
)
# Build and validate the full ManifestAgent. # Build and validate the full ManifestAgent.
agent_dict: dict[str, object] = { agent_dict: dict[str, object] = {
"bottle": bottle_name,
"skills": fm.get("skills", []), "skills": fm.get("skills", []),
"prompt": body.strip(), "prompt": body.strip(),
} }
if agent_bottle:
agent_dict["bottle"] = agent_bottle
if "git-gate" in fm: if "git-gate" in fm:
agent_dict["git-gate"] = fm["git-gate"] agent_dict["git-gate"] = fm["git-gate"]
agent = ManifestAgent.from_dict(agent_name, agent_dict, {bottle_name}) # Pass the effective bottle name as the known-bottles set so agents
# that have bottle: set are validated; agents without bottle: pass {}
# since bottle_names were already resolved above.
known = {effective_bottle_name} if effective_bottle_name else set()
agent = ManifestAgent.from_dict(agent_name, agent_dict, known)
merged_user = _merge_git_user(agent.git_user, raw_bottle.git_user) merged_user = _merge_git_user(agent.git_user, raw_bottle.git_user)
bottle = raw_bottle if merged_user == raw_bottle.git_user else replace(raw_bottle, git_user=merged_user) bottle = raw_bottle if merged_user == raw_bottle.git_user else replace(raw_bottle, git_user=merged_user)
+16 -13
View File
@@ -109,7 +109,8 @@ class ManifestAgentProvider:
@dataclass(frozen=True) @dataclass(frozen=True)
class ManifestAgent: class ManifestAgent:
bottle: str # Optional: when empty the operator selects bottles at launch time.
bottle: str = ""
skills: tuple[str, ...] = () skills: tuple[str, ...] = ()
prompt: str = "" prompt: str = ""
# Per-agent git identity (issue #94). Overlays the referenced # Per-agent git identity (issue #94). Overlays the referenced
@@ -129,18 +130,20 @@ class ManifestAgent:
f"allowed keys are {allowed}." f"allowed keys are {allowed}."
) )
bottle = d.get("bottle") bottle_raw = d.get("bottle")
if not isinstance(bottle, str) or not bottle: bottle = ""
raise ManifestError( if bottle_raw is not None:
f"agent '{name}' must declare a 'bottle' field naming a " if not isinstance(bottle_raw, str) or not bottle_raw:
f"defined bottle" raise ManifestError(
) f"agent '{name}' bottle must be a non-empty string when declared"
if bottle not in bottle_names: )
available = ", ".join(sorted(bottle_names)) or "(none defined)" if bottle_raw not in bottle_names:
raise ManifestError( available = ", ".join(sorted(bottle_names)) or "(none defined)"
f"agent '{name}' references bottle '{bottle}', which is not defined. " raise ManifestError(
f"Available: {available}" f"agent '{name}' references bottle '{bottle_raw}', which is not defined. "
) f"Available: {available}"
)
bottle = bottle_raw
skills: tuple[str, ...] = () skills: tuple[str, ...] = ()
skills_raw = d.get("skills") skills_raw = d.get("skills")
+52
View File
@@ -9,6 +9,58 @@ if TYPE_CHECKING:
from .manifest_egress import ManifestEgressConfig from .manifest_egress import ManifestEgressConfig
def merge_bottles_runtime(bottles: "list[ManifestBottle]") -> "ManifestBottle":
"""Merge an ordered list of pre-resolved ManifestBottle objects.
Index 0 is the base; each subsequent entry is applied on top using
the same field-merge rules as the file-based extends machinery:
env: dict merge, later wins; git_user: per-field overlay, later
wins on non-empty; git (repos): union by name, later wins; egress
routes: concatenate; agent_provider, supervise: later replaces.
"""
if not bottles:
raise ValueError("merge_bottles_runtime requires at least one bottle")
result = bottles[0]
for override in bottles[1:]:
result = _merge_two_bottles_runtime(result, override)
return result
def _merge_two_bottles_runtime(base: "ManifestBottle", override: "ManifestBottle") -> "ManifestBottle":
from .manifest import ManifestBottle, ManifestGitUser
from .manifest_egress import ManifestEgressConfig
merged_env = {**base.env, **override.env}
merged_git_user = ManifestGitUser(
name=override.git_user.name or base.git_user.name,
email=override.git_user.email or base.git_user.email,
)
# git repos: union keyed by Name, override wins per-name.
base_repos_by_name = {entry.Name: entry for entry in base.git}
override_repos_by_name = {entry.Name: entry for entry in override.git}
merged_repos_names = list(base_repos_by_name) + [
n for n in override_repos_by_name if n not in base_repos_by_name
]
merged_git = tuple(
override_repos_by_name.get(n, base_repos_by_name[n])
for n in merged_repos_names
)
merged_routes = base.egress.routes + override.egress.routes
merged_egress = ManifestEgressConfig(routes=merged_routes, Log=override.egress.Log)
return ManifestBottle(
env=merged_env,
agent_provider=override.agent_provider,
git=merged_git,
git_user=merged_git_user,
egress=merged_egress,
supervise=override.supervise,
)
def resolve_bottles(raws: dict[str, dict[str, object]]) -> dict[str, ManifestBottle]: def resolve_bottles(raws: dict[str, dict[str, object]]) -> dict[str, ManifestBottle]:
"""Apply `extends:` chains and return resolved ManifestBottle objects.""" """Apply `extends:` chains and return resolved ManifestBottle objects."""
cache: dict[str, ManifestBottle] = {} cache: dict[str, ManifestBottle] = {}
+19
View File
@@ -32,6 +32,25 @@ def check_stale_json(dir_path: Path, md_dir: Path, label: str) -> None:
) )
def scan_bottle_names(bottles_dir: Path) -> list[str]:
"""Scan `<bottles_dir>/*.md` for valid filenames and return sorted bottle names.
No file content is read. Invalid filenames are skipped with a warning."""
result: list[str] = []
if not bottles_dir.is_dir():
return result
for path in sorted(bottles_dir.glob("*.md")):
name = entity_name_from_path(path)
if name is None:
warn(
f"skipping {path}: filename must match "
f"[a-z][a-z0-9-]*.md (got {path.name!r})"
)
continue
result.append(name)
return result
def scan_agent_names(agents_dir: Path) -> dict[str, Path]: def scan_agent_names(agents_dir: Path) -> dict[str, Path]:
"""Scan `<agents_dir>/*.md` for valid filenames and return `{name: path}`. """Scan `<agents_dir>/*.md` for valid filenames and return `{name: path}`.
+2 -2
View File
@@ -18,8 +18,8 @@ _FILENAME_RX = re.compile(r"^[a-z][a-z0-9-]*$")
BOTTLE_KEYS = frozenset( BOTTLE_KEYS = frozenset(
{"env", "extends", "agent_provider", "git-gate", "egress", "supervise"} {"env", "extends", "agent_provider", "git-gate", "egress", "supervise"}
) )
AGENT_KEYS_REQUIRED = frozenset({"bottle"}) AGENT_KEYS_REQUIRED: frozenset[str] = frozenset()
AGENT_KEYS_OPTIONAL = frozenset({"skills", "git-gate"}) AGENT_KEYS_OPTIONAL = frozenset({"bottle", "skills", "git-gate"})
# Claude Code subagent fields bot-bottle ignores at launch but does # Claude Code subagent fields bot-bottle ignores at launch but does
# not reject. This lets the same file double as # not reject. This lets the same file double as
@@ -0,0 +1,216 @@
# PRD 0066: Separate agent and bottle selection
- **Status:** Active
- **Author:** claude
- **Created:** 2026-06-25
- **Issue:** #269
## Summary
Agents and bottles are two separate concerns: agents carry a system prompt and
skills; bottles carry infrastructure configuration (egress, git-gate, env,
agent provider). Today an agent's manifest file hard-codes a single `bottle:`
reference, which prevents the same agent prompt from being reused across
projects that need different bottle configurations. This PRD decouples them: at
launch time, after choosing the agent, the operator picks an ordered list of
bottles via a multi-select picker. The selected bottles are merged in order
(later entries override earlier ones) to produce the effective bottle for the
session.
## Problem
The current `bottle: <name>` field on an agent manifest file binds the agent
permanently to one bottle. To use the same system prompt with a different bottle
(e.g. `claude-implementer` at home vs. at a client site that needs a different
egress policy), the operator must duplicate the agent file and change the
`bottle:` field. Duplicate agent files drift out of sync.
## Goals / Success Criteria
1. `bottle:` in an agent's frontmatter becomes optional. Existing manifests with
`bottle:` continue to work unchanged (backward compat).
2. After selecting an agent (via the existing single-select picker), a new
multi-select bottle picker appears showing all available bottles.
3. The multi-select picker pre-populates with the agent's `bottle:` value when
present.
4. Confirming with one or more bottles selected uses those bottles, merged in
selection order, as the effective bottle for the session.
5. Confirming with an empty selection falls back to the agent's `bottle:` field.
If neither is set, a ManifestError is raised pointing the operator at the fix.
6. The ordered bottle list is stored in launch metadata so `./cli.py resume`
uses the same bottles.
7. The preflight summary (`y/N` screen) shows the effective bottle name(s).
8. The multi-select picker supports incremental filtering, Space/Enter to toggle
selection, an ordered "Selected: ..." summary line, Ctrl-D to confirm, and
Esc/q to cancel the whole start operation.
9. Unit tests cover: multi-select widget (filter, toggle, confirm, cancel),
the `cmd_start` bottle-picker step, and the manifest `load_for_agent`
runtime-bottle-merge path.
## Non-goals
- Reordering the selection list from within the picker (order = insertion order;
drag-and-drop is out of scope).
- Storing bottle selection history / MRU.
- Changes to `./cli.py edit`, `./cli.py list`, or `./cli.py info`.
- Removing the `bottle:` key from the agent schema (it stays, now optional).
## Design
### `bot_bottle/cli/tui.py``filter_multiselect`
```python
def filter_multiselect(
items: list[str],
*,
title: str = "",
initial: list[str] | None = None,
tty_path: str = "/dev/tty",
) -> list[str] | None:
"""Multi-select variant of filter_select.
Returns the ordered list of selected items, or None on cancel.
Press Space/Enter to toggle the item under the cursor.
Press Ctrl-D to confirm. Press Esc/q to cancel.
"""
```
Layout:
```
Select bottles
Filter: _
─────────────────────────────────────────
> [*] claude
[ ] dev
[ ] codex
─────────────────────────────────────────
Selected (in order): claude
─────────────────────────────────────────
[↑↓/jk] move [Space] toggle [Ctrl-D] done [Esc] cancel
```
`initial` pre-populates the ordered selection. `None` means no pre-selection.
Items added are appended in insertion order; items removed leave the remaining
order unchanged.
### `bot_bottle/manifest_schema.py` — optional `bottle:`
`bottle` moves from `AGENT_KEYS_REQUIRED` to `AGENT_KEYS_OPTIONAL`.
### `bot_bottle/manifest_agent.py` — optional `bottle:`
`ManifestAgent.bottle` changes from `str` (required) to `str = ""`.
`from_dict` no longer requires the key to be present; the bottle-exists
validation is skipped when the key is absent.
### `bot_bottle/manifest_loader.py``scan_bottle_names`
```python
def scan_bottle_names(bottles_dir: Path) -> list[str]:
"""Scan <bottles_dir>/*.md and return sorted bottle names."""
```
### `bot_bottle/manifest.py``ManifestIndex` changes
**`all_bottle_names` property** — analogous to `all_agent_names`; scans
`home_md / "bottles"` in lazy mode, returns `sorted(self.bottles.keys())` in
eager mode.
**`load_for_agent(agent_name, bottle_names: tuple[str, ...] = ())`** — new
`bottle_names` parameter. When non-empty, the listed bottles are resolved and
merged in order (index 0 is the base; each subsequent bottle is applied on top
using the same field-merge rules as `extends:`). The result replaces the bottle
that `agent.bottle` would have provided. When empty, falls back to `agent.bottle`.
Raises ManifestError if neither `bottle_names` nor `agent.bottle` is set.
### `bot_bottle/manifest_extends.py``merge_bottles_runtime`
```python
def merge_bottles_runtime(bottles: list[ManifestBottle]) -> ManifestBottle:
"""Merge an ordered list of pre-resolved ManifestBottle objects.
Index 0 is the base; each subsequent entry overrides the previous using
the same rules as the file-based extends machinery:
- env: dict merge, later wins
- git_user: per-field overlay, later wins on non-empty
- git (repos): union by name, later wins per-name
- egress.routes: concatenate
- agent_provider, supervise: later bottle's value replaces earlier
"""
```
This function operates on already-parsed `ManifestBottle` objects, so it does
not need to touch the raw-dict path.
### `bot_bottle/backend/__init__.py``BottleSpec` + `_validate`
`BottleSpec` gains `bottle_names: tuple[str, ...] = ()`.
`BottleBackend._validate` passes `spec.bottle_names` to `load_for_agent`:
```python
manifest = spec.manifest.load_for_agent(spec.agent_name, spec.bottle_names)
```
The preflight print updates `info(f"bottle: {agent.bottle}")` to display the
effective bottle name(s). When `spec.bottle_names` is non-empty those are
shown; when empty and `agent.bottle` is set, the agent's `bottle:` is shown.
### `bot_bottle/bottle_state.py` — persist bottle names
`BottleMetadata` gains `bottle_names: tuple[str, ...] = ()`. `read_metadata`
reads this from JSON (default `()`). `write_launch_metadata` passes
`spec.bottle_names` through.
### `bot_bottle/cli/start.py` — bottle multiselect step
After agent selection, before the name/color modal:
```python
available_bottle_names = manifest.all_bottle_names
# Peek at agent's bottle default for pre-population
initial_bottle = _peek_agent_bottle(manifest, agent_name)
initial = [initial_bottle] if initial_bottle else []
bottle_names_list = tui.filter_multiselect(
available_bottle_names,
title="Select bottles",
initial=initial,
)
if bottle_names_list is None:
return 0 # user cancelled
bottle_names = tuple(bottle_names_list)
```
`_peek_agent_bottle` reads the agent file's frontmatter without full parsing,
returning the `bottle:` value or `""` when absent.
`BottleSpec` is built with `bottle_names=bottle_names`.
### `bot_bottle/cli/resume.py` — bottle names from metadata
```python
spec = BottleSpec(
...
bottle_names=tuple(metadata.bottle_names),
)
```
## Implementation chunks
1. **Schema + model**`manifest_schema.py`, `manifest_agent.py` (optional
`bottle:`), `manifest_loader.py` (`scan_bottle_names`), `manifest.py`
(`all_bottle_names`, `load_for_agent` signature), `manifest_extends.py`
(`merge_bottles_runtime`), `bottle_state.py` (`bottle_names` field),
`resolve_common.py` (thread through).
2. **Backend**`BottleSpec.bottle_names`, `_validate`, preflight print.
3. **TUI**`filter_multiselect` in `tui.py` + unit tests.
4. **CLI wiring**`start.py` bottle picker step, `resume.py` metadata load.
5. **Tests**`test_cli_start_selector.py` bottle-picker cases,
`test_manifest_agent.py` optional-bottle cases, new
`test_manifest_bottle_merge.py` for `merge_bottles_runtime`.
## Open questions
None.
+227
View File
@@ -0,0 +1,227 @@
# PRD prd-new: smolmachines backend on Linux
- **Status:** Draft
- **Author:** Claude
- **Created:** 2026-06-25
- **Issue:** #283
## Summary
Make the `smolmachines` backend (PRD 0023) runnable on Linux, not
just macOS. `smolvm` already supports Linux via KVM (`/dev/kvm`);
the gap is entirely in bot-bottle's host-side glue, which hard-codes
macOS assumptions in three places:
1. **Preflight** only checks that `smolvm` is on `PATH` — it never
checks the Linux KVM prerequisite, so a misconfigured host fails
deep in the launch flow with an opaque `smolvm` error.
2. **The TSI allowlist enforcement** (`force_allowlist`) — the
security property that confines the agent VM to its sidecar
bundle's `/32`**no-ops on Linux today, failing _open_**. The
smolvm state-DB path it patches is hard-coded to macOS's
`~/Library/Application Support/...`.
3. **Per-bottle loopback scoping** (`allocate`) returns the shared
`127.0.0.1` on Linux, which would let the agent VM reach every
service on host loopback — a downgrade from the per-bottle alias
isolation macOS gets.
This PRD closes all three so a bottle launched with
`BOT_BOTTLE_BACKEND=smolmachines` on Linux gets the same isolation
guarantee it gets on macOS, and documents the Linux/NixOS host
setup. The primary validation target is NixOS, but the changes are
distro-agnostic.
## Problem
The smolmachines backend runs each bottle's agent inside a libkrun
microVM via `smolvm`, with egress confined by TSI's `--allow-cidr`
allowlist set to a single `/32` — the sidecar bundle's loopback
address. Everything else (host loopback, LAN, internet) is denied at
the VMM layer. That security property is the entire reason the
backend exists.
libkrun runs on Hypervisor.framework (macOS) **and** KVM (Linux), and
`smolvm` ships Linux x86_64 / aarch64 builds that require `/dev/kvm`.
So the microVM layer already works on Linux. What does not work is
bot-bottle's host integration, which PRD 0023 explicitly scoped to
macOS-only for v1. Three concrete blockers:
- **No KVM preflight.** On a Linux host without `/dev/kvm` (kernel
module not loaded) or without access to it (user not in the `kvm`
group), the failure surfaces as a cryptic `smolvm` non-zero exit
mid-launch instead of an actionable message.
- **TSI enforcement fails open on Linux.** `force_allowlist`
early-returns on non-macOS. It exists because `smolvm` 0.8.0
silently drops `--allow-cidr` when combined with `--from`, so the
allowlist has to be patched into smolvm's persisted state DB before
`machine start`. On Linux that patch never runs **and** the DB path
is the macOS path, so the booted VM's TSI allowlist is whatever
smolvm defaulted to — potentially all of `127.0.0.0/8`. That is the
exact sandbox-escape the backend is supposed to prevent.
- **No per-bottle loopback isolation on Linux.** `allocate` returns
`127.0.0.1` on Linux. Even with a correct allowlist, `127.0.0.1/32`
is shared by every service on host loopback, so the agent could
reach other bottles' published ports and host services. On macOS
this is solved with per-bottle `127.0.0.16..31` aliases added via
`sudo ifconfig lo0 alias`. On Linux the whole `127.0.0.0/8` is
already routed to `lo`, so docker can publish to `127.0.0.<N>`
with **no `ifconfig`/sudo step at all** — the isolation is actually
cheaper to achieve than on macOS.
## Goals / Success Criteria
- `BOT_BOTTLE_BACKEND=smolmachines ./cli.py start <agent>` launches,
runs, and tears down a bottle on a Linux host with `/dev/kvm`.
- The TSI allowlist is enforced on Linux: PRD 0022's
`tests/integration/test_sandbox_escape.py` passes against
`BOT_BOTTLE_BACKEND=smolmachines` on Linux (the acceptance gate).
- Each Linux bottle is scoped to its own `127.0.0.<N>/32`, matching
the macOS per-bottle isolation property.
- A clear, actionable preflight error when `/dev/kvm` is missing or
inaccessible, with remediation (load `kvm-intel`/`kvm-amd`, join the
`kvm` group).
- **Fail-closed:** if bot-bottle cannot positively confirm the TSI
allowlist was persisted for a machine (DB missing, row missing,
patch didn't take), it `die()`s before `machine start` rather than
booting a VM with an unverified allowlist.
- macOS behavior is unchanged.
- README documents Linux + NixOS host setup.
## Non-goals
- Rootless / non-KVM fallbacks (e.g. software emulation). Linux
smolmachines requires `/dev/kvm`, full stop.
- Removing Docker as a host dependency — the sidecar bundle and
image-build pipeline still use Docker on Linux, same as macOS.
- Auto-installing `smolvm` or configuring KVM on the operator's
behalf. Preflight reports; the operator remediates.
- Nested-virtualization tuning for running the runner itself inside a
VM (documented as a caveat, not solved here).
## Design
### Platform detection
Reuse the existing `platform.system()` check already in
`loopback_alias.py` (`_is_macos()`). "Linux" is "not macOS" for every
branch below; no new third-platform path.
### Preflight: KVM gate (`util.smolmachines_preflight`)
After the existing `smolvm`-on-`PATH` check, add a Linux-only gate:
- `/dev/kvm` must exist → else `die()` with "enable KVM
(`kvm-intel`/`kvm-amd` kernel module)".
- `/dev/kvm` must be readable + writable by the current user
(`os.access(..., R_OK | W_OK)`) → else `die()` with "add your user
to the `kvm` group (and re-login)".
macOS is unaffected (Hypervisor.framework needs no device node).
### smolvm state-DB path (platform-aware)
`loopback_alias._SMOLVM_DB_PATH` becomes platform-derived:
- macOS: `~/Library/Application Support/smolvm/server/smolvm.db`
(unchanged).
- Linux: `$XDG_DATA_HOME/smolvm/server/smolvm.db`, defaulting to
`~/.local/share/smolvm/server/smolvm.db`.
> **Verification note:** the Linux DB location is inferred from
> smolvm's documented `~/.local/share` install layout and the XDG
> base-dir spec. It must be confirmed on a real Linux smolvm install;
> if smolvm uses a different path or schema, the fail-closed check
> below turns that into a clear `die()` at launch rather than a silent
> escape.
### TSI enforcement: cross-platform + fail-closed (`force_allowlist`)
Rework `force_allowlist(machine_name, allowed_cidrs)` to run on
**both** platforms and to fail closed:
1. Resolve the state DB; if the file is missing, `die()` (cannot
confirm enforcement → refuse to launch).
2. Read the machine's persisted row; if the row is missing, `die()`.
3. If the row's `allowed_cidrs` already equals the requested list
(e.g. a newer `smolvm` that honors `--allow-cidr` at create), do
nothing — no write.
4. Otherwise patch `allowed_cidrs` (the existing BLOB-encoded write)
and re-read.
5. If, after the patch, `allowed_cidrs` still does not equal the
requested list, `die()`.
This is robust across smolvm versions: it works whether `--allow-cidr`
is silently dropped (0.8.0) or honored (newer), and it never boots a
VM whose persisted allowlist it could not confirm. It is a strict
improvement on macOS too (today's code writes unconditionally and
never verifies).
> The persisted-row check confirms our write took, not that smolvm's
> runtime TSI enforces it. The runtime guarantee is covered by the
> sandbox-escape acceptance test; the persisted check is the cheap
> fail-closed guard at launch.
### Per-bottle loopback scoping on Linux (`allocate`)
`allocate` runs the same docker-state-driven allocation on Linux as on
macOS (`_allocate_locked`, the file lock, and `_aliases_in_use` via
`docker inspect` are all already cross-platform). The only macOS-only
step, `ensure_pool` (the `sudo ifconfig lo0 alias` dance), stays
macOS-only: on Linux `127.0.0.0/8` is already loopback, so docker can
publish bundle ports directly on `127.0.0.<N>` with no setup.
Net effect: Linux bottles get per-bottle `127.0.0.16..31/32` scoping
identical to macOS, without sudo.
### Launch flow
`launch.py` needs no structural change — `_allocate_resources` already
calls `ensure_pool()` (now a Linux no-op) then `allocate()` (now
per-bottle on Linux), and `_launch_vm` already calls
`force_allowlist()` (now active on Linux). Only the macOS-specific
docstrings are updated to describe the cross-platform behavior.
## Implementation chunks
1. **Preflight KVM gate**`util.smolmachines_preflight` +
unit tests for the missing-device and no-access branches.
2. **Platform-aware DB path + fail-closed `force_allowlist`**
`loopback_alias.py`; update/extend `TestForceAllowlist`.
3. **Cross-platform `allocate`** — drop the Linux early-return; update
`TestAllocate` / `TestAllocateLock` for the new Linux behavior.
4. **Docstring + comment cleanup** in `launch.py` and module headers.
5. **Docs** — README requirements + a Linux/NixOS host-setup section.
## Testing Strategy
- **Unit (CI, any OS):** the suite mocks `platform.system()` /
`subprocess` and patches `_SMOLVM_DB_PATH`, so the new Linux
branches are testable on the macOS/Linux CI runner without `smolvm`
or KVM. Covers: KVM preflight branches, fail-closed `force_allowlist`
(DB missing, row missing, patch-doesn't-take), per-bottle Linux
allocation + locking, platform-derived DB path.
- **Integration (Linux host with KVM — the acceptance gate):**
`tests/integration/test_sandbox_escape.py` against
`BOT_BOTTLE_BACKEND=smolmachines`. This cannot run on the macOS dev
box and must be executed on NixOS before merge.
## Open questions / verification pending
- **Confirm the Linux smolvm state-DB path and schema** on a real
install (the `~/.local/share/...` inference above).
- **Confirm whether the current smolvm Linux build still drops
`--allow-cidr` with `--from`** (the 0.8.0 bug). The fail-closed
design handles either answer, but knowing lets us drop the DB patch
if upstream fixed it.
- **Confirm docker publishing to `127.0.0.<N>` on Linux** behaves as
expected end-to-end with TSI (high confidence; standard loopback
behavior, but unverified on the target host).
## References
- PRD 0023 — smolmachines bottle backend (macOS v1).
- PRD 0022 — `test_sandbox_escape.py` acceptance gate.
- PRD 0024 — sidecar bundle image.
- smolvm: https://github.com/smol-machines/smolvm
+209 -46
View File
@@ -1,7 +1,8 @@
"""Unit: cmd_start selector dispatch (PRD 0051). """Unit: cmd_start selector dispatch (PRD 0051, issue #269).
Tests that cmd_start calls filter_select only when the agent name is Tests that cmd_start calls filter_select only when the agent name is
absent, skips it when the agent is explicit, and returns 0 on cancel. absent, shows the bottle multiselect after agent selection, and skips
pickers when both are explicitly set.
All actual launch work is stubbed so no container is created. All actual launch work is stubbed so no container is created.
""" """
@@ -10,6 +11,7 @@ from __future__ import annotations
import os import os
import unittest import unittest
from collections.abc import Mapping, Sequence
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
@@ -17,10 +19,16 @@ import bot_bottle.cli.tui as tui_mod
from bot_bottle.backend import ActiveAgent from bot_bottle.backend import ActiveAgent
def _make_manifest(agent_names: list[str]): def _make_manifest(
agent_names: list[str],
bottle_names: list[str] | None = None,
agent_bottle: str = "",
):
manifest = MagicMock() manifest = MagicMock()
manifest.agents = {name: MagicMock() for name in agent_names} manifest.agents = {name: MagicMock(bottle=agent_bottle) for name in agent_names}
manifest.all_agent_names = sorted(agent_names) manifest.all_agent_names = sorted(agent_names)
manifest.all_bottle_names = sorted(bottle_names or [])
manifest.home_md = None # eager mode so _peek_agent_bottle uses agents dict
return manifest return manifest
@@ -28,27 +36,27 @@ class TestCmdStartSelector(unittest.TestCase):
"""Drive cmd_start with a minimal set of stubs.""" """Drive cmd_start with a minimal set of stubs."""
def setUp(self): def setUp(self):
# Stub Manifest.resolve so no on-disk manifest is needed. self._manifest = _make_manifest(["researcher", "implementer"], ["claude", "dev"])
self._manifest = _make_manifest(["researcher", "implementer"])
self._resolve_patch = patch( self._resolve_patch = patch(
"bot_bottle.cli.start.ManifestIndex.resolve", "bot_bottle.cli.start.ManifestIndex.resolve",
return_value=self._manifest, return_value=self._manifest,
) )
self._resolve_patch.start() self._resolve_patch.start()
# Stub _launch_bottle so no real container work happens.
self._launch_patch = patch( self._launch_patch = patch(
"bot_bottle.cli.start._launch_bottle", "bot_bottle.cli.start._launch_bottle",
return_value=0, return_value=0,
) )
self._launch_mock = self._launch_patch.start() self._launch_mock = self._launch_patch.start()
# Stub filter_select to avoid opening /dev/tty. # Stub filter_select (agent picker) and filter_multiselect (bottle picker).
self._tui_patch = patch.object(tui_mod, "filter_select") self._agent_picker_patch = patch.object(tui_mod, "filter_select")
self._tui_mock = self._tui_patch.start() self._agent_picker_mock = self._agent_picker_patch.start()
self._bottle_picker_patch = patch.object(tui_mod, "filter_multiselect")
self._bottle_picker_mock = self._bottle_picker_patch.start()
self._bottle_picker_mock.return_value = ["claude"] # default: one bottle selected
# Ensure BOT_BOTTLE_BACKEND is absent so omitted --backend
# flows through to the resolver default.
self._env_patch = patch.dict(os.environ, {}, clear=False) self._env_patch = patch.dict(os.environ, {}, clear=False)
self._env_patch.start() self._env_patch.start()
os.environ.pop("BOT_BOTTLE_BACKEND", None) os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -56,50 +64,108 @@ class TestCmdStartSelector(unittest.TestCase):
def tearDown(self): def tearDown(self):
self._resolve_patch.stop() self._resolve_patch.stop()
self._launch_patch.stop() self._launch_patch.stop()
self._tui_patch.stop() self._agent_picker_patch.stop()
self._bottle_picker_patch.stop()
self._env_patch.stop() self._env_patch.stop()
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Both explicit — no picker shown # Agent explicit — agent picker skipped; bottle picker always shown
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def test_both_explicit_skips_picker(self): def test_explicit_agent_skips_agent_picker(self):
self._tui_mock.return_value = "researcher"
rc = start_mod.cmd_start(["--backend=docker", "researcher"]) rc = start_mod.cmd_start(["--backend=docker", "researcher"])
self.assertEqual(0, rc) self.assertEqual(0, rc)
self._tui_mock.assert_not_called() self._agent_picker_mock.assert_not_called()
self._bottle_picker_mock.assert_called_once()
self._launch_mock.assert_called_once() self._launch_mock.assert_called_once()
_, kwargs = self._launch_mock.call_args
self.assertEqual("docker", kwargs["backend_name"]) def test_explicit_agent_bottle_picker_shows_available_bottles(self):
start_mod.cmd_start(["researcher"])
call_kwargs = self._bottle_picker_mock.call_args
self.assertEqual(["claude", "dev"], call_kwargs[0][0])
self.assertIn("bottle", call_kwargs[1]["title"].lower())
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Agent absent → agent picker fires; backend explicit # Agent absent → agent picker fires; bottle picker always follows
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def test_agent_absent_shows_agent_picker(self): def test_agent_absent_shows_agent_picker(self):
self._tui_mock.return_value = "researcher" self._agent_picker_mock.return_value = "researcher"
rc = start_mod.cmd_start(["--backend=docker"]) rc = start_mod.cmd_start(["--backend=docker"])
self.assertEqual(0, rc) self.assertEqual(0, rc)
self._tui_mock.assert_called_once() self._agent_picker_mock.assert_called_once()
call_kwargs = self._tui_mock.call_args call_kwargs = self._agent_picker_mock.call_args
self.assertEqual(["implementer", "researcher"], call_kwargs[0][0]) self.assertEqual(["implementer", "researcher"], call_kwargs[0][0])
self.assertIn("agent", call_kwargs[1]["title"].lower()) self.assertIn("agent", call_kwargs[1]["title"].lower())
# Bottle picker must also fire after agent selection.
self._bottle_picker_mock.assert_called_once()
def test_agent_picker_cancel_returns_0(self): def test_agent_picker_cancel_skips_bottle_picker(self):
self._tui_mock.return_value = None self._agent_picker_mock.return_value = None
rc = start_mod.cmd_start(["--backend=docker"]) rc = start_mod.cmd_start(["--backend=docker"])
self.assertEqual(0, rc) self.assertEqual(0, rc)
self._bottle_picker_mock.assert_not_called()
self._launch_mock.assert_not_called()
def test_bottle_picker_cancel_returns_0(self):
self._bottle_picker_mock.return_value = None
rc = start_mod.cmd_start(["researcher"])
self.assertEqual(0, rc)
self._launch_mock.assert_not_called() self._launch_mock.assert_not_called()
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Agent explicit, backend absent → no picker # Bottle selection is forwarded to BottleSpec
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def test_backend_absent_uses_default_without_picker(self): def test_selected_bottles_forwarded_to_spec(self):
rc = start_mod.cmd_start(["researcher"]) self._bottle_picker_mock.return_value = ["claude", "dev"]
self.assertEqual(0, rc) start_mod.cmd_start(["researcher"])
self._tui_mock.assert_not_called()
self._launch_mock.assert_called_once() self._launch_mock.assert_called_once()
spec = self._launch_mock.call_args[0][0]
self.assertEqual(("claude", "dev"), spec.bottle_names)
def test_empty_bottle_selection_forwarded(self):
self._bottle_picker_mock.return_value = []
start_mod.cmd_start(["researcher"])
self._launch_mock.assert_called_once()
spec = self._launch_mock.call_args[0][0]
self.assertEqual((), spec.bottle_names)
# ------------------------------------------------------------------
# Agent default bottle pre-populates the picker
# ------------------------------------------------------------------
def test_agent_bottle_prepopulates_bottle_picker(self):
manifest = _make_manifest(
["implementer"], ["claude", "dev"], agent_bottle="claude"
)
with patch(
"bot_bottle.cli.start.ManifestIndex.resolve", return_value=manifest
):
start_mod.cmd_start(["implementer"])
call_kwargs = self._bottle_picker_mock.call_args
self.assertEqual(["claude"], call_kwargs[1]["initial"])
def test_no_agent_bottle_empty_initial(self):
manifest = _make_manifest(["researcher"], ["claude", "dev"], agent_bottle="")
with patch(
"bot_bottle.cli.start.ManifestIndex.resolve", return_value=manifest
):
start_mod.cmd_start(["researcher"])
call_kwargs = self._bottle_picker_mock.call_args
self.assertEqual([], call_kwargs[1]["initial"])
# ------------------------------------------------------------------
# Backend wiring
# ------------------------------------------------------------------
def test_explicit_backend_forwarded(self):
start_mod.cmd_start(["--backend=docker", "researcher"])
_, kwargs = self._launch_mock.call_args
self.assertEqual("docker", kwargs["backend_name"])
def test_absent_backend_uses_default(self):
start_mod.cmd_start(["researcher"])
_, kwargs = self._launch_mock.call_args _, kwargs = self._launch_mock.call_args
self.assertIsNone(kwargs["backend_name"]) self.assertIsNone(kwargs["backend_name"])
@@ -110,28 +176,21 @@ class TestCmdStartSelector(unittest.TestCase):
finally: finally:
os.environ.pop("BOT_BOTTLE_BACKEND", None) os.environ.pop("BOT_BOTTLE_BACKEND", None)
self.assertEqual(0, rc) self.assertEqual(0, rc)
self._tui_mock.assert_not_called()
# ------------------------------------------------------------------ def test_both_absent_shows_agent_picker_then_bottle_picker(self):
# Both absent → only agent picker self._agent_picker_mock.return_value = "researcher"
# ------------------------------------------------------------------
def test_both_absent_shows_only_agent_picker(self):
self._tui_mock.return_value = "researcher"
rc = start_mod.cmd_start([]) rc = start_mod.cmd_start([])
self.assertEqual(0, rc) self.assertEqual(0, rc)
self._tui_mock.assert_called_once() self._agent_picker_mock.assert_called_once()
title = self._tui_mock.call_args[1]["title"].lower() self._bottle_picker_mock.assert_called_once()
self.assertIn("agent", title)
self._launch_mock.assert_called_once() self._launch_mock.assert_called_once()
_, kwargs = self._launch_mock.call_args
self.assertIsNone(kwargs["backend_name"])
def test_both_absent_agent_cancel_skips_backend_picker(self): def test_both_absent_agent_cancel_skips_bottle_and_launch(self):
self._tui_mock.side_effect = [None] self._agent_picker_mock.return_value = None
rc = start_mod.cmd_start([]) rc = start_mod.cmd_start([])
self.assertEqual(0, rc) self.assertEqual(0, rc)
self.assertEqual(1, self._tui_mock.call_count) self._agent_picker_mock.assert_called_once()
self._bottle_picker_mock.assert_not_called()
self._launch_mock.assert_not_called() self._launch_mock.assert_not_called()
@@ -149,11 +208,13 @@ class TestCmdStartLabelCollision(unittest.TestCase):
"""cmd_start re-prompts when the label's slug is already running.""" """cmd_start re-prompts when the label's slug is already running."""
def setUp(self): def setUp(self):
self._manifest = _make_manifest(["researcher"]) self._manifest = _make_manifest(["researcher"], ["claude"])
patch("bot_bottle.cli.start.ManifestIndex.resolve", return_value=self._manifest).start() patch("bot_bottle.cli.start.ManifestIndex.resolve", return_value=self._manifest).start()
self._launch_mock = patch( self._launch_mock = patch(
"bot_bottle.cli.start._launch_bottle", return_value=0, "bot_bottle.cli.start._launch_bottle", return_value=0,
).start() ).start()
# Stub the bottle picker to always return a selection.
patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start()
self.addCleanup(patch.stopall) self.addCleanup(patch.stopall)
def test_no_collision_proceeds_without_reprompt(self): def test_no_collision_proceeds_without_reprompt(self):
@@ -193,5 +254,107 @@ class TestCmdStartLabelCollision(unittest.TestCase):
self.assertIn("already in use", second_call_kwargs.get("disclaimer", "")) self.assertIn("already in use", second_call_kwargs.get("disclaimer", ""))
class TestBottleLineage(unittest.TestCase):
"""Unit tests for _bottle_lineage."""
def test_returns_empty_in_eager_mode(self):
manifest = _make_manifest(["agent"], ["base", "dev"])
# home_md is None in eager mode → no file reads, returns {}
result = start_mod._bottle_lineage(manifest)
self.assertEqual({}, result)
def test_reads_extends_chain_from_files(self):
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as tmp:
bottles_dir = Path(tmp) / "bottles"
bottles_dir.mkdir()
(bottles_dir / "base.md").write_text("---\n{}\n---\n")
(bottles_dir / "mid.md").write_text("---\nextends: base\n---\n")
(bottles_dir / "leaf.md").write_text("---\nextends: mid\n---\n")
manifest = MagicMock()
manifest.home_md = Path(tmp)
result = start_mod._bottle_lineage(manifest)
self.assertNotIn("base", result) # no parent → not in map
self.assertEqual("base -> mid", result["mid"])
self.assertEqual("base -> mid -> leaf", result["leaf"])
def test_cycle_protection(self):
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as tmp:
bottles_dir = Path(tmp) / "bottles"
bottles_dir.mkdir()
(bottles_dir / "a.md").write_text("---\nextends: b\n---\n")
(bottles_dir / "b.md").write_text("---\nextends: a\n---\n")
manifest = MagicMock()
manifest.home_md = Path(tmp)
result = start_mod._bottle_lineage(manifest)
# Cycle must not hang; each should get a two-element chain.
for name in ("a", "b"):
self.assertIn(name, result)
self.assertIn("->", result[name])
class TestManifestToYaml(unittest.TestCase):
"""Unit tests for _manifest_to_yaml."""
def _make_manifest_obj(
self,
*,
skills: Sequence[str] = (),
env: Mapping[str, str] | None = None,
supervise: bool = True,
agent_provider_template: str = "claude",
):
from bot_bottle.manifest import Manifest, ManifestBottle
from bot_bottle.manifest_agent import ManifestAgent, ManifestAgentProvider
agent = ManifestAgent(skills=tuple(skills))
bottle = ManifestBottle(
env=env or {},
supervise=supervise,
agent_provider=ManifestAgentProvider(template=agent_provider_template),
)
return Manifest(agent=agent, bottle=bottle)
def test_includes_agent_section(self):
m = self._make_manifest_obj(skills=["researcher"])
yaml = start_mod._manifest_to_yaml(m)
self.assertIn("agent:", yaml)
self.assertIn("- researcher", yaml)
def test_includes_bottle_section(self):
m = self._make_manifest_obj(env={"FOO": "bar"})
yaml = start_mod._manifest_to_yaml(m)
self.assertIn("bottle:", yaml)
self.assertIn("FOO: bar", yaml)
def test_supervise_rendered(self):
m_true = self._make_manifest_obj(supervise=True)
m_false = self._make_manifest_obj(supervise=False)
self.assertIn("supervise: true", start_mod._manifest_to_yaml(m_true))
self.assertIn("supervise: false", start_mod._manifest_to_yaml(m_false))
def test_non_claude_provider_shown(self):
m = self._make_manifest_obj(agent_provider_template="codex")
yaml = start_mod._manifest_to_yaml(m)
self.assertIn("agent_provider:", yaml)
self.assertIn("template: codex", yaml)
def test_default_claude_provider_omitted(self):
m = self._make_manifest_obj(agent_provider_template="claude")
yaml = start_mod._manifest_to_yaml(m)
self.assertNotIn("agent_provider:", yaml)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+128 -2
View File
@@ -1,4 +1,4 @@
"""Unit tests for bot_bottle.cli.tui — filter_select internals. """Unit tests for bot_bottle.cli.tui — filter_select and filter_multiselect.
We test the pure-Python logic (_filter_items, cursor movement, confirm, We test the pure-Python logic (_filter_items, cursor movement, confirm,
cancel) by exercising the internal helpers directly, without spinning up cancel) by exercising the internal helpers directly, without spinning up
@@ -8,8 +8,15 @@ a real curses session (which requires a TTY).
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from typing import Any, Optional
from bot_bottle.cli.tui import _filter_items, filter_select from bot_bottle.cli.tui import _filter_items, _multiselect_loop, filter_multiselect, filter_select
_KEY_SPACE = 32
_KEY_ENTER = 10
_KEY_ESC = 27
_KEY_CTRL_D = 4
class TestFilterItems(unittest.TestCase): class TestFilterItems(unittest.TestCase):
@@ -46,5 +53,124 @@ class TestFilterSelectEmptyItems(unittest.TestCase):
self.assertIsNone(result) self.assertIsNone(result)
class TestFilterMultiselectEmptyItems(unittest.TestCase):
def test_returns_empty_list_for_empty_items(self):
# No TTY needed — short-circuits before opening tty.
result = filter_multiselect([], title="Select", tty_path="/dev/null")
self.assertEqual([], result)
def test_returns_none_when_tty_unavailable(self):
result = filter_multiselect(["a", "b"], tty_path="/nonexistent/tty")
self.assertIsNone(result)
class TestMultiselectLoopReordering(unittest.TestCase):
"""Exercise _multiselect_loop key handling without a real curses terminal.
We drive the loop via a fake screen that feeds a pre-recorded key sequence
and records what was drawn we only need the return value, so the fake
screen's getch() raises StopIteration after the key list is exhausted, and
the loop is expected to return before that via Ctrl-D.
"""
def _run(self, keys: list[int], items: list[str], initial: list[str]) -> Optional[list[str]]:
"""Run _multiselect_loop with a synthetic screen feeding `keys`."""
key_iter = iter(keys)
class FakeScreen:
def erase(self) -> None: pass
def getmaxyx(self) -> tuple[int, int]: return (40, 80)
def refresh(self) -> None: pass
def getch(self) -> int: return next(key_iter)
def addstr(self, *a: Any) -> None: pass
def keypad(self, *a: Any) -> None: pass
return _multiselect_loop(FakeScreen(), items, title="", initial=initial) # type: ignore[arg-type]
def test_ctrl_d_confirms_initial_selection(self):
result = self._run([_KEY_CTRL_D], ["a", "b", "c"], ["a", "b"])
self.assertEqual(["a", "b"], result)
def test_esc_cancels(self):
result = self._run([_KEY_ESC], ["a", "b"], ["a"])
self.assertIsNone(result)
def test_tab_then_K_moves_item_up(self):
# Start: selected = ["a", "b", "c"]
# Tab → order mode (order_cursor=0 on "a")
# ↓ → order_cursor=1 (on "b")
# K → swap b and a → ["b", "a", "c"], order_cursor=0
# Ctrl-D → confirm
DOWN = ord("j")
result = self._run(
[ord("\t"), DOWN, ord("K"), _KEY_CTRL_D],
["a", "b", "c"],
["a", "b", "c"],
)
self.assertEqual(["b", "a", "c"], result)
def test_tab_then_J_moves_item_down(self):
# selected = ["a", "b", "c"], focus order, cursor=0
# J → swap a and b → ["b", "a", "c"], cursor=1
# Ctrl-D → confirm
result = self._run(
[ord("\t"), ord("J"), _KEY_CTRL_D],
["a", "b", "c"],
["a", "b", "c"],
)
self.assertEqual(["b", "a", "c"], result)
def test_K_at_top_is_no_op(self):
# cursor already at 0, K should not change order
result = self._run(
[ord("\t"), ord("K"), _KEY_CTRL_D],
["a", "b"],
["a", "b"],
)
self.assertEqual(["a", "b"], result)
def test_J_at_bottom_is_no_op(self):
DOWN = ord("j")
result = self._run(
[ord("\t"), DOWN, ord("J"), _KEY_CTRL_D],
["a", "b"],
["a", "b"],
)
self.assertEqual(["a", "b"], result)
def test_tab_back_to_filter_then_confirm(self):
# Tab → order, Tab → filter, Ctrl-D confirms unchanged
result = self._run(
[ord("\t"), ord("\t"), _KEY_CTRL_D],
["a", "b"],
["a", "b"],
)
self.assertEqual(["a", "b"], result)
def test_space_toggles_item_on(self):
# Space on an unselected item selects it; Ctrl-D confirms.
result = self._run([_KEY_SPACE, _KEY_CTRL_D], ["a", "b"], [])
self.assertEqual(["a"], result)
def test_space_toggles_item_off(self):
# Space on a selected item deselects it; Ctrl-D confirms empty.
result = self._run([_KEY_SPACE, _KEY_CTRL_D], ["a", "b"], ["a"])
self.assertEqual([], result)
def test_enter_confirms_without_toggle(self):
# Enter immediately confirms the current selection without toggling.
result = self._run([_KEY_ENTER], ["a", "b"], ["a"])
self.assertEqual(["a"], result)
def test_enter_confirms_empty_selection(self):
result = self._run([_KEY_ENTER], ["a", "b"], [])
self.assertEqual([], result)
def test_space_then_enter_confirms(self):
# Space selects "a", Enter confirms.
result = self._run([_KEY_SPACE, _KEY_ENTER], ["a", "b"], [])
self.assertEqual(["a"], result)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+200
View File
@@ -0,0 +1,200 @@
"""Unit: runtime bottle composition (issue #269).
Tests for merge_bottles_runtime and ManifestIndex.load_for_agent with
the new bottle_names parameter.
"""
from __future__ import annotations
import os
import shutil
import tempfile
import textwrap
import unittest
from pathlib import Path
from bot_bottle.manifest import ManifestBottle, ManifestError, ManifestIndex
from bot_bottle.manifest_extends import merge_bottles_runtime
def _index(bottles: dict[str, object], agents: dict[str, object]) -> ManifestIndex:
return ManifestIndex.from_json_obj({"bottles": bottles, "agents": agents})
def _bottle(**kwargs: object) -> ManifestBottle:
return ManifestBottle.from_dict("test", kwargs)
class TestMergeBottlesRuntime(unittest.TestCase):
def test_single_bottle_returns_as_is(self):
b = _bottle(env={"FOO": "1"})
result = merge_bottles_runtime([b])
self.assertEqual({"FOO": "1"}, dict(result.env))
def test_env_later_wins(self):
base = _bottle(env={"FOO": "base", "ONLY_BASE": "x"})
override = _bottle(env={"FOO": "override", "ONLY_OVERRIDE": "y"})
result = merge_bottles_runtime([base, override])
self.assertEqual("override", result.env["FOO"])
self.assertEqual("x", result.env["ONLY_BASE"])
self.assertEqual("y", result.env["ONLY_OVERRIDE"])
def test_egress_routes_concatenated(self):
from bot_bottle.manifest_egress import ManifestEgressConfig, ManifestEgressRoute
r1 = ManifestEgressRoute(Host="api.a.com")
r2 = ManifestEgressRoute(Host="api.b.com")
base = ManifestBottle(egress=ManifestEgressConfig(routes=(r1,)))
override = ManifestBottle(egress=ManifestEgressConfig(routes=(r2,)))
result = merge_bottles_runtime([base, override])
hosts = [r.Host for r in result.egress.routes]
self.assertIn("api.a.com", hosts)
self.assertIn("api.b.com", hosts)
def test_supervise_later_wins(self):
base = _bottle(supervise=True)
override = _bottle(supervise=False)
result = merge_bottles_runtime([base, override])
self.assertFalse(result.supervise)
def test_three_bottles_merged_left_to_right(self):
b1 = _bottle(env={"A": "1", "B": "1", "C": "1"})
b2 = _bottle(env={"B": "2", "C": "2"})
b3 = _bottle(env={"C": "3"})
result = merge_bottles_runtime([b1, b2, b3])
self.assertEqual("1", result.env["A"])
self.assertEqual("2", result.env["B"])
self.assertEqual("3", result.env["C"])
def test_empty_list_raises(self):
with self.assertRaises(ValueError):
merge_bottles_runtime([])
class TestLoadForAgentWithBottleNames(unittest.TestCase):
def test_bottle_names_override_agent_bottle(self):
idx = _index(
bottles={
"base": {"env": {"X": "base"}},
"override": {"env": {"X": "override"}},
},
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
)
m = idx.load_for_agent("impl", ("override",))
self.assertEqual("override", m.bottle.env["X"])
def test_bottle_names_merged_in_order(self):
idx = _index(
bottles={
"a": {"env": {"X": "a", "A": "only-a"}},
"b": {"env": {"X": "b", "B": "only-b"}},
},
agents={"impl": {"bottle": "a", "skills": [], "prompt": ""}},
)
m = idx.load_for_agent("impl", ("a", "b"))
self.assertEqual("b", m.bottle.env["X"])
self.assertEqual("only-a", m.bottle.env["A"])
self.assertEqual("only-b", m.bottle.env["B"])
def test_empty_bottle_names_uses_agent_bottle(self):
idx = _index(
bottles={"base": {"env": {"X": "base"}}},
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
)
m = idx.load_for_agent("impl", ())
self.assertEqual("base", m.bottle.env["X"])
def test_no_bottle_and_no_bottle_names_raises(self):
idx = _index(
bottles={"base": {}},
agents={"impl": {"skills": [], "prompt": ""}},
)
with self.assertRaises(ManifestError) as ctx:
idx.load_for_agent("impl", ())
self.assertIn("no 'bottle' field", str(ctx.exception))
def test_unknown_bottle_name_raises(self):
idx = _index(
bottles={"base": {}},
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
)
with self.assertRaises(ManifestError) as ctx:
idx.load_for_agent("impl", ("nonexistent",))
self.assertIn("nonexistent", str(ctx.exception))
def test_agent_without_bottle_works_with_bottle_names(self):
idx = _index(
bottles={"base": {"env": {"X": "base"}}},
agents={"impl": {"skills": [], "prompt": ""}},
)
m = idx.load_for_agent("impl", ("base",))
self.assertEqual("base", m.bottle.env["X"])
class TestAllBottleNames(unittest.TestCase):
def test_eager_mode_returns_bottle_names(self):
idx = _index(
bottles={"alpha": {}, "beta": {}, "gamma": {}},
agents={"impl": {"bottle": "alpha", "skills": [], "prompt": ""}},
)
self.assertEqual(["alpha", "beta", "gamma"], idx.all_bottle_names)
def test_lazy_mode_scans_files(self):
home = Path(tempfile.mkdtemp(prefix="cb-home-"))
orig_home = os.environ.get("HOME")
os.environ["HOME"] = str(home)
try:
bottles_dir = home / ".bot-bottle" / "bottles"
agents_dir = home / ".bot-bottle" / "agents"
bottles_dir.mkdir(parents=True)
agents_dir.mkdir(parents=True)
(bottles_dir / "claude.md").write_text("---\n---\n")
(bottles_dir / "dev.md").write_text("---\n---\n")
(agents_dir / "impl.md").write_text("---\nbottle: claude\n---\n")
idx = ManifestIndex.resolve(str(home))
self.assertEqual(["claude", "dev"], idx.all_bottle_names)
finally:
if orig_home is None:
os.environ.pop("HOME", None)
else:
os.environ["HOME"] = orig_home
shutil.rmtree(home, ignore_errors=True)
class TestAgentOptionalBottleMd(unittest.TestCase):
"""Agent file without bottle: works when bottle_names are provided at launch."""
def setUp(self) -> None:
self.home = Path(tempfile.mkdtemp(prefix="cb-home-"))
self._orig_home = os.environ.get("HOME")
os.environ["HOME"] = str(self.home)
def tearDown(self) -> None:
if self._orig_home is None:
os.environ.pop("HOME", None)
else:
os.environ["HOME"] = self._orig_home
shutil.rmtree(self.home, ignore_errors=True)
def _write(self, rel: str, text: str) -> None:
p = self.home / ".bot-bottle" / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(textwrap.dedent(text).lstrip("\n"))
def test_agent_without_bottle_resolves_with_bottle_names(self):
self._write("bottles/dev.md", "---\nenv:\n X: dev\n---\n")
self._write("agents/impl.md", "---\n---\nimpl agent.\n")
idx = ManifestIndex.resolve(str(self.home))
m = idx.load_for_agent("impl", ("dev",))
self.assertEqual("dev", m.bottle.env["X"])
def test_agent_without_bottle_fails_without_bottle_names(self):
self._write("bottles/dev.md", "---\n---\n")
self._write("agents/impl.md", "---\n---\nimpl agent.\n")
idx = ManifestIndex.resolve(str(self.home))
with self.assertRaises(ManifestError) as ctx:
idx.load_for_agent("impl", ())
self.assertIn("no 'bottle' field", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
+119 -16
View File
@@ -8,6 +8,7 @@ inspecting running bundle containers' port bindings."""
from __future__ import annotations from __future__ import annotations
import json import json
import os
import sqlite3 import sqlite3
import subprocess import subprocess
import tempfile import tempfile
@@ -112,9 +113,16 @@ class TestEnsurePool(unittest.TestCase):
class TestAllocate(unittest.TestCase): class TestAllocate(unittest.TestCase):
def test_returns_loopback_on_linux(self): def test_per_bottle_alias_on_linux(self):
with patch.object(loopback_alias, "_is_macos", return_value=False): # Linux gets the same per-bottle scoping as macOS (127/8 is
self.assertEqual("127.0.0.1", loopback_alias.allocate("demo")) # already loopback, so no ifconfig is needed). A fresh host
# with no running bundles allocates the first pool entry.
with tempfile.TemporaryDirectory() as tmp:
lock_path = Path(tmp) / "smolmachines.lock"
with patch.object(loopback_alias, "_is_macos", return_value=False), \
patch.object(loopback_alias, "_ALLOC_LOCK_PATH", lock_path), \
patch.object(loopback_alias, "_aliases_in_use", return_value=set()):
self.assertEqual("127.0.0.16", loopback_alias.allocate("demo"))
def test_picks_lowest_unused_on_macos(self): def test_picks_lowest_unused_on_macos(self):
# No bundles running -> first pool entry. # No bundles running -> first pool entry.
@@ -166,12 +174,25 @@ class TestAllocateLock(unittest.TestCase):
self.assertIn(fcntl_mod.LOCK_EX, flock_calls) self.assertIn(fcntl_mod.LOCK_EX, flock_calls)
def test_no_lock_on_linux(self): def test_acquires_exclusive_lock_on_linux(self):
# Linux early-returns before touching the lock file. # Linux allocates per-bottle too, so it must take the same
with patch.object(loopback_alias, "_is_macos", return_value=False), \ # lock to serialise concurrent launches.
patch.object(loopback_alias.fcntl, "flock") as flock: import fcntl as fcntl_mod
loopback_alias.allocate("demo") flock_calls: list[int] = []
flock.assert_not_called()
def record_flock(fd, op): # type: ignore
flock_calls.append(op)
with tempfile.TemporaryDirectory() as tmp:
lock_path = Path(tmp) / "smolmachines.lock"
with patch.object(loopback_alias, "_is_macos", return_value=False), \
patch.object(loopback_alias, "_ALLOC_LOCK_PATH", lock_path), \
patch.object(loopback_alias, "_aliases_in_use", return_value=set()), \
patch.object(loopback_alias.fcntl, "flock",
side_effect=record_flock):
loopback_alias.allocate("demo")
self.assertIn(fcntl_mod.LOCK_EX, flock_calls)
def test_sequential_allocations_with_shared_lock_are_serialised(self): def test_sequential_allocations_with_shared_lock_are_serialised(self):
# Two sequential calls share the same lock file. The second # Two sequential calls share the same lock file. The second
@@ -241,10 +262,12 @@ class TestAliasInUseDetection(unittest.TestCase):
class TestForceAllowlist(unittest.TestCase): class TestForceAllowlist(unittest.TestCase):
"""Smolvm 0.8.0 silently drops `--allow-cidr` with `--from`, """Smolvm 0.8.0 silently drops `--allow-cidr` with `--from`, so
so `force_allowlist` opens the state DB directly and sets `force_allowlist` opens the state DB directly and sets the row's
the row's `allowed_cidrs` field. Round-trip tests against a `allowed_cidrs` field on both macOS and Linux. It is
real SQLite DB to lock down the BLOB encoding.""" fail-closed: it dies rather than launching a VM whose allowlist
it can't confirm. Round-trip tests against a real SQLite DB to
lock down the BLOB encoding."""
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="smolvm-db.") self._tmp = tempfile.TemporaryDirectory(prefix="smolvm-db.")
@@ -290,17 +313,67 @@ class TestForceAllowlist(unittest.TestCase):
self.assertEqual(4, cfg["cpus"]) self.assertEqual(4, cfg["cpus"])
self.assertTrue(cfg["network"]) self.assertTrue(cfg["network"])
def test_noop_on_linux(self): def test_patches_on_linux_too(self):
# force_allowlist no longer no-ops on Linux — the TSI
# allowlist must be enforced there as well.
with patch.object(loopback_alias, "_is_macos", return_value=False), \ with patch.object(loopback_alias, "_is_macos", return_value=False), \
patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db): patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db):
loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"]) loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"])
# DB row should be untouched.
con = sqlite3.connect(str(self.db)) con = sqlite3.connect(str(self.db))
cfg = json.loads(con.execute( cfg = json.loads(con.execute(
"SELECT data FROM vms WHERE name='demo-vm'", "SELECT data FROM vms WHERE name='demo-vm'",
).fetchone()[0]) ).fetchone()[0])
con.close() con.close()
self.assertIsNone(cfg["allowed_cidrs"]) self.assertEqual(["127.0.0.16/32"], cfg["allowed_cidrs"])
def test_skips_write_when_already_matching(self):
# A newer smolvm that honors --allow-cidr at create leaves the
# row already correct; force_allowlist must not rewrite it. We
# detect a no-write by comparing the raw BLOB byte-for-byte
# (a rewrite re-serialises the JSON, changing key order/bytes
# is not guaranteed, but mtime/identity isn't observable — so
# we assert the stored bytes are exactly what we pre-seeded).
seeded = json.dumps({
"name": "demo-vm", "cpus": 4, "mem": 8192,
"network": True, "allowed_cidrs": ["127.0.0.16/32"],
}).encode()
con = sqlite3.connect(str(self.db))
con.execute(
"UPDATE vms SET data=? WHERE name='demo-vm'",
(sqlite3.Binary(seeded),),
)
con.commit()
con.close()
with patch.object(loopback_alias, "_is_macos", return_value=True), \
patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db):
loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"])
con = sqlite3.connect(str(self.db))
stored = con.execute(
"SELECT data FROM vms WHERE name='demo-vm'").fetchone()[0]
con.close()
self.assertEqual(seeded, bytes(stored))
def test_dies_when_patch_does_not_take(self):
# If the persisted allowlist still doesn't match after the
# patch (e.g. wrong schema / smolvm stores it elsewhere),
# force_allowlist must fail closed rather than boot the VM.
original = loopback_alias._read_machine_cfg
def stale_cfg(con, name):
# Always report the un-patched row so the post-write
# verification never sees the requested cidrs.
cfg = original(con, name)
cfg["allowed_cidrs"] = None
return cfg
with patch.object(loopback_alias, "_is_macos", return_value=True), \
patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db), \
patch.object(loopback_alias, "_read_machine_cfg", side_effect=stale_cfg), \
patch.object(loopback_alias, "die", side_effect=SystemExit("die")):
with self.assertRaises(SystemExit):
loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"])
def test_dies_on_missing_db(self): def test_dies_on_missing_db(self):
with patch.object(loopback_alias, "_is_macos", return_value=True), \ with patch.object(loopback_alias, "_is_macos", return_value=True), \
@@ -323,5 +396,35 @@ class TestForceAllowlist(unittest.TestCase):
loopback_alias.force_allowlist("not-in-db", ["127.0.0.16/32"]) loopback_alias.force_allowlist("not-in-db", ["127.0.0.16/32"])
class TestSmolvmDbPath(unittest.TestCase):
"""The smolvm state-DB path is platform-derived: Application
Support on macOS, XDG data dir on Linux."""
def test_macos_path(self):
with patch.object(loopback_alias.platform, "system", return_value="Darwin"):
p = loopback_alias._smolvm_db_path()
self.assertEqual(
("Library", "Application Support", "smolvm", "server", "smolvm.db"),
p.parts[-5:],
)
def test_linux_default_xdg_path(self):
env = {k: v for k, v in os.environ.items() if k != "XDG_DATA_HOME"}
with patch.object(loopback_alias.platform, "system", return_value="Linux"), \
patch.dict(loopback_alias.os.environ, env, clear=True):
p = loopback_alias._smolvm_db_path()
self.assertEqual(
(".local", "share", "smolvm", "server", "smolvm.db"),
p.parts[-5:],
)
def test_linux_respects_xdg_data_home(self):
with patch.object(loopback_alias.platform, "system", return_value="Linux"), \
patch.dict(loopback_alias.os.environ,
{"XDG_DATA_HOME": "/custom/data"}, clear=False):
p = loopback_alias._smolvm_db_path()
self.assertEqual(Path("/custom/data/smolvm/server/smolvm.db"), p)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+63
View File
@@ -56,9 +56,14 @@ class TestBundleSubnet(unittest.TestCase):
class TestPreflight(unittest.TestCase): class TestPreflight(unittest.TestCase):
def test_smolvm_present_returns_none(self): def test_smolvm_present_returns_none(self):
# Pin macOS so the Linux KVM gate doesn't fire on a CI runner
# (ubuntu, no /dev/kvm) — this test isolates the PATH check.
with patch( with patch(
"bot_bottle.backend.smolmachines.util.shutil.which", "bot_bottle.backend.smolmachines.util.shutil.which",
return_value="/usr/local/bin/smolvm", return_value="/usr/local/bin/smolvm",
), patch(
"bot_bottle.backend.smolmachines.util.platform.system",
return_value="Darwin",
): ):
self.assertIsNone(smolmachines_preflight()) self.assertIsNone(smolmachines_preflight())
@@ -88,5 +93,63 @@ class TestPreflight(unittest.TestCase):
self.assertIn("BOT_BOTTLE_BACKEND=docker", msg) self.assertIn("BOT_BOTTLE_BACKEND=docker", msg)
class TestKvmPreflight(unittest.TestCase):
"""Linux-only KVM gate: smolvm needs /dev/kvm present and
accessible. macOS skips this entirely (Hypervisor.framework)."""
def _run(self, *, system, exists, access):
with patch(
"bot_bottle.backend.smolmachines.util.shutil.which",
return_value="/usr/bin/smolvm",
), patch(
"bot_bottle.backend.smolmachines.util.platform.system",
return_value=system,
), patch(
"bot_bottle.backend.smolmachines.util.os.path.exists",
return_value=exists,
), patch(
"bot_bottle.backend.smolmachines.util.os.access",
return_value=access,
):
return smolmachines_preflight()
def test_macos_skips_kvm_check(self):
# Even with /dev/kvm absent, macOS must not run the gate.
self.assertIsNone(self._run(system="Darwin", exists=False, access=False))
def test_linux_ok_returns_none(self):
self.assertIsNone(self._run(system="Linux", exists=True, access=True))
def test_linux_missing_device_dies(self):
with self.assertRaises(SystemExit):
self._run(system="Linux", exists=False, access=False)
def test_linux_no_access_dies(self):
with self.assertRaises(SystemExit):
self._run(system="Linux", exists=True, access=False)
def test_linux_missing_device_message(self):
import io
import sys
captured = io.StringIO()
with patch.object(sys, "stderr", captured):
with self.assertRaises(SystemExit):
self._run(system="Linux", exists=False, access=False)
msg = captured.getvalue()
self.assertIn("/dev/kvm", msg)
self.assertIn("kvm-intel", msg)
def test_linux_no_access_message(self):
import io
import sys
captured = io.StringIO()
with patch.object(sys, "stderr", captured):
with self.assertRaises(SystemExit):
self._run(system="Linux", exists=True, access=False)
msg = captured.getvalue()
self.assertIn("kvm", msg)
self.assertIn("group", msg)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()