feat(smolmachines): smolvm subprocess wrapper (PRD 0023 chunk 2b)

claude_bottle/backend/smolmachines/smolvm.py — one thin Python
function per smolvm CLI subcommand the launch flow needs:

  - pack_create(image, output)            → smolvm pack create
  - machine_create(name, from_path,
                   smolfile)               → smolvm machine create
  - machine_start(name)                   → smolvm machine start
  - machine_stop(name)                    → smolvm machine stop
  - machine_delete(name)                  → smolvm machine delete -f
  - machine_exec(name, argv, env,
                 workdir, timeout)         → smolvm machine exec
  - machine_cp(src, dst)                  → smolvm machine cp
  - is_available()                        → shutil.which check

The wrapper hides the CLI's inconsistent name-flag style
(positional NAME on create/delete, --name on start/stop/exec/
status) behind a uniform `name=` kwarg.

Two return shapes:
  - SmolvmRunResult (returncode + stdout + stderr) from
    machine_exec, because callers care about the in-VM
    command's exit code.
  - Raises SmolvmError on non-zero for all other commands;
    failure to create/start/stop a VM is fatal to the launch
    flow, not branched on.

Tests:
  - 15 unit cases mocking subprocess.run, covering argv shape
    per subcommand (the --name vs positional inconsistency
    locked down), SmolvmError on non-zero for non-exec paths,
    SmolvmRunResult passthrough on exec, empty-path cp no-op.
  - 2 integration cases against the real smolvm binary
    (gated on Darwin + smolvm on PATH + not GITEA_ACTIONS):
    smolvm --help responds, machine ls --json parses as a
    list (the contract chunk 4's list_active will consume).

531 unit tests passing. Real-smolvm smoke green locally.

Bundle bringup + launch wiring + the localhost-reach /
egress-port-bypass probes land in chunks 2c + 2d.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

claude_bottle/backend/smolmachines/smolvm.py

@@ -0,0 +1,189 @@
+"""Thin subprocess wrapper around the `smolvm` CLI (PRD 0023).
+
+One thin Python function per smolvm subcommand the launch flow
+needs. Two design choices worth flagging:
+
+  - **No daemon, no SDK.** smolvm 0.8.0 ships a `smolvm serve`
+    HTTP API as the long-term-clean integration target. The
+    project's stdlib-first ethos + the lower-overhead CLI calls
+    push v1 to shell out via `subprocess.run`. If a future
+    smolvm release makes `serve` mandatory (or significantly
+    faster), revisit.
+
+  - **Two return shapes.** `SmolvmRunResult` (returncode + stdout
+    + stderr captured) is returned by `machine_exec` because the
+    caller cares about the in-VM command's exit status, and by
+    test helpers that introspect output. The other calls
+    (`machine_start`, `machine_stop`, `pack_create`, etc.) raise
+    `SmolvmError` on non-zero exit — failure to start a VM is
+    fatal to the launch flow, not something callers want to
+    branch on.
+
+The wrapper is unit-tested with `subprocess.run` mocked; the
+integration smoke test (chunk 2d) exercises against a real
+smolvm binary."""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Mapping, Sequence
+
+
+_SMOLVM = "smolvm"
+
+
+@dataclass(frozen=True)
+class SmolvmRunResult:
+    """Captured result of an in-VM command. Mirrors the structure
+    `Bottle.exec` returns so callers can hand it straight through."""
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+class SmolvmError(RuntimeError):
+    """Raised when a smolvm subprocess returns non-zero on a path
+    where the caller has no useful branch to take (start failed,
+    pack failed, etc.). Carries the captured stderr for the
+    operator-facing log line."""
+
+    def __init__(self, argv: Sequence[str], result: subprocess.CompletedProcess):
+        self.argv = list(argv)
+        self.returncode = result.returncode
+        self.stdout = result.stdout
+        self.stderr = result.stderr
+        cmd = " ".join(self.argv)
+        super().__init__(
+            f"{cmd!r} failed (exit {result.returncode}): "
+            f"{(result.stderr or '').strip() or '<no stderr>'}"
+        )
+
+
+def _smolvm(*args: str, env: Mapping[str, str] | None = None,
+            check: bool = True) -> subprocess.CompletedProcess:
+    """One subprocess call into the smolvm CLI. `check=True`
+    raises SmolvmError on non-zero; `check=False` returns the
+    CompletedProcess for the caller to inspect."""
+    argv = [_SMOLVM, *args]
+    result = subprocess.run(
+        argv,
+        capture_output=True,
+        text=True,
+        env=dict(env) if env is not None else None,
+        check=False,
+    )
+    if check and result.returncode != 0:
+        raise SmolvmError(argv, result)
+    return result
+
+
+# --- Pack ----------------------------------------------------------------
+
+
+def pack_create(image: str, output: Path) -> None:
+    """`smolvm pack create --image <image> -o <output>`. Converts
+    an OCI image into a self-contained `.smolmachine` artifact
+    smolvm can boot via `machine create --from`. Idempotent on the
+    smolvm side — re-running with the same image+output rebuilds
+    from layer cache."""
+    _smolvm("pack", "create", "--image", image, "-o", str(output))
+
+
+# --- Machine lifecycle ---------------------------------------------------
+
+
+def machine_create(
+    name: str,
+    *,
+    from_path: Path | None = None,
+    smolfile: Path | None = None,
+) -> None:
+    """`smolvm machine create NAME [--from PATH] [--smolfile PATH]`.
+    NAME is positional (the CLI's exception to the `--name`
+    pattern other subcommands use)."""
+    args: list[str] = ["machine", "create"]
+    if from_path is not None:
+        args += ["--from", str(from_path)]
+    if smolfile is not None:
+        args += ["--smolfile", str(smolfile)]
+    args.append(name)
+    _smolvm(*args)
+
+
+def machine_start(name: str) -> None:
+    """`smolvm machine start --name NAME`."""
+    _smolvm("machine", "start", "--name", name)
+
+
+def machine_stop(name: str) -> None:
+    """`smolvm machine stop --name NAME`. Idempotent against
+    already-stopped machines: smolvm prints a notice and exits 0
+    in that case, so no special handling here."""
+    _smolvm("machine", "stop", "--name", name)
+
+
+def machine_delete(name: str) -> None:
+    """`smolvm machine delete -f NAME`. NAME is positional. `-f`
+    skips the interactive confirmation — required for
+    non-interactive teardown."""
+    _smolvm("machine", "delete", "-f", name)
+
+
+def machine_exec(
+    name: str,
+    argv: Sequence[str],
+    *,
+    env: Mapping[str, str] | None = None,
+    workdir: str | None = None,
+    timeout: str | None = None,
+) -> SmolvmRunResult:
+    """`smolvm machine exec --name NAME [-w DIR] [--timeout DUR]
+    [-e K=V ...] -- ARGV...`. Returns the captured result rather
+    than raising — callers (including `Bottle.exec`) care about
+    the in-VM command's exit code, not just whether smolvm ran.
+
+    `env` here is in-VM env vars (`-e K=V`), not the host
+    subprocess env — smolvm's own argv carries them through the
+    VMM."""
+    flags: list[str] = ["machine", "exec", "--name", name]
+    if workdir is not None:
+        flags += ["-w", workdir]
+    if timeout is not None:
+        flags += ["--timeout", timeout]
+    if env:
+        for k, v in env.items():
+            flags += ["-e", f"{k}={v}"]
+    # `--` separator before the command. smolvm's CLI requires it
+    # so its own flag parser doesn't grab argv items that look
+    # like flags.
+    flags.append("--")
+    flags += list(argv)
+    result = _smolvm(*flags, check=False)
+    return SmolvmRunResult(
+        returncode=result.returncode,
+        stdout=result.stdout or "",
+        stderr=result.stderr or "",
+    )
+
+
+def machine_cp(src: str, dst: str) -> None:
+    """`smolvm machine cp SRC DST`. Path syntax: `machine:path` to
+    reference a path inside the VM, bare path for the host. Both
+    SRC and DST are positional; either side can be machine: or
+    bare. Empty path is a no-op (returns immediately without
+    invoking smolvm)."""
+    if not src or not dst:
+        return
+    _smolvm("machine", "cp", src, dst)
+
+
+# --- Discovery -----------------------------------------------------------
+
+
+def is_available() -> bool:
+    """True iff `smolvm` is on PATH. Used by the integration test
+    suite's skip-guards."""
+    return shutil.which(_SMOLVM) is not None

tests/integration/test_smolmachines_smolvm_smoke.py

@@ -0,0 +1,64 @@
+"""Integration: PRD 0023 chunk 2b — smolvm subprocess wrapper
+exercised against the real binary.
+
+The full machine-lifecycle round trip (create → start → exec →
+delete) is gated behind macOS + Darwin platform check and lives
+in chunk 2d's smoke. This file just verifies `is_available()`
+correctly reports presence and `_smolvm()` can run a no-op
+subcommand without errors — enough to flag wrapper drift if
+smolvm's flag parser changes shape across versions."""
+
+from __future__ import annotations
+
+import os
+import platform
+import subprocess
+import unittest
+
+from claude_bottle.backend.smolmachines.smolvm import is_available
+
+
+@unittest.skipIf(
+    os.environ.get("GITEA_ACTIONS") == "true",
+    "skipped under act_runner: smolvm not installed on the runner",
+)
+@unittest.skipUnless(
+    platform.system() == "Darwin",
+    "smolvm is macOS-only for v1; Linux+KVM path is a future PRD",
+)
+@unittest.skipUnless(
+    is_available(),
+    "smolvm not on PATH; install via "
+    "curl -sSL https://smolmachines.com/install.sh | sh",
+)
+class TestSmolvmSmoke(unittest.TestCase):
+    def test_smolvm_help_responds(self):
+        # `smolvm --help` exits 0 (per `smolvm machine --help`
+        # convention) — verifies the binary launches and the
+        # top-level parser is intact.
+        r = subprocess.run(
+            ["smolvm", "--help"],
+            capture_output=True, text=True, check=False,
+        )
+        # Either exit-code 0 (clean) or 1 (some CLIs return 1
+        # from --help by convention; smolvm 0.8.0 does this). The
+        # point is the binary runs and emits help text.
+        self.assertIn("smolvm", r.stdout)
+        self.assertIn("machine", r.stdout)
+
+    def test_machine_ls_empty_returns_json_array(self):
+        # `machine ls --json` is the contract chunk 4's
+        # list_active wires to. Lock in that the JSON shape is
+        # parseable now so chunk 4 doesn't surprise us.
+        import json
+        r = subprocess.run(
+            ["smolvm", "machine", "ls", "--json"],
+            capture_output=True, text=True, check=False,
+        )
+        self.assertEqual(0, r.returncode, r.stderr)
+        parsed = json.loads(r.stdout)
+        self.assertIsInstance(parsed, list)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_smolvm.py

@@ -0,0 +1,212 @@
+"""Unit: smolvm subprocess wrapper (PRD 0023 chunk 2b).
+
+The wrapper is one thin function per smolvm CLI subcommand. Tests
+mock `subprocess.run` and assert on the constructed argv +
+SmolvmError raising on non-zero. The actual smolvm binary's
+behavior is exercised in chunk 2d's integration smoke test."""
+
+from __future__ import annotations
+
+import subprocess
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from claude_bottle.backend.smolmachines.smolvm import (
+    SmolvmError,
+    SmolvmRunResult,
+    is_available,
+    machine_cp,
+    machine_create,
+    machine_delete,
+    machine_exec,
+    machine_start,
+    machine_stop,
+    pack_create,
+)
+
+
+def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=0, stdout=stdout, stderr=stderr,
+    )
+
+
+def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=1, stdout="", stderr=stderr,
+    )
+
+
+class TestArgvShapes(unittest.TestCase):
+    """The CLI mixes `--name NAME` and positional-NAME styles
+    across subcommands. The wrapper hides that inconsistency
+    behind a uniform `name=` kwarg; lock down which form lands
+    in each argv."""
+
+    def _patch_run(self):
+        return patch(
+            "claude_bottle.backend.smolmachines.smolvm.subprocess.run",
+            return_value=_ok(),
+        )
+
+    def test_pack_create_argv(self):
+        with self._patch_run() as m:
+            pack_create("claude-bottle:latest", Path("/tmp/agent.smolmachine"))
+        argv = m.call_args.args[0]
+        self.assertEqual(
+            ["smolvm", "pack", "create",
+             "--image", "claude-bottle:latest",
+             "-o", "/tmp/agent.smolmachine"],
+            argv,
+        )
+
+    def test_machine_create_minimal(self):
+        with self._patch_run() as m:
+            machine_create("agent-xyz")
+        self.assertEqual(
+            ["smolvm", "machine", "create", "agent-xyz"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_create_with_from_and_smolfile(self):
+        with self._patch_run() as m:
+            machine_create(
+                "agent-xyz",
+                from_path=Path("/stage/agent.smolmachine"),
+                smolfile=Path("/stage/smolfile.toml"),
+            )
+        self.assertEqual(
+            ["smolvm", "machine", "create",
+             "--from", "/stage/agent.smolmachine",
+             "--smolfile", "/stage/smolfile.toml",
+             "agent-xyz"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_start_uses_dash_name(self):
+        # `start` is the --name flag form, NOT positional.
+        with self._patch_run() as m:
+            machine_start("agent-xyz")
+        self.assertEqual(
+            ["smolvm", "machine", "start", "--name", "agent-xyz"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_stop_uses_dash_name(self):
+        with self._patch_run() as m:
+            machine_stop("agent-xyz")
+        self.assertEqual(
+            ["smolvm", "machine", "stop", "--name", "agent-xyz"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_delete_uses_positional_name_and_force(self):
+        # delete NAME is positional; -f required so no interactive
+        # confirmation blocks teardown.
+        with self._patch_run() as m:
+            machine_delete("agent-xyz")
+        self.assertEqual(
+            ["smolvm", "machine", "delete", "-f", "agent-xyz"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_exec_argv_with_separator(self):
+        # `--` separator before the command — smolvm's flag parser
+        # would otherwise grab argv items that look like flags.
+        with self._patch_run() as m:
+            machine_exec("agent-xyz", ["echo", "hello"])
+        self.assertEqual(
+            ["smolvm", "machine", "exec", "--name", "agent-xyz",
+             "--", "echo", "hello"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_exec_env_workdir_timeout(self):
+        with self._patch_run() as m:
+            machine_exec(
+                "agent-xyz",
+                ["ls"],
+                env={"FOO": "bar", "BAZ": "qux"},
+                workdir="/app",
+                timeout="30s",
+            )
+        argv = m.call_args.args[0]
+        self.assertIn("-w", argv); self.assertIn("/app", argv)
+        self.assertIn("--timeout", argv); self.assertIn("30s", argv)
+        # Both env vars present as -e K=V pairs.
+        for pair in ("FOO=bar", "BAZ=qux"):
+            self.assertIn(pair, argv)
+
+    def test_machine_cp_positional_argv(self):
+        with self._patch_run() as m:
+            machine_cp("/host/file", "agent-xyz:/dest/file")
+        self.assertEqual(
+            ["smolvm", "machine", "cp",
+             "/host/file", "agent-xyz:/dest/file"],
+            m.call_args.args[0],
+        )
+
+    def test_machine_cp_empty_is_noop(self):
+        # Guard against an upstream caller passing an unset path —
+        # cp with an empty string is meaningless and would just
+        # confuse smolvm's error message.
+        with self._patch_run() as m:
+            machine_cp("", "agent-xyz:/dest")
+            machine_cp("/host", "")
+        self.assertEqual(0, m.call_count)
+
+
+class TestErrorPath(unittest.TestCase):
+    """`check=True` paths raise SmolvmError on non-zero; `exec` is
+    the one path that returns a result regardless."""
+
+    def test_create_failure_raises(self):
+        with patch(
+            "claude_bottle.backend.smolmachines.smolvm.subprocess.run",
+            return_value=_fail("no such image"),
+        ):
+            with self.assertRaises(SmolvmError) as cm:
+                machine_create("agent-xyz")
+            self.assertEqual(1, cm.exception.returncode)
+            self.assertIn("no such image", str(cm.exception))
+
+    def test_pack_create_failure_raises(self):
+        with patch(
+            "claude_bottle.backend.smolmachines.smolvm.subprocess.run",
+            return_value=_fail("pack failed"),
+        ):
+            with self.assertRaises(SmolvmError):
+                pack_create("missing:tag", Path("/tmp/out"))
+
+    def test_exec_failure_returns_result(self):
+        # The in-VM command's exit code is what Bottle.exec sees;
+        # `false` exiting non-zero is not a smolvm failure.
+        with patch(
+            "claude_bottle.backend.smolmachines.smolvm.subprocess.run",
+            return_value=subprocess.CompletedProcess(
+                args=[], returncode=42, stdout="", stderr="nope",
+            ),
+        ):
+            r = machine_exec("agent-xyz", ["sh", "-c", "exit 42"])
+        self.assertEqual(SmolvmRunResult(42, "", "nope"), r)
+
+
+class TestIsAvailable(unittest.TestCase):
+    def test_true_when_on_path(self):
+        with patch(
+            "claude_bottle.backend.smolmachines.smolvm.shutil.which",
+            return_value="/usr/local/bin/smolvm",
+        ):
+            self.assertTrue(is_available())
+
+    def test_false_when_missing(self):
+        with patch(
+            "claude_bottle.backend.smolmachines.smolvm.shutil.which",
+            return_value=None,
+        ):
+            self.assertFalse(is_available())
+
+
+if __name__ == "__main__":
+    unittest.main()