bot_bottle/backend/__init__.py

@@ -312,15 +312,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
     def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
         """Build/run the bottle and yield a handle; tear down on exit."""
 
-    def provision(self, plan: PlanT, target: str) -> str | None:
+    def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy host-side files (CA cert, prompt, skills, .git) into
         the running bottle. Called from `launch` after the container
-        / machine is up. `target` identifies the running instance in
-        backend-specific terms (Docker: resolved container name; fly:
-        machine id). Returns the in-container prompt path if a prompt
-        was provisioned, else None — the Bottle handle uses it to
-        decide whether to add provider-specific prompt args to the agent's
-        argv.
+        / machine is up. Returns the in-container prompt path if a
+        prompt was provisioned, else None — the Bottle handle uses it
+        to decide whether to add provider-specific prompt args to the
+        agent's argv.
 
         Default orchestration: ca → prompt → skills → workspace → git →
         supervise. CA install runs first so the agent's trust store
@@ -333,16 +331,16 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         on the agent's HTTP_PROXY path so every tool that respects
         HTTPS_PROXY (claude-code, git over HTTPS, npm, curl) is
         intercepted without per-tool reconfiguration."""
-        self.provision_ca(plan, target)
-        prompt_path = self.provision_prompt(plan, target)
-        self.provision_provider_auth(plan, target)
-        self.provision_skills(plan, target)
-        self.provision_workspace(plan, target)
-        self.provision_git(plan, target)
-        self.provision_supervise(plan, target)
+        self.provision_ca(plan, bottle)
+        prompt_path = self.provision_prompt(plan, bottle)
+        self.provision_provider_auth(plan, bottle)
+        self.provision_skills(plan, bottle)
+        self.provision_workspace(plan, bottle)
+        self.provision_git(plan, bottle)
+        self.provision_supervise(plan, bottle)
         return prompt_path
 
-    def provision_ca(self, plan: PlanT, target: str) -> None:
+    def provision_ca(self, plan: PlanT, bottle: "Bottle") -> None:
         """Install the per-bottle CA into the agent's trust store so
         the agent trusts the bumped CONNECT cert egress (was
         pipelock, pre-PRD-0017) presents. Default impl is a no-op so
@@ -351,34 +349,34 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         backend overrides to docker-cp the cert in and run
         `update-ca-certificates`."""
 
-    def provision_provider_auth(self, plan: PlanT, target: str) -> None:
+    def provision_provider_auth(self, plan: PlanT, bottle: "Bottle") -> None:
         """Install non-secret provider auth marker files into the agent
         home when a provider needs them to select the right auth mode.
         The default is no-op."""
 
     @abstractmethod
-    def provision_prompt(self, plan: PlanT, target: str) -> str | None:
+    def provision_prompt(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy the prompt file into the running bottle. Returns the
         in-container path iff the agent has a non-empty prompt;
         callers use the return value to decide whether to add
         provider-specific prompt args to the agent's argv."""
 
     @abstractmethod
-    def provision_skills(self, plan: PlanT, target: str) -> None:
+    def provision_skills(self, plan: PlanT, bottle: "Bottle") -> None:
         """Copy the agent's named skills from the host into the
         running bottle. No-op when the agent has no skills."""
 
-    def provision_workspace(self, plan: PlanT, target: str) -> None:
+    def provision_workspace(self, plan: PlanT, bottle: "Bottle") -> None:
         """Copy the operator workspace into the running bottle when
         the backend cannot bake it into the agent image. Default is
         no-op for backends like Docker that handle this before launch."""
 
     @abstractmethod
-    def provision_git(self, plan: PlanT, target: str) -> None:
+    def provision_git(self, plan: PlanT, bottle: "Bottle") -> None:
         """Copy the host's cwd `.git` directory into the running
         bottle if the user requested --cwd. No-op otherwise."""
 
-    def provision_supervise(self, plan: PlanT, target: str) -> None:
+    def provision_supervise(self, plan: PlanT, bottle: "Bottle") -> None:
         """Write the in-bottle Claude Code MCP config so the agent
         discovers the per-bottle supervise sidecar (PRD 0013).
         No-op when bottle.supervise is False or the backend doesn't

bot_bottle/backend/docker/backend.py

@@ -18,7 +18,7 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Generator, Sequence
 
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, Bottle, BottleBackend, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -57,23 +57,23 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
         with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 
-    def provision_ca(self, plan: DockerBottlePlan, target: str) -> None:
-        _ca.provision_ca(plan, target)
+    def provision_ca(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
+        _ca.provision_ca(plan, bottle)
 
-    def provision_prompt(self, plan: DockerBottlePlan, target: str) -> str | None:
-        return _prompt.provision_prompt(plan, target)
+    def provision_prompt(self, plan: DockerBottlePlan, bottle: Bottle) -> str | None:
+        return _prompt.provision_prompt(plan, bottle)
 
-    def provision_provider_auth(self, plan: DockerBottlePlan, target: str) -> None:
-        _provider_auth.provision_provider_auth(plan, target)
+    def provision_provider_auth(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
+        _provider_auth.provision_provider_auth(plan, bottle)
 
-    def provision_skills(self, plan: DockerBottlePlan, target: str) -> None:
-        _skills.provision_skills(plan, target)
+    def provision_skills(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
+        _skills.provision_skills(plan, bottle)
 
-    def provision_git(self, plan: DockerBottlePlan, target: str) -> None:
-        _git.provision_git(plan, target)
+    def provision_git(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
+        _git.provision_git(plan, bottle)
 
-    def provision_supervise(self, plan: DockerBottlePlan, target: str) -> None:
-        _supervise_prov.provision_supervise(plan, target)
+    def provision_supervise(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
+        _supervise_prov.provision_supervise(plan, bottle)
 
     def prepare_cleanup(self) -> DockerBottleCleanupPlan:
         return _cleanup.prepare_cleanup()

bot_bottle/backend/docker/launch.py

@@ -208,19 +208,21 @@ def launch(
             compose_dump_logs, project, compose_file, compose_log_path(state_dir),
         )
 
-        # Step 8: provision. Unchanged — uses `docker exec` against
-        # the agent container by its known name.
-        prompt_path = provision(plan, plan.container_name)
+        # Step 8: provision. Create the bottle first so provisioners
+        # can use bottle.exec / bottle.cp_in; set the prompt path
+        # returned by provision_prompt after the fact.
+        bottle = DockerBottle(
+            plan.container_name,
+            teardown,
+            None,
+            agent_command=plan.agent_command,
+            agent_prompt_mode=plan.agent_prompt_mode,
+        )
+        bottle._prompt_path = provision(plan, bottle)
 
         # Step 9: yield. exec_agent continues to use `docker exec -it`
         # — the agent runs `sleep infinity` per the renderer's
         # service spec.
-        yield DockerBottle(
-            plan.container_name,
-            teardown,
-            prompt_path,
-            agent_command=plan.agent_command,
-            agent_prompt_mode=plan.agent_prompt_mode,
-        )
+        yield bottle
     finally:
         teardown()

bot_bottle/backend/docker/provision/__init__.py

@@ -1,7 +1,7 @@
 """Per-provisioner modules for the Docker backend.
 
 Each module exports one top-level function:
-    provision_<thing>(plan: DockerBottlePlan, target: str) -> ...
+    provision_<thing>(plan: DockerBottlePlan, bottle: Bottle) -> ...
 
 `DockerBottleBackend.provision_*` methods delegate to these. The
 abstract `BottleBackend.provision_*` surface is unchanged; this

bot_bottle/backend/docker/provision/ca.py

@@ -31,33 +31,21 @@ stage dir; nothing in the agent ever sees it."""
 
 from __future__ import annotations
 
-import subprocess
-
+from ... import Bottle
 from ...util import AGENT_CA_PATH, log_ca_fingerprint, select_ca_cert
 from ..bottle_plan import DockerBottlePlan
 
 
-def provision_ca(plan: DockerBottlePlan, target: str) -> None:
+def provision_ca(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Copy the agent-facing CA cert into the agent, rebuild the
     trust bundle, emit a one-line fingerprint log. Called from
     `BottleBackend.provision` after the agent container is up."""
-    container = target
     cert_host_path, label = select_ca_cert(plan.egress_plan, plan.proxy_plan)
 
-    subprocess.run(
-        ["docker", "cp", str(cert_host_path), f"{container}:{AGENT_CA_PATH}"],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, "chmod", "644", AGENT_CA_PATH],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, "update-ca-certificates"],
-        stdout=subprocess.DEVNULL,
-        check=True,
+    bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
+    bottle.exec(
+        f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates",
+        user="root",
     )
 
     log_ca_fingerprint(cert_host_path, label)

bot_bottle/backend/docker/provision/git.py

@@ -19,74 +19,63 @@ Three concerns, all about git in the agent:
 from __future__ import annotations
 
 import os
-import subprocess
+import shlex
 
 from ....git_gate import GIT_GATE_HOSTNAME, git_gate_render_gitconfig
 from ....log import info
-from .. import util as docker_mod
+from ... import Bottle
 from ..bottle_plan import DockerBottlePlan
 
 
-def provision_git(plan: DockerBottlePlan, target: str) -> None:
+def provision_git(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Set up git inside the bottle. Runs all three subcases; each
     no-ops when its condition isn't met."""
-    _provision_cwd_git(plan, target)
-    _provision_git_gate_config(plan, target)
-    _provision_git_user(plan, target)
+    _provision_cwd_git(plan, bottle)
+    _provision_git_gate_config(plan, bottle)
+    _provision_git_user(plan, bottle)
 
 
-def _provision_cwd_git(plan: DockerBottlePlan, target: str) -> None:
+def _provision_cwd_git(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """If --cwd was set and the host cwd has a .git directory, copy
     it into /home/node/workspace/.git and fix ownership. No-op
     otherwise."""
     workspace = plan.workspace_plan
     if not (workspace.enabled and workspace.copy_git and workspace.has_host_git_dir):
         return
-    container = target
     guest_workspace_git = f"{workspace.guest_path}/.git"
     host_git = str(workspace.host_path / ".git")
-    info(f"copying {host_git} -> {container}:{guest_workspace_git}")
-    subprocess.run(
-        ["docker", "cp", host_git, f"{container}:{guest_workspace_git}"],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-    subprocess.run(
-        [
-            "docker", "exec", "-u", "0", container,
-            "chown", "-R", workspace.owner, guest_workspace_git,
-        ],
-        stdout=subprocess.DEVNULL,
-        check=True,
+    info(f"copying {host_git} -> {bottle.name}:{guest_workspace_git}")
+    bottle.cp_in(host_git, guest_workspace_git)
+    bottle.exec(
+        f"chown -R {shlex.quote(workspace.owner)} {shlex.quote(guest_workspace_git)}",
+        user="root",
     )
 
 
-def _provision_git_gate_config(plan: DockerBottlePlan, target: str) -> None:
+def _provision_git_gate_config(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Write ~/.gitconfig in the bottle with the git-gate
     insteadOf rules. No-op when the bottle has no `git` entries."""
-    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
-    if not bottle.git:
+    manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    if not manifest_bottle.git:
         return
-    container = target
     container_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
     container_gitconfig = f"{container_home}/.gitconfig"
 
-    content = git_gate_render_gitconfig(bottle.git, GIT_GATE_HOSTNAME)
+    content = git_gate_render_gitconfig(manifest_bottle.git, GIT_GATE_HOSTNAME)
     config_file = plan.stage_dir / "agent_gitconfig"
     config_file.write_text(content)
     config_file.chmod(0o600)
 
-    info(f"writing {container_gitconfig} with {len(bottle.git)} insteadOf rule(s)")
-    subprocess.run(
-        ["docker", "cp", str(config_file), f"{container}:{container_gitconfig}"],
-        stdout=subprocess.DEVNULL,
-        check=True,
+    info(f"writing {container_gitconfig} with {len(manifest_bottle.git)} insteadOf rule(s)")
+    bottle.cp_in(str(config_file), container_gitconfig)
+    bottle.exec(
+        f"chown node:node {shlex.quote(container_gitconfig)} && "
+        f"chmod 644 {shlex.quote(container_gitconfig)}",
+        user="root",
     )
-    docker_mod.docker_exec_root(container, ["chown", "node:node", container_gitconfig])
-    docker_mod.docker_exec_root(container, ["chmod", "644", container_gitconfig])
 
 
-def _provision_git_user(plan: DockerBottlePlan, target: str) -> None:
+def _provision_git_user(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Apply `git config --global user.{name,email}` inside the
     bottle so the agent's commits are attributed to the operator-
     chosen identity instead of the agent image's default
@@ -101,23 +90,19 @@ def _provision_git_user(plan: DockerBottlePlan, target: str) -> None:
     Each field set independently — name-only or email-only
     configs only run the `git config` line for the field
     present."""
-    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
-    gu = bottle.git_user
+    manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    gu = manifest_bottle.git_user
     if gu.is_empty():
         return
     if gu.name:
         info(f"git config --global user.name = {gu.name!r}")
-        subprocess.run(
-            ["docker", "exec", "-u", "node", target,
-             "git", "config", "--global", "user.name", gu.name],
-            stdout=subprocess.DEVNULL,
-            check=True,
+        bottle.exec(
+            f"git config --global user.name {shlex.quote(gu.name)}",
+            user="node",
         )
     if gu.email:
         info(f"git config --global user.email = {gu.email!r}")
-        subprocess.run(
-            ["docker", "exec", "-u", "node", target,
-             "git", "config", "--global", "user.email", gu.email],
-            stdout=subprocess.DEVNULL,
-            check=True,
+        bottle.exec(
+            f"git config --global user.email {shlex.quote(gu.email)}",
+            user="node",
         )

bot_bottle/backend/docker/provision/prompt.py

@@ -7,36 +7,26 @@ actually has a prompt — the return value signals which case."""
 from __future__ import annotations
 
 import os
-import subprocess
 
+from ... import Bottle
 from ..bottle_plan import DockerBottlePlan
 
 
-def provision_prompt(plan: DockerBottlePlan, target: str) -> str | None:
+def provision_prompt(plan: DockerBottlePlan, bottle: Bottle) -> str | None:
     """Copy the prompt file into the container, fix ownership/mode.
     Returns the in-container path if the agent has a non-empty
     prompt (drives --append-system-prompt-file), else None. The
     file is copied either way so the path always exists."""
-    container = target
     container_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
     in_container_prompt_path = f"{container_home}/.bot-bottle-prompt.txt"
 
-    subprocess.run(
-        ["docker", "cp", str(plan.prompt_file), f"{container}:{in_container_prompt_path}"],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
+    bottle.cp_in(str(plan.prompt_file), in_container_prompt_path)
     # `docker cp` preserves host UID; re-own/mode as root so node
     # can read its own mode-600 prompt regardless of host UID.
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, "chown", "node:node", in_container_prompt_path],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, "chmod", "600", in_container_prompt_path],
-        stdout=subprocess.DEVNULL,
-        check=True,
+    bottle.exec(
+        f"chown node:node {in_container_prompt_path} && "
+        f"chmod 600 {in_container_prompt_path}",
+        user="root",
     )
 
     agent = plan.spec.manifest.agents[plan.spec.agent_name]

bot_bottle/backend/docker/provision/provider_auth.py

@@ -2,35 +2,34 @@
 
 from __future__ import annotations
 
-import subprocess
+import shlex
 
+from ....log import die
+from ... import Bottle
 from ..bottle_plan import DockerBottlePlan
 
 
-def provision_provider_auth(plan: DockerBottlePlan, target: str) -> None:
-    """Apply provider-owned guest setup through Docker primitives."""
+def provision_provider_auth(plan: DockerBottlePlan, bottle: Bottle) -> None:
+    """Apply provider-owned guest setup through the bottle's exec / cp_in."""
     provision = plan.agent_provision
     for d in provision.dirs:
-        _exec(target, ["mkdir", "-p", d.guest_path])
-        _exec(target, ["chown", d.owner, d.guest_path])
-        _exec(target, ["chmod", d.mode, d.guest_path])
+        _exec(bottle, f"mkdir -p {shlex.quote(d.guest_path)}", d.guest_path)
+        _exec(bottle, f"chown {shlex.quote(d.owner)} {shlex.quote(d.guest_path)}", d.guest_path)
+        _exec(bottle, f"chmod {shlex.quote(d.mode)} {shlex.quote(d.guest_path)}", d.guest_path)
     for command in provision.pre_copy:
-        _exec(target, list(command.argv))
+        _exec(bottle, shlex.join(command.argv), command.error)
     for f in provision.files:
-        subprocess.run(
-            ["docker", "cp", str(f.host_path), f"{target}:{f.guest_path}"],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        _exec(target, ["chown", f.owner, f.guest_path])
-        _exec(target, ["chmod", f.mode, f.guest_path])
+        bottle.cp_in(str(f.host_path), f.guest_path)
+        _exec(bottle, f"chown {shlex.quote(f.owner)} {shlex.quote(f.guest_path)}", f.guest_path)
+        _exec(bottle, f"chmod {shlex.quote(f.mode)} {shlex.quote(f.guest_path)}", f.guest_path)
     for command in provision.verify:
-        _exec(target, list(command.argv))
+        _exec(bottle, shlex.join(command.argv), command.error)
 
 
-def _exec(target: str, argv: list[str]) -> None:
-    subprocess.run(
-        ["docker", "exec", "-u", "0", target, *argv],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
+def _exec(bottle: Bottle, script: str, error: str) -> None:
+    result = bottle.exec(script, user="root")
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        if detail:
+            detail = f": {detail}"
+        die(f"agent provider provisioning: {error}{detail}")

bot_bottle/backend/docker/provision/skills.py

@@ -9,54 +9,36 @@ a partial container."""
 from __future__ import annotations
 
 import os
-import subprocess
 
 from ....log import die, info
 from ...util import host_skill_dir
+from ... import Bottle
 from ..bottle_plan import DockerBottlePlan
 
 
-def provision_skills(plan: DockerBottlePlan, target: str) -> None:
+def provision_skills(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Copy each of the agent's named skills from the host's
     ~/.claude/skills/<name>/ into the container's equivalent path.
     For each skill: ensure parent dir, wipe any prior copy, then
-    `docker cp <host>/. <container>:<dst>/` so the contents are
+    `cp_in <host>/. <container>:<dst>/` so the contents are
     copied into a freshly-created destination dir. No-op when the
     agent has no skills."""
     agent = plan.spec.manifest.agents[plan.spec.agent_name]
     if not agent.skills:
         return
 
-    container = target
     container_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
     skills_dir = os.environ.get(
         "BOT_BOTTLE_CONTAINER_SKILLS_DIR", f"{container_home}/.claude/skills"
     )
 
-    subprocess.run(
-        ["docker", "exec", container, "mkdir", "-p", skills_dir],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
+    bottle.exec(f"mkdir -p {skills_dir}", user="node")
 
     for n in agent.skills:
         src = host_skill_dir(n)
         if not os.path.isdir(src):
             die(f"skill '{n}' disappeared from host between validation and copy at {src}.")
         dst = f"{skills_dir}/{n}"
-        info(f"copying skill {n} into {container}:{dst}")
-        subprocess.run(
-            ["docker", "exec", container, "rm", "-rf", dst],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        subprocess.run(
-            ["docker", "exec", container, "mkdir", "-p", dst],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        subprocess.run(
-            ["docker", "cp", f"{src}/.", f"{container}:{dst}/"],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
+        info(f"copying skill {n} into {bottle.name}:{dst}")
+        bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="node")
+        bottle.cp_in(f"{src}/.", f"{dst}/")

bot_bottle/backend/docker/provision/supervise.py

@@ -18,10 +18,9 @@ sidecar that isn't running.
 
 from __future__ import annotations
 
-import subprocess
-
 from ....log import info, warn
 from ....supervise import SUPERVISE_HOSTNAME, SUPERVISE_PORT
+from ... import Bottle
 from ..bottle_plan import DockerBottlePlan
 
 
@@ -32,7 +31,7 @@ def supervise_mcp_url() -> str:
     return f"http://{SUPERVISE_HOSTNAME}:{SUPERVISE_PORT}/"
 
 
-def provision_supervise(plan: DockerBottlePlan, target: str) -> None:
+def provision_supervise(plan: DockerBottlePlan, bottle: Bottle) -> None:
     """Run `claude mcp add` inside the agent container to register
     the supervise sidecar in claude-code's user config. No-op when
     bottle.supervise is False.
@@ -43,16 +42,11 @@ def provision_supervise(plan: DockerBottlePlan, target: str) -> None:
     if plan.supervise_plan is None:
         return
     url = supervise_mcp_url()
-    argv = [
-        "docker", "exec", "-u", "node", target,
-        "claude", "mcp", "add",
-        "--scope", "user",
-        "--transport", "http",
-        _SUPERVISE_MCP_NAME,
-        url,
-    ]
     info(f"registering supervise MCP server in agent claude config → {url}")
-    r = subprocess.run(argv, capture_output=True, text=True, check=False)
+    r = bottle.exec(
+        f"claude mcp add --scope user --transport http {_SUPERVISE_MCP_NAME} {url}",
+        user="node",
+    )
     if r.returncode != 0:
         warn(
             f"`claude mcp add supervise` failed (exit {r.returncode}): "

bot_bottle/backend/smolmachines/backend.py

@@ -7,7 +7,7 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Generator, Sequence
 
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, Bottle, BottleBackend, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -54,39 +54,39 @@ class SmolmachinesBottleBackend(
             yield bottle
 
     def provision_ca(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _ca.provision_ca(plan, target)
+        _ca.provision_ca(plan, bottle)
 
     def provision_prompt(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> str | None:
-        return _prompt.provision_prompt(plan, target)
+        return _prompt.provision_prompt(plan, bottle)
 
     def provision_provider_auth(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _provider_auth.provision_provider_auth(plan, target)
+        _provider_auth.provision_provider_auth(plan, bottle)
 
     def provision_skills(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _skills.provision_skills(plan, target)
+        _skills.provision_skills(plan, bottle)
 
     def provision_workspace(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _workspace.provision_workspace(plan, target)
+        _workspace.provision_workspace(plan, bottle)
 
     def provision_git(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _git.provision_git(plan, target)
+        _git.provision_git(plan, bottle)
 
     def provision_supervise(
-        self, plan: SmolmachinesBottlePlan, target: str
+        self, plan: SmolmachinesBottlePlan, bottle: Bottle
     ) -> None:
-        _supervise.provision_supervise(plan, target)
+        _supervise.provision_supervise(plan, bottle)
 
     def prepare_cleanup(self) -> SmolmachinesBottleCleanupPlan:
         return _cleanup.prepare_cleanup()

bot_bottle/backend/smolmachines/launch.py

@@ -113,15 +113,16 @@ def launch(
         _launch_vm(plan, agent_from_path, loopback_ip, stack)
         _init_vm(plan)
 
-        prompt_path = provision(plan, plan.machine_name)
-
-        yield SmolmachinesBottle(
+        bottle = SmolmachinesBottle(
             plan.machine_name,
-            prompt_path=prompt_path,
+            prompt_path=None,
             guest_env=plan.guest_env,
             agent_command=plan.agent_command,
             agent_prompt_mode=plan.agent_prompt_mode,
         )
+        bottle._prompt_path = provision(plan, bottle)
+
+        yield bottle
     finally:
         _teardown_smolmachines(stack, plan)
 

bot_bottle/backend/smolmachines/provision/ca.py

@@ -2,8 +2,8 @@
 trust store (PRD 0023 chunk 4d).
 
 Mirrors `backend.docker.provision.ca`: select the right CA (egress
-when the bottle has routes, else pipelock), `smolvm machine cp` it
-to Debian's `/usr/local/share/ca-certificates/` path,
+when the bottle has routes, else pipelock), copy it to Debian's
+`/usr/local/share/ca-certificates/` path,
 `update-ca-certificates` to rebuild the trust bundle, and log the
 fingerprint once. The selected cert depends on the agent's
 HTTP_PROXY target — same logic as the docker backend, since the
@@ -24,20 +24,20 @@ from ...util import (
     log_ca_fingerprint,
     select_ca_cert,
 )
-from .. import smolvm as _smolvm
+from ... import Bottle, ExecResult
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
 _SIGKILL_EXIT = 128 + 9
 
 
-def provision_ca(plan: SmolmachinesBottlePlan, target: str) -> None:
+def provision_ca(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """Copy the agent-facing CA cert into the guest, rebuild the
     trust bundle, emit a one-line fingerprint log. Called from
     `BottleBackend.provision` after the smolvm guest is up."""
     cert_host_path, label = select_ca_cert(plan.egress_plan, plan.proxy_plan)
 
-    _smolvm.machine_cp(str(cert_host_path), f"{target}:{AGENT_CA_PATH}")
+    bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
     # Mode 0644 — readable to non-root tools in the guest.
     # update-ca-certificates rebuilds the bundle at AGENT_CA_BUNDLE,
     # which is what curl / Python ssl / OpenSSL-based tools read by
@@ -45,21 +45,21 @@ def provision_ca(plan: SmolmachinesBottlePlan, target: str) -> None:
     # REQUESTS_CA_BUNDLE) on the guest_env covers Node + Python
     # `requests` / libraries that don't load the system bundle.
     #
-    r = _install_ca(target)
+    r = _install_ca(bottle)
     if r.returncode == _SIGKILL_EXIT:
         # smolvm/libkrun can SIGKILL an otherwise-normal exec
         # during early-VM provisioning. `update-ca-certificates`
         # is idempotent, so retry the same install once after a
         # short settle delay before treating it as fatal.
         time.sleep(1.0)
-        r = _install_ca(target)
+        r = _install_ca(bottle)
 
     if r.returncode != 0:
         # update-ca-certificates not adding our cert is fatal —
         # claude-code's TLS handshake against the egress-MITM'd
         # api.anthropic.com would fail downstream. Bail early
-        # with what we can see (output is captured by smolvm so
-        # we can surface it).
+        # with what we can see (output is captured so we can
+        # surface it).
         die(
             f"update-ca-certificates didn't add the agent CA "
             f"(exit {r.returncode}): "
@@ -70,21 +70,21 @@ def provision_ca(plan: SmolmachinesBottlePlan, target: str) -> None:
     log_ca_fingerprint(cert_host_path, label)
 
 
-def _install_ca(target: str) -> _smolvm.SmolvmRunResult:
+def _install_ca(bottle: Bottle) -> ExecResult:
     # chown + chmod + update-ca-certificates + bundle
-    # verification run in one `sh -c` so we only pay one
-    # machine_exec round trip; the `&&` chaining surfaces the
-    # first failure as the return code. The verify check is more
-    # stable than requiring "1 added" in stdout: a retry after a
+    # verification run in one exec so we only pay one
+    # round trip; the `&&` chaining surfaces the first failure
+    # as the return code. The verify check is more stable than
+    # requiring "1 added" in stdout: a retry after a
     # partially-completed first run may legitimately report "0
     # added" while the cert is already installed.
-    return _smolvm.machine_exec(target, [
-        "sh", "-c",
+    return bottle.exec(
         f"chown root:root {AGENT_CA_PATH} && "
         f"chmod 644 {AGENT_CA_PATH} && "
         f"update-ca-certificates && "
         f"openssl verify -CAfile {AGENT_CA_BUNDLE} {AGENT_CA_PATH}",
-    ])
+        user="root",
+    )
 
 
 # Re-exported for the launch/provision_ca caller + tests. The path

bot_bottle/backend/smolmachines/provision/git.py

@@ -26,12 +26,13 @@ git_gate module."""
 from __future__ import annotations
 
 import os
+import shlex
 import tempfile
 from pathlib import Path
 
 from ....git_gate import git_gate_render_gitconfig
 from ....log import info
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
@@ -46,15 +47,15 @@ def _guest_home() -> str:
     return os.environ.get("BOT_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
 
 
-def provision_git(plan: SmolmachinesBottlePlan, target: str) -> None:
+def provision_git(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """Set up git inside the guest. Runs all three subcases; each
     no-ops when its condition isn't met."""
-    _provision_cwd_git(plan, target)
-    _provision_git_gate_config(plan, target)
-    _provision_git_user(plan, target)
+    _provision_cwd_git(plan, bottle)
+    _provision_git_gate_config(plan, bottle)
+    _provision_git_user(plan, bottle)
 
 
-def _provision_cwd_git(plan: SmolmachinesBottlePlan, target: str) -> None:
+def _provision_cwd_git(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """If --cwd was set and the host cwd has a .git directory, copy
     it into <guest_home>/workspace/.git and fix ownership. No-op
     otherwise."""
@@ -63,25 +64,26 @@ def _provision_cwd_git(plan: SmolmachinesBottlePlan, target: str) -> None:
         return
     guest_workspace_git = f"{workspace.guest_path}/.git"
     host_git = str(workspace.host_path / ".git")
-    info(f"copying {host_git} -> {target}:{guest_workspace_git}")
-    # mkdir -p the workspace dir so `machine cp` lands the .git
+    info(f"copying {host_git} -> {bottle.name}:{guest_workspace_git}")
+    # mkdir -p the workspace dir so cp_in lands the .git
     # directly there even on first-time bottles.
-    _smolvm.machine_exec(target, ["mkdir", "-p", workspace.guest_path])
-    _smolvm.machine_cp(
-        host_git, f"{target}:{guest_workspace_git}",
-    )
-    # `machine cp` lands files as root; the agent runs as node so
+    bottle.exec(f"mkdir -p {shlex.quote(workspace.guest_path)}", user="root")
+    bottle.cp_in(host_git, guest_workspace_git)
+    # cp_in lands files as root; the agent runs as node so
     # the workspace tree must be chowned over.
-    _smolvm.machine_exec(
-        target, ["chown", "-R", workspace.owner, guest_workspace_git],
+    bottle.exec(
+        f"chown -R {shlex.quote(workspace.owner)} {shlex.quote(guest_workspace_git)}",
+        user="root",
     )
 
 
-def _provision_git_gate_config(plan: SmolmachinesBottlePlan, target: str) -> None:
+def _provision_git_gate_config(
+    plan: SmolmachinesBottlePlan, bottle: Bottle
+) -> None:
     """Write ~/.gitconfig in the guest with the git-gate insteadOf
     rules. No-op when the bottle has no `git` entries."""
-    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
-    if not bottle.git:
+    manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    if not manifest_bottle.git:
         return
 
     # `<loopback alias>:<host port>` form: the bundle's git-gate
@@ -90,11 +92,11 @@ def _provision_git_gate_config(plan: SmolmachinesBottlePlan, target: str) -> Non
     # TSI, not the docker bridge IP) can dial it. launch.py
     # populates `plan.agent_git_gate_host` after bundle bringup.
     content = git_gate_render_gitconfig(
-        bottle.git, plan.agent_git_gate_host, scheme="http",
+        manifest_bottle.git, plan.agent_git_gate_host, scheme="http",
     )
 
     guest_gitconfig = f"{_guest_home()}/.gitconfig"
-    # Stage the file under the plan's stage_dir so `machine cp`
+    # Stage the file under the plan's stage_dir so cp_in
     # has a stable host path. The plan's stage_dir is cleaned up
     # by start.py's session-end teardown.
     with tempfile.NamedTemporaryFile(
@@ -105,41 +107,38 @@ def _provision_git_gate_config(plan: SmolmachinesBottlePlan, target: str) -> Non
         config_file = Path(f.name)
     os.chmod(config_file, 0o600)
 
-    info(f"writing {guest_gitconfig} with {len(bottle.git)} insteadOf rule(s)")
-    _smolvm.machine_cp(str(config_file), f"{target}:{guest_gitconfig}")
-    _smolvm.machine_exec(target, ["chown", "node:node", guest_gitconfig])
-    _smolvm.machine_exec(target, ["chmod", "644", guest_gitconfig])
+    info(f"writing {guest_gitconfig} with {len(manifest_bottle.git)} insteadOf rule(s)")
+    bottle.cp_in(str(config_file), guest_gitconfig)
+    bottle.exec(
+        f"chown node:node {shlex.quote(guest_gitconfig)} && "
+        f"chmod 644 {shlex.quote(guest_gitconfig)}",
+        user="root",
+    )
 
 
 def _provision_git_user(
-    plan: SmolmachinesBottlePlan, target: str,
+    plan: SmolmachinesBottlePlan, bottle: Bottle,
 ) -> None:
     """Apply `git config --global user.{name,email}` inside the
     guest as the node user so --global lands in the same
     `/home/node/.gitconfig` that `_provision_git_gate_config`
     writes to. No-op when the bottle didn't declare `git.user`.
 
-    Runs via `runuser -u node --`; HOME is forced via smolvm's
-    `-e` flag because runuser (without -l) inherits root's
-    HOME=/root, which would put --global in the wrong file."""
-    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
-    gu = bottle.git_user
+    SmolmachinesBottle.exec(user="node") automatically sets
+    HOME=/home/node so --global writes to /home/node/.gitconfig."""
+    manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    gu = manifest_bottle.git_user
     if gu.is_empty():
         return
-    env = {"HOME": _guest_home(), "USER": "node"}
     if gu.name:
         info(f"git config --global user.name = {gu.name!r}")
-        _smolvm.machine_exec(
-            target,
-            ["runuser", "-u", "node", "--",
-             "git", "config", "--global", "user.name", gu.name],
-            env=env,
+        bottle.exec(
+            f"git config --global user.name {shlex.quote(gu.name)}",
+            user="node",
         )
     if gu.email:
         info(f"git config --global user.email = {gu.email!r}")
-        _smolvm.machine_exec(
-            target,
-            ["runuser", "-u", "node", "--",
-             "git", "config", "--global", "user.email", gu.email],
-            env=env,
+        bottle.exec(
+            f"git config --global user.email {shlex.quote(gu.email)}",
+            user="node",
         )

bot_bottle/backend/smolmachines/provision/prompt.py

@@ -5,7 +5,7 @@ exists) but `--append-system-prompt-file` only fires when the
 agent actually has a prompt — the return value signals which
 case, mirroring the docker backend's contract.
 
-`smolvm machine cp` lands files as root inside the VM; the claude
+cp_in lands files as root inside the VM; the claude
 process runs as `node`, so we chown + chmod the prompt after the
 copy. Same flow as the docker backend's provision_prompt."""
 
@@ -13,7 +13,7 @@ from __future__ import annotations
 
 import os
 
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
@@ -23,7 +23,7 @@ from ..bottle_plan import SmolmachinesBottlePlan
 _DEFAULT_GUEST_HOME = "/home/node"
 
 
-def provision_prompt(plan: SmolmachinesBottlePlan, target: str) -> str | None:
+def provision_prompt(plan: SmolmachinesBottlePlan, bottle: Bottle) -> str | None:
     """Copy the prompt file into the running smolvm guest, fix
     ownership/mode. Returns the in-guest path if the agent has a
     non-empty prompt (drives --append-system-prompt-file), else
@@ -32,11 +32,13 @@ def provision_prompt(plan: SmolmachinesBottlePlan, target: str) -> str | None:
     guest_home = os.environ.get("BOT_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
     in_guest_prompt_path = f"{guest_home}/.bot-bottle-prompt.txt"
 
-    _smolvm.machine_cp(str(plan.prompt_file), f"{target}:{in_guest_prompt_path}")
-    # machine cp lands as root, source's 0o600 mode is preserved —
+    bottle.cp_in(str(plan.prompt_file), in_guest_prompt_path)
+    # cp_in lands as root, source's 0o600 mode is preserved —
     # node can't read its own prompt without these two.
-    _smolvm.machine_exec(target, ["chown", "node:node", in_guest_prompt_path])
-    _smolvm.machine_exec(target, ["chmod", "600", in_guest_prompt_path])
+    bottle.exec(
+        f"chown node:node {in_guest_prompt_path} && chmod 600 {in_guest_prompt_path}",
+        user="root",
+    )
 
     agent = plan.spec.manifest.agents[plan.spec.agent_name]
     return in_guest_prompt_path if agent.prompt else None

bot_bottle/backend/smolmachines/provision/provider_auth.py

@@ -2,30 +2,32 @@
 
 from __future__ import annotations
 
+import shlex
+
 from ....log import die
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
-def provision_provider_auth(plan: SmolmachinesBottlePlan, target: str) -> None:
-    """Apply provider-owned guest setup through smolvm primitives."""
+def provision_provider_auth(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
+    """Apply provider-owned guest setup through the bottle's exec / cp_in."""
     provision = plan.agent_provision
     for d in provision.dirs:
-        _exec(target, ["mkdir", "-p", d.guest_path], f"could not create {d.guest_path}")
-        _exec(target, ["chown", d.owner, d.guest_path], f"could not chown {d.guest_path}")
-        _exec(target, ["chmod", d.mode, d.guest_path], f"could not chmod {d.guest_path}")
+        _exec(bottle, f"mkdir -p {shlex.quote(d.guest_path)}", f"could not create {d.guest_path}")
+        _exec(bottle, f"chown {shlex.quote(d.owner)} {shlex.quote(d.guest_path)}", f"could not chown {d.guest_path}")
+        _exec(bottle, f"chmod {shlex.quote(d.mode)} {shlex.quote(d.guest_path)}", f"could not chmod {d.guest_path}")
     for command in provision.pre_copy:
-        _exec(target, list(command.argv), command.error)
+        _exec(bottle, shlex.join(command.argv), command.error)
     for f in provision.files:
-        _smolvm.machine_cp(str(f.host_path), f"{target}:{f.guest_path}")
-        _exec(target, ["chown", f.owner, f.guest_path], f"could not chown {f.guest_path}")
-        _exec(target, ["chmod", f.mode, f.guest_path], f"could not chmod {f.guest_path}")
+        bottle.cp_in(str(f.host_path), f.guest_path)
+        _exec(bottle, f"chown {shlex.quote(f.owner)} {shlex.quote(f.guest_path)}", f"could not chown {f.guest_path}")
+        _exec(bottle, f"chmod {shlex.quote(f.mode)} {shlex.quote(f.guest_path)}", f"could not chmod {f.guest_path}")
     for command in provision.verify:
-        _exec(target, list(command.argv), command.error)
+        _exec(bottle, shlex.join(command.argv), command.error)
 
 
-def _exec(target: str, argv: list[str], error: str) -> None:
-    result = _smolvm.machine_exec(target, argv)
+def _exec(bottle: Bottle, script: str, error: str) -> None:
+    result = bottle.exec(script, user="root")
     if result.returncode != 0:
         detail = (result.stderr or result.stdout).strip()
         if detail:

bot_bottle/backend/smolmachines/provision/skills.py

@@ -13,7 +13,7 @@ import os
 
 from ....log import die, info
 from ...util import host_skill_dir
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
@@ -24,18 +24,18 @@ from ..bottle_plan import SmolmachinesBottlePlan
 _DEFAULT_SKILLS_DIR = "/home/node/.claude/skills"
 
 
-def provision_skills(plan: SmolmachinesBottlePlan, target: str) -> None:
+def provision_skills(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """Copy each of the agent's named skills from the host's
     ~/.claude/skills/<name>/ into the guest's equivalent path.
-    For each skill: `mkdir -p` the destination, `smolvm machine cp`
-    the host source dir over, then chown the result to node:node so
-    the agent can read it. No-op when the agent has no skills.
+    For each skill: `mkdir -p` the destination, cp_in the host
+    source dir over, then chown the result to node:node so the
+    agent can read it. No-op when the agent has no skills.
 
-    smolvm machine cp on a directory copies recursively (same
-    semantics as `cp -r`); unlike docker cp's trailing-slash
-    convention, smolvm doesn't need the `/.` suffix dance.
+    cp_in on a directory copies recursively; unlike docker cp's
+    trailing-slash convention, smolvm doesn't need the `/.` suffix
+    dance.
 
-    machine cp lands files as root inside the VM, so we chown each
+    cp_in lands files as root inside the VM, so we chown each
     skill tree over to node:node after the copy — same pattern as
     the docker backend's provision_prompt."""
     agent = plan.spec.manifest.agents[plan.spec.agent_name]
@@ -46,7 +46,7 @@ def provision_skills(plan: SmolmachinesBottlePlan, target: str) -> None:
         "BOT_BOTTLE_GUEST_SKILLS_DIR", _DEFAULT_SKILLS_DIR,
     )
 
-    _smolvm.machine_exec(target, ["mkdir", "-p", skills_dir])
+    bottle.exec(f"mkdir -p {skills_dir}", user="root")
 
     for name in agent.skills:
         src = host_skill_dir(name)
@@ -56,8 +56,8 @@ def provision_skills(plan: SmolmachinesBottlePlan, target: str) -> None:
                 f"validation and copy at {src}."
             )
         dst = f"{skills_dir}/{name}"
-        info(f"copying skill {name} into {target}:{dst}")
+        info(f"copying skill {name} into {bottle.name}:{dst}")
         # Wipe any prior copy so re-runs don't accumulate.
-        _smolvm.machine_exec(target, ["rm", "-rf", dst])
-        _smolvm.machine_cp(src, f"{target}:{dst}")
-        _smolvm.machine_exec(target, ["chown", "-R", "node:node", dst])
+        bottle.exec(f"rm -rf {dst}", user="root")
+        bottle.cp_in(src, dst)
+        bottle.exec(f"chown -R node:node {dst}", user="root")

bot_bottle/backend/smolmachines/provision/supervise.py

@@ -7,21 +7,21 @@ stuck-recovery MCP tools (pipelock-block, capability-block) at
 startup.
 
 Mirrors `backend.docker.provision.supervise` — same `claude mcp
-add` call, just dispatched via `smolvm machine exec` instead of
+add` call, just dispatched via bottle.exec instead of
 `docker exec`, and against `<bundle_ip>:<port>` instead of the
 short `supervise` alias (no DNS in the TSI-allowlisted guest)."""
 
 from __future__ import annotations
 
 from ....log import info, warn
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
 _SUPERVISE_MCP_NAME = "supervise"
 
 
-def provision_supervise(plan: SmolmachinesBottlePlan, target: str) -> None:
+def provision_supervise(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """Run `claude mcp add` inside the guest to register the
     supervise sidecar in claude-code's user config. No-op when
     bottle.supervise is False.
@@ -38,22 +38,13 @@ def provision_supervise(plan: SmolmachinesBottlePlan, target: str) -> None:
         return
     url = plan.agent_supervise_url
     info(f"registering supervise MCP server in agent claude config → {url}")
-    # `claude mcp add --scope user` writes to ~/.claude.json. The
-    # agent is the `node` user; smolvm machine_exec runs as root
-    # by default, so we have to switch user explicitly and set
-    # HOME so the config lands in /home/node/.claude.json (where
-    # the agent's claude actually reads it from).
-    r = _smolvm.machine_exec(
-        target,
-        [
-            "runuser", "-u", "node", "--",
-            "env", "HOME=/home/node",
-            "claude", "mcp", "add",
-            "--scope", "user",
-            "--transport", "http",
-            _SUPERVISE_MCP_NAME,
-            url,
-        ],
+    # `claude mcp add --scope user` writes to ~/.claude.json. Run
+    # as node so the config lands in /home/node/.claude.json.
+    # SmolmachinesBottle.exec sets HOME and USER automatically
+    # for the requested user.
+    r = bottle.exec(
+        f"claude mcp add --scope user --transport http {_SUPERVISE_MCP_NAME} {url}",
+        user="node",
     )
     if r.returncode != 0:
         warn(

bot_bottle/backend/smolmachines/provision/workspace.py

@@ -5,11 +5,11 @@ from __future__ import annotations
 import shlex
 
 from ....log import info
-from .. import smolvm as _smolvm
+from ... import Bottle
 from ..bottle_plan import SmolmachinesBottlePlan
 
 
-def provision_workspace(plan: SmolmachinesBottlePlan, target: str) -> None:
+def provision_workspace(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
     """Copy host cwd contents to the planned guest workspace."""
     workspace = plan.workspace_plan
     if not (workspace.enabled and workspace.copy_contents):
@@ -20,17 +20,13 @@ def provision_workspace(plan: SmolmachinesBottlePlan, target: str) -> None:
     guest_parent_q = shlex.quote(guest_parent)
     owner_q = shlex.quote(workspace.owner)
     mode_q = shlex.quote(workspace.mode)
-    info(f"copying {workspace.host_path} -> {target}:{workspace.guest_path}")
-    _smolvm.machine_exec(
-        target,
-        ["sh", "-c", f"rm -rf {guest_path_q} && mkdir -p {guest_parent_q}"],
+    info(f"copying {workspace.host_path} -> {bottle.name}:{workspace.guest_path}")
+    bottle.exec(
+        f"rm -rf {guest_path_q} && mkdir -p {guest_parent_q}",
+        user="root",
     )
-    _smolvm.machine_cp(str(workspace.host_path), f"{target}:{workspace.guest_path}")
-    _smolvm.machine_exec(
-        target,
-        [
-            "sh", "-c",
-            f"chown -R {owner_q} {guest_path_q} && "
-            f"chmod {mode_q} {guest_path_q}",
-        ],
+    bottle.cp_in(str(workspace.host_path), workspace.guest_path)
+    bottle.exec(
+        f"chown -R {owner_q} {guest_path_q} && chmod {mode_q} {guest_path_q}",
+        user="root",
     )

tests/unit/test_docker_provision_git_user.py

@@ -1,20 +1,20 @@
 """Unit: docker backend `_provision_git_user` (issue #86).
 
-Mocks `subprocess.run` and asserts the `docker exec -u node …
-git config --global …` argv shape. The cwd + git-gate passes
-are covered indirectly by the existing integration-shaped tests
-in test_smolmachines_provision; this file targets just the new
-git_user pass."""
+Mocks `bottle.exec` / `bottle.cp_in` and asserts on the script
+strings and user parameter. The cwd + git-gate passes are covered
+indirectly by the existing integration-shaped tests in
+test_smolmachines_provision; this file targets just the git_user
+pass."""
 
 from __future__ import annotations
 
 import tempfile
 import unittest
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, call
 
 from bot_bottle.agent_provider import AgentProvisionPlan
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import Bottle, BottleSpec, ExecResult
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.docker.provision import git as _git
 from bot_bottle.egress import EgressPlan
@@ -82,16 +82,22 @@ def _plan(*, git_user: dict | None = None,
     )
 
 
-def _git_config_calls(mock_run) -> list[list[str]]:
-    """Filter `subprocess.run` calls down to the ones that run
-    `git config --global` inside the bottle, returning each argv."""
-    out: list[list[str]] = []
-    for call in mock_run.call_args_list:
-        argv = call.args[0]
-        if (len(argv) >= 5
-                and argv[0] == "docker" and argv[1] == "exec"
-                and "git" in argv and "config" in argv):
-            out.append(list(argv))
+def _make_bottle(name: str = "bot-bottle-demo-abc12") -> MagicMock:
+    bottle = MagicMock(spec=Bottle)
+    bottle.name = name
+    bottle.exec.return_value = ExecResult(returncode=0, stdout="", stderr="")
+    return bottle
+
+
+def _git_config_exec_calls(bottle: MagicMock) -> list[tuple[str, str]]:
+    """Filter bottle.exec calls to git-config invocations.
+    Returns list of (script, user) tuples."""
+    out = []
+    for c in bottle.exec.call_args_list:
+        script = c.args[0] if c.args else c.kwargs.get("script", "")
+        user = c.kwargs.get("user", c.args[1] if len(c.args) > 1 else "node")
+        if "git config" in script:
+            out.append((script, user))
     return out
 
 
@@ -104,71 +110,65 @@ class TestProvisionGitUser(unittest.TestCase):
         self._tmp.cleanup()
 
     def test_noop_when_no_git_user(self):
-        with patch.object(_git.subprocess, "run") as run:
-            _git._provision_git_user(
-                _plan(stage_dir=self.stage), "bot-bottle-demo-abc12",
-            )
-        self.assertEqual([], _git_config_calls(run))
+        bottle = _make_bottle()
+        _git._provision_git_user(_plan(stage_dir=self.stage), bottle)
+        self.assertEqual([], _git_config_exec_calls(bottle))
 
     def test_copies_cwd_git_to_workspace_plan_path(self):
         cwd = self.stage / "cwd"
         (cwd / ".git").mkdir(parents=True)
         plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage)
-        with patch.object(_git.subprocess, "run") as run:
-            _git._provision_cwd_git(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle()
+        _git._provision_cwd_git(plan, bottle)
 
-        self.assertEqual(
-            [
-                "docker", "cp", f"{cwd}/.git",
-                "bot-bottle-demo-abc12:/home/node/workspace/.git",
-            ],
-            run.call_args_list[0].args[0],
-        )
-        self.assertEqual(
-            [
-                "docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-                "chown", "-R", "node:node", "/home/node/workspace/.git",
-            ],
-            run.call_args_list[1].args[0],
+        bottle.cp_in.assert_called_once_with(
+            f"{cwd}/.git",
+            "/home/node/workspace/.git",
         )
+        chown_calls = [
+            c for c in bottle.exec.call_args_list
+            if "chown" in (c.args[0] if c.args else "")
+        ]
+        self.assertEqual(1, len(chown_calls))
+        self.assertIn("node:node", chown_calls[0].args[0])
+        self.assertIn("/home/node/workspace/.git", chown_calls[0].args[0])
 
     def test_sets_name_and_email(self):
         plan = _plan(
             git_user={"name": "Eric Bauerfeld", "email": "eric@dideric.is"},
             stage_dir=self.stage,
         )
-        with patch.object(_git.subprocess, "run") as run:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = _git_config_calls(run)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = _git_config_exec_calls(bottle)
         self.assertEqual(2, len(calls))
-        # All `docker exec` invocations run as `-u node` so the
-        # --global config lands in /home/node/.gitconfig.
-        for argv in calls:
-            self.assertEqual(
-                ["docker", "exec", "-u", "node", "bot-bottle-demo-abc12",
-                 "git", "config", "--global"],
-                argv[:8],
-            )
-        self.assertEqual(["user.name", "Eric Bauerfeld"], calls[0][8:])
-        self.assertEqual(["user.email", "eric@dideric.is"], calls[1][8:])
+        for script, user in calls:
+            self.assertEqual("node", user)
+            self.assertIn("git config --global", script)
+        self.assertIn("user.name", calls[0][0])
+        self.assertIn("Eric Bauerfeld", calls[0][0])
+        self.assertIn("user.email", calls[1][0])
+        self.assertIn("eric@dideric.is", calls[1][0])
 
     def test_name_only_sets_only_name(self):
         plan = _plan(git_user={"name": "Bot"}, stage_dir=self.stage)
-        with patch.object(_git.subprocess, "run") as run:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = _git_config_calls(run)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = _git_config_exec_calls(bottle)
         self.assertEqual(1, len(calls))
-        self.assertEqual(["user.name", "Bot"], calls[0][8:])
+        self.assertIn("user.name", calls[0][0])
+        self.assertIn("Bot", calls[0][0])
 
     def test_email_only_sets_only_email(self):
         plan = _plan(
             git_user={"email": "bot@example.com"}, stage_dir=self.stage,
         )
-        with patch.object(_git.subprocess, "run") as run:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = _git_config_calls(run)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = _git_config_exec_calls(bottle)
         self.assertEqual(1, len(calls))
-        self.assertEqual(["user.email", "bot@example.com"], calls[0][8:])
+        self.assertIn("user.email", calls[0][0])
+        self.assertIn("bot@example.com", calls[0][0])
 
 
 if __name__ == "__main__":

tests/unit/test_docker_provision_provider_auth.py

@@ -4,14 +4,14 @@ from __future__ import annotations
 
 import unittest
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock
 
 from bot_bottle.agent_provider import (
     AgentProvisionDir,
     AgentProvisionFile,
     AgentProvisionPlan,
 )
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import Bottle, BottleSpec, ExecResult
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.docker.provision import provider_auth as _provider_auth
 from bot_bottle.egress import EgressPlan
@@ -110,80 +110,62 @@ def _agent_provision(
     )
 
 
+def _make_bottle(name: str = "bot-bottle-demo-abc12") -> MagicMock:
+    bottle = MagicMock(spec=Bottle)
+    bottle.name = name
+    bottle.exec.return_value = ExecResult(returncode=0, stdout="", stderr="")
+    return bottle
+
+
 class TestProvisionProviderAuth(unittest.TestCase):
     def test_noop_for_non_codex_provider(self):
-        with patch.object(_provider_auth.subprocess, "run") as run:
-            _provider_auth.provision_provider_auth(
-                _plan(agent_provider_template="claude"), "bot-bottle-demo-abc12",
-            )
-        self.assertEqual(0, run.call_count)
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(
+            _plan(agent_provider_template="claude"), bottle,
+        )
+        self.assertEqual(0, bottle.cp_in.call_count)
+        self.assertEqual(0, bottle.exec.call_count)
 
     def test_codex_provider_trusts_launch_dir_without_auth_file(self):
-        with patch.object(_provider_auth.subprocess, "run") as run:
-            _provider_auth.provision_provider_auth(
-                _plan(), "bot-bottle-demo-abc12",
-            )
-        argvs = [call.args[0] for call in run.call_args_list]
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(_plan(), bottle)
+        scripts = [c.args[0] for c in bottle.exec.call_args_list]
+        self.assertTrue(
+            any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts)
+        )
+        cp_calls = [c.args for c in bottle.cp_in.call_args_list]
         self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "mkdir", "-p", "/home/node/.codex"],
-            argvs,
+            ("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
+            cp_calls,
         )
-        trust_config = next(
-            a for a in argvs
-            if a[:2] == ["docker", "cp"] and a[2] == "/tmp/codex-config.toml"
+        self.assertTrue(
+            any("chown" in s and "/home/node/.codex/config.toml" in s for s in scripts)
         )
-        self.assertEqual(
-            "bot-bottle-demo-abc12:/home/node/.codex/config.toml",
-            trust_config[3],
-        )
-        self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chown", "node:node", "/home/node/.codex/config.toml"],
-            argvs,
-        )
-        self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chmod", "600", "/home/node/.codex/config.toml"],
-            argvs,
+        self.assertTrue(
+            any("chmod" in s and "/home/node/.codex/config.toml" in s for s in scripts)
         )
 
     def test_copies_dummy_auth_json_to_codex_home(self):
-        with patch.object(_provider_auth.subprocess, "run") as run:
-            _provider_auth.provision_provider_auth(
-                _plan(codex_auth_file=Path("/tmp/codex-auth.json")),
-                "bot-bottle-demo-abc12",
-            )
-        argvs = [call.args[0] for call in run.call_args_list]
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(
+            _plan(codex_auth_file=Path("/tmp/codex-auth.json")),
+            bottle,
+        )
+        cp_calls = [c.args for c in bottle.cp_in.call_args_list]
         self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "mkdir", "-p", "/home/node/.codex"],
-            argvs,
+            ("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
+            cp_calls,
         )
         self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chown", "node:node", "/home/node/.codex"],
-            argvs,
+            ("/tmp/codex-auth.json", "/home/node/.codex/auth.json"),
+            cp_calls,
         )
-        self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chmod", "700", "/home/node/.codex"],
-            argvs,
+        scripts = [c.args[0] for c in bottle.exec.call_args_list]
+        self.assertTrue(
+            any("chown" in s and "/home/node/.codex/auth.json" in s for s in scripts)
         )
-        self.assertIn(
-            ["docker", "cp", "/tmp/codex-auth.json",
-             "bot-bottle-demo-abc12:/home/node/.codex/auth.json"],
-            argvs,
-        )
-        self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chown", "node:node", "/home/node/.codex/auth.json"],
-            argvs,
-        )
-        self.assertIn(
-            ["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
-             "chmod", "600", "/home/node/.codex/auth.json"],
-            argvs,
+        self.assertTrue(
+            any("chmod" in s and "/home/node/.codex/auth.json" in s for s in scripts)
         )
 
 

tests/unit/test_smolmachines_provision.py

@@ -1,8 +1,8 @@
 """Unit: smolmachines provisioning helpers (PRD 0023 chunks 4a + 4d).
 
-Tests mock `smolvm.machine_cp` / `smolvm.machine_exec` and assert
-on the dispatched call shape. The real round-trip lives in the
-chunk-4 integration smoke."""
+Tests mock `bottle.exec` / `bottle.cp_in` and assert on the
+dispatched script shape. The real round-trip lives in the chunk-4
+integration smoke."""
 
 from __future__ import annotations
 
@@ -11,7 +11,7 @@ import tempfile
 import unittest
 from dataclasses import replace
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from bot_bottle.agent_provider import (
     AgentProvisionCommand,
@@ -19,7 +19,7 @@ from bot_bottle.agent_provider import (
     AgentProvisionFile,
     AgentProvisionPlan,
 )
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import Bottle, BottleSpec, ExecResult
 from bot_bottle.backend.smolmachines.bottle_plan import (
     SmolmachinesBottlePlan,
 )
@@ -33,7 +33,6 @@ from bot_bottle.backend.smolmachines.provision import (
     workspace as _workspace,
 )
 from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
-from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
 from bot_bottle.egress import EgressPlan, EgressRoute
 from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
 from bot_bottle.manifest import GitEntry, Manifest
@@ -42,6 +41,28 @@ from bot_bottle.supervise import SupervisePlan
 from bot_bottle.workspace import workspace_plan
 
 
+def _make_bottle(
+    name: str = "bot-bottle-demo-abc12",
+    exec_result: ExecResult | None = None,
+) -> MagicMock:
+    bottle = MagicMock(spec=Bottle)
+    bottle.name = name
+    bottle.exec.return_value = (
+        exec_result if exec_result is not None
+        else ExecResult(returncode=0, stdout="", stderr="")
+    )
+    return bottle
+
+
+def _exec_scripts(bottle: MagicMock) -> list[str]:
+    """All script strings passed to bottle.exec, in call order."""
+    return [c.args[0] for c in bottle.exec.call_args_list]
+
+
+def _exec_users(bottle: MagicMock) -> list[str]:
+    """user= kwarg from each bottle.exec call, in order."""
+    return [c.kwargs.get("user", "node") for c in bottle.exec.call_args_list]
+
 
 def _plan(
     *,
@@ -203,235 +224,181 @@ def _agent_provision(
 
 
 class TestProvisionPrompt(unittest.TestCase):
-    def test_cp_uses_smolvm_machine_cp_with_machine_path_syntax(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
-        ):
-            _prompt.provision_prompt(_plan(), "bot-bottle-demo-abc12")
-        cp.assert_called_once_with(
+    def test_cp_uses_bottle_cp_in(self):
+        bottle = _make_bottle()
+        _prompt.provision_prompt(_plan(), bottle)
+        bottle.cp_in.assert_called_once_with(
             "/tmp/state/demo-abc12/agent/prompt.txt",
-            "bot-bottle-demo-abc12:/home/node/.bot-bottle-prompt.txt",
+            "/home/node/.bot-bottle-prompt.txt",
         )
 
     def test_returns_path_when_agent_has_prompt(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
-        ), patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
-        ):
-            r = _prompt.provision_prompt(
-                _plan(agent_prompt="You are a helpful assistant."),
-                "bot-bottle-demo-abc12",
-            )
+        bottle = _make_bottle()
+        r = _prompt.provision_prompt(
+            _plan(agent_prompt="You are a helpful assistant."),
+            bottle,
+        )
         self.assertEqual("/home/node/.bot-bottle-prompt.txt", r)
 
     def test_returns_none_when_agent_has_no_prompt(self):
         # The file is still copied (path-must-exist contract);
         # only the return value differs.
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
-        ):
-            r = _prompt.provision_prompt(_plan(agent_prompt=""), "bot-bottle-demo-abc12")
+        bottle = _make_bottle()
+        r = _prompt.provision_prompt(_plan(agent_prompt=""), bottle)
         self.assertIsNone(r)
-        cp.assert_called_once()
+        bottle.cp_in.assert_called_once()
 
     def test_chowns_to_node_after_copy(self):
-        # machine cp lands as root; without the chown, the node user
+        # cp_in lands as root; without the chown, the node user
         # can't read its own mode-600 prompt.
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
-        ), patch(
-            "bot_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
-        ) as ex:
-            _prompt.provision_prompt(_plan(), "bot-bottle-demo-abc12")
-        argv_seen = [call.args[1] for call in ex.call_args_list]
-        self.assertIn(
-            ["chown", "node:node", "/home/node/.bot-bottle-prompt.txt"],
-            argv_seen,
+        bottle = _make_bottle()
+        _prompt.provision_prompt(_plan(), bottle)
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(
+            any("chown node:node" in s and "/home/node/.bot-bottle-prompt.txt" in s
+                for s in scripts)
         )
-        self.assertIn(
-            ["chmod", "600", "/home/node/.bot-bottle-prompt.txt"],
-            argv_seen,
+        self.assertTrue(
+            any("chmod 600" in s and "/home/node/.bot-bottle-prompt.txt" in s
+                for s in scripts)
         )
 
 
 class TestProvisionProviderAuth(unittest.TestCase):
-    def _patch(self):
-        return (
-            patch(
-                "bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_cp"
-            ),
-            patch(
-                "bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_exec"
-            ),
-        )
-
     def test_noop_for_non_codex_provider(self):
-        cp_p, ex_p = self._patch()
-        with cp_p as cp, ex_p as ex:
-            _provider_auth.provision_provider_auth(_plan(), "bot-bottle-demo-abc12")
-        self.assertEqual(0, cp.call_count)
-        self.assertEqual(0, ex.call_count)
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(_plan(), bottle)
+        self.assertEqual(0, bottle.cp_in.call_count)
+        self.assertEqual(0, bottle.exec.call_count)
 
     def test_codex_provider_trusts_launch_dir_without_auth_file(self):
-        cp_p, ex_p = self._patch()
-        with cp_p as cp, ex_p as ex:
-            ex.return_value = SmolvmRunResult(0, "", "")
-            _provider_auth.provision_provider_auth(
-                _plan(agent_provider_template="codex"),
-                "bot-bottle-demo-abc12",
-            )
-        cp.assert_called_once_with(
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(
+            _plan(agent_provider_template="codex"),
+            bottle,
+        )
+        bottle.cp_in.assert_called_once_with(
             "/tmp/codex-config.toml",
-            "bot-bottle-demo-abc12:/home/node/.codex/config.toml",
+            "/home/node/.codex/config.toml",
         )
-        argv_seen = [call.args[1] for call in ex.call_args_list]
-        self.assertIn(["mkdir", "-p", "/home/node/.codex"], argv_seen)
-        self.assertIn(
-            ["chown", "node:node", "/home/node/.codex/config.toml"],
-            argv_seen,
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts))
+        self.assertTrue(
+            any("chown" in s and "node:node" in s and "/home/node/.codex/config.toml" in s
+                for s in scripts)
+        )
+        self.assertTrue(
+            any("chmod" in s and "600" in s and "/home/node/.codex/config.toml" in s
+                for s in scripts)
         )
-        self.assertIn(["chmod", "600", "/home/node/.codex/config.toml"], argv_seen)
 
     def test_copies_dummy_auth_json_to_codex_home(self):
-        cp_p, ex_p = self._patch()
-        with cp_p as cp, ex_p as ex:
-            ex.return_value = SmolvmRunResult(0, "Logged in using ChatGPT\n", "")
-            _provider_auth.provision_provider_auth(
-                _plan(
-                    agent_provider_template="codex",
-                    codex_auth_file=Path("/tmp/codex-auth.json"),
-                ),
-                "bot-bottle-demo-abc12",
-            )
-        cp_calls = [call.args for call in cp.call_args_list]
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(
+            _plan(
+                agent_provider_template="codex",
+                codex_auth_file=Path("/tmp/codex-auth.json"),
+            ),
+            bottle,
+        )
+        cp_calls = [c.args for c in bottle.cp_in.call_args_list]
         self.assertIn(
-            ("/tmp/codex-config.toml",
-             "bot-bottle-demo-abc12:/home/node/.codex/config.toml"),
+            ("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
             cp_calls,
         )
         self.assertIn(
-            ("/tmp/codex-auth.json",
-             "bot-bottle-demo-abc12:/home/node/.codex/auth.json"),
+            ("/tmp/codex-auth.json", "/home/node/.codex/auth.json"),
             cp_calls,
         )
-        argv_seen = [call.args[1] for call in ex.call_args_list]
-        self.assertIn(["mkdir", "-p", "/home/node/.codex"], argv_seen)
-        self.assertIn(
-            ["chown", "node:node", "/home/node/.codex"],
-            argv_seen,
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts))
+        self.assertTrue(
+            any("chown" in s and "node:node" in s and s.rstrip().endswith("/home/node/.codex")
+                for s in scripts)
         )
-        self.assertIn(
-            ["chmod", "700", "/home/node/.codex"],
-            argv_seen,
+        self.assertTrue(
+            any("chmod" in s and "700" in s and s.rstrip().endswith("/home/node/.codex")
+                for s in scripts)
         )
-        self.assertIn(
-            [
-                "find", "/home/node/.codex",
-                "-maxdepth", "1",
-                "-type", "f",
-                "(",
-                "-name", "*.sqlite",
-                "-o", "-name", "*.sqlite-*",
-                "-o", "-name", "*.codex-repair-*.bak",
-                ")",
-                "-delete",
-            ],
-            argv_seen,
+        # The pre_copy `find ... -delete` script should be present
+        # (shlex.join properly quotes the `(`/`)`/`*.sqlite`).
+        self.assertTrue(
+            any("find" in s and "-delete" in s and "*.sqlite" in s for s in scripts)
         )
-        self.assertIn(
-            ["chown", "node:node", "/home/node/.codex/auth.json"],
-            argv_seen,
+        self.assertTrue(
+            any("chown" in s and "/home/node/.codex/auth.json" in s for s in scripts)
         )
-        self.assertIn(["chmod", "600", "/home/node/.codex/auth.json"], argv_seen)
-        self.assertIn(
-            [
-                "runuser", "-u", "node", "--",
-                "env",
-                "HOME=/home/node",
-                "CODEX_HOME=/home/node/.codex",
-                "codex", "login", "status",
-            ],
-            argv_seen,
+        self.assertTrue(
+            any("chmod" in s and "600" in s and "/home/node/.codex/auth.json" in s
+                for s in scripts)
+        )
+        # Verify command runs `codex login status` via runuser node.
+        self.assertTrue(
+            any("runuser" in s and "codex login status" in s for s in scripts)
         )
 
     def test_honors_codex_home_from_guest_env(self):
-        cp_p, ex_p = self._patch()
-        with cp_p as cp, ex_p as ex:
-            ex.return_value = SmolvmRunResult(0, "Logged in using ChatGPT\n", "")
+        bottle = _make_bottle()
+        _provider_auth.provision_provider_auth(
+            _plan(
+                agent_provider_template="codex",
+                codex_auth_file=Path("/tmp/codex-auth.json"),
+                guest_env={"CODEX_HOME": "/run/codex-home"},
+            ),
+            bottle,
+        )
+        cp_calls = [c.args for c in bottle.cp_in.call_args_list]
+        self.assertIn(
+            ("/tmp/codex-config.toml", "/run/codex-home/config.toml"),
+            cp_calls,
+        )
+        self.assertIn(
+            ("/tmp/codex-auth.json", "/run/codex-home/auth.json"),
+            cp_calls,
+        )
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(
+            any("runuser" in s and "CODEX_HOME=/run/codex-home" in s and "codex login status" in s
+                for s in scripts)
+        )
+
+    def test_dies_when_codex_home_cannot_be_created(self):
+        bottle = _make_bottle(
+            exec_result=ExecResult(1, "", "mkdir: nope\n"),
+        )
+        with self.assertRaises(SystemExit):
             _provider_auth.provision_provider_auth(
                 _plan(
                     agent_provider_template="codex",
                     codex_auth_file=Path("/tmp/codex-auth.json"),
-                    guest_env={"CODEX_HOME": "/run/codex-home"},
                 ),
-                "bot-bottle-demo-abc12",
+                bottle,
             )
-        cp_calls = [call.args for call in cp.call_args_list]
-        self.assertIn(
-            ("/tmp/codex-config.toml",
-             "bot-bottle-demo-abc12:/run/codex-home/config.toml"),
-            cp_calls,
-        )
-        self.assertIn(
-            ("/tmp/codex-auth.json",
-             "bot-bottle-demo-abc12:/run/codex-home/auth.json"),
-            cp_calls,
-        )
-        argv_seen = [call.args[1] for call in ex.call_args_list]
-        self.assertIn(
-            [
-                "runuser", "-u", "node", "--",
-                "env",
-                "HOME=/home/node",
-                "CODEX_HOME=/run/codex-home",
-                "codex", "login", "status",
-            ],
-            argv_seen,
-        )
-
-    def test_dies_when_codex_home_cannot_be_created(self):
-        cp_p, ex_p = self._patch()
-        with cp_p as cp, ex_p as ex:
-            ex.return_value = SmolvmRunResult(1, "", "mkdir: nope\n")
-            with self.assertRaises(SystemExit):
-                _provider_auth.provision_provider_auth(
-                    _plan(
-                        agent_provider_template="codex",
-                        codex_auth_file=Path("/tmp/codex-auth.json"),
-                    ),
-                    "bot-bottle-demo-abc12",
-                )
-        self.assertEqual(0, cp.call_count)
-        self.assertEqual(1, ex.call_count)
+        self.assertEqual(0, bottle.cp_in.call_count)
+        self.assertEqual(1, bottle.exec.call_count)
 
     def test_dies_when_codex_rejects_dummy_auth(self):
-        cp_p, ex_p = self._patch()
-        with cp_p, ex_p as ex:
-            # CODEX_HOME setup ok (0), but codex login status fails (1).
-            ex.side_effect = [
-                SmolvmRunResult(0, "", ""),  # mkdir CODEX_HOME
-                SmolvmRunResult(0, "", ""),  # chown CODEX_HOME
-                SmolvmRunResult(0, "", ""),  # chmod CODEX_HOME
-                SmolvmRunResult(0, "", ""),  # reset runtime db files
-                SmolvmRunResult(0, "", ""),  # chown config.toml
-                SmolvmRunResult(0, "", ""),  # chmod config.toml
-                SmolvmRunResult(0, "", ""),  # chown auth.json
-                SmolvmRunResult(0, "", ""),  # chmod auth.json
-                SmolvmRunResult(1, "Not logged in\n", ""),  # login status
-            ]
-            with self.assertRaises(SystemExit):
-                _provider_auth.provision_provider_auth(
-                    _plan(
-                        agent_provider_template="codex",
-                        codex_auth_file=Path("/tmp/codex-auth.json"),
-                    ),
-                    "bot-bottle-demo-abc12",
-                )
+        # CODEX_HOME setup ok, but codex login status fails (last exec).
+        bottle = _make_bottle()
+        bottle.exec.side_effect = [
+            ExecResult(0, "", ""),  # mkdir CODEX_HOME
+            ExecResult(0, "", ""),  # chown CODEX_HOME
+            ExecResult(0, "", ""),  # chmod CODEX_HOME
+            ExecResult(0, "", ""),  # find ... -delete (pre_copy)
+            ExecResult(0, "", ""),  # chown config.toml
+            ExecResult(0, "", ""),  # chmod config.toml
+            ExecResult(0, "", ""),  # chown auth.json
+            ExecResult(0, "", ""),  # chmod auth.json
+            ExecResult(1, "Not logged in\n", ""),  # login status (verify)
+        ]
+        with self.assertRaises(SystemExit):
+            _provider_auth.provision_provider_auth(
+                _plan(
+                    agent_provider_template="codex",
+                    codex_auth_file=Path("/tmp/codex-auth.json"),
+                ),
+                bottle,
+            )
 
 
 class TestProvisionSkills(unittest.TestCase):
@@ -442,98 +409,69 @@ class TestProvisionSkills(unittest.TestCase):
         )
 
     def test_no_op_when_agent_has_no_skills(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_exec"
-        ) as ex:
-            _skills.provision_skills(_plan(skills=[]), "bot-bottle-demo-abc12")
-        self.assertEqual(0, cp.call_count)
-        self.assertEqual(0, ex.call_count)
+        bottle = _make_bottle()
+        _skills.provision_skills(_plan(skills=[]), bottle)
+        self.assertEqual(0, bottle.cp_in.call_count)
+        self.assertEqual(0, bottle.exec.call_count)
 
     def test_mkdir_plus_cp_per_skill(self):
+        bottle = _make_bottle()
         with self._patch_host_skill_dir({
             "init-prd": "/host/skills/init-prd",
             "verify": "/host/skills/verify",
         }), patch(
             "bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
             return_value=True,
-        ), patch(
-            "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_exec"
-        ) as ex:
+        ):
             _skills.provision_skills(
                 _plan(skills=["init-prd", "verify"]),
-                "bot-bottle-demo-abc12",
+                bottle,
             )
 
-        # mkdir -p once + (rm -rf + chown) per skill = 5 exec calls.
-        self.assertEqual(5, ex.call_count)
-        mkdir_call = ex.call_args_list[0]
-        self.assertEqual(
-            ("bot-bottle-demo-abc12", ["mkdir", "-p", "/home/node/.claude/skills"]),
-            mkdir_call.args,
+        # mkdir skills_dir once + (rm -rf + chown) per skill = 5 exec calls.
+        self.assertEqual(5, bottle.exec.call_count)
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(
+            any("mkdir -p" in s and "/home/node/.claude/skills" in s for s in scripts)
         )
         # Two cp calls, one per skill, into the per-skill subdir.
-        self.assertEqual(2, cp.call_count)
-        cp_targets = {call.args[1] for call in cp.call_args_list}
-        self.assertEqual(
-            {
-                "bot-bottle-demo-abc12:/home/node/.claude/skills/init-prd",
-                "bot-bottle-demo-abc12:/home/node/.claude/skills/verify",
-            },
-            cp_targets,
-        )
-        # Each skill gets a chown -R node:node so claude can read it.
-        chown_argvs = [
-            call.args[1] for call in ex.call_args_list
-            if call.args[1][:1] == ["chown"]
-        ]
-        self.assertEqual(2, len(chown_argvs))
-        chown_targets = {argv[-1] for argv in chown_argvs}
+        self.assertEqual(2, bottle.cp_in.call_count)
+        cp_targets = {call.args[1] for call in bottle.cp_in.call_args_list}
         self.assertEqual(
             {
                 "/home/node/.claude/skills/init-prd",
                 "/home/node/.claude/skills/verify",
             },
-            chown_targets,
+            cp_targets,
         )
+        # Each skill gets a chown -R node:node so claude can read it.
+        chown_scripts = [s for s in scripts if "chown -R node:node" in s]
+        self.assertEqual(2, len(chown_scripts))
 
     def test_skills_dir_overridable_via_env(self):
         import os
+        bottle = _make_bottle()
         with self._patch_host_skill_dir({"init-prd": "/host/skills/init-prd"}), \
              patch(
                  "bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
                  return_value=True,
              ), \
-             patch.dict(os.environ, {"BOT_BOTTLE_GUEST_SKILLS_DIR": "/home/node/.claude/skills"}), \
-             patch(
-                 "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_cp"
-             ) as cp, \
-             patch(
-                 "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_exec"
-             ):
-            _skills.provision_skills(_plan(skills=["init-prd"]), "bot-bottle-demo-abc12")
+             patch.dict(os.environ, {"BOT_BOTTLE_GUEST_SKILLS_DIR": "/home/node/.claude/skills"}):
+            _skills.provision_skills(_plan(skills=["init-prd"]), bottle)
         self.assertEqual(
-            "bot-bottle-demo-abc12:/home/node/.claude/skills/init-prd",
-            cp.call_args.args[1],
+            "/home/node/.claude/skills/init-prd",
+            bottle.cp_in.call_args.args[1],
         )
 
     def test_missing_skill_dies(self):
+        bottle = _make_bottle()
         with self._patch_host_skill_dir({"init-prd": "/host/skills/init-prd"}), \
              patch(
                  "bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
                  return_value=False,
-             ), \
-             patch(
-                 "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_cp"
-             ), \
-             patch(
-                 "bot_bottle.backend.smolmachines.provision.skills._smolvm.machine_exec"
              ):
             with self.assertRaises(SystemExit):
-                _skills.provision_skills(_plan(skills=["init-prd"]), "bot-bottle-demo-abc12")
+                _skills.provision_skills(_plan(skills=["init-prd"]), bottle)
 
 
 def _write_self_signed_cert(path: Path) -> None:
@@ -553,7 +491,7 @@ def _write_self_signed_cert(path: Path) -> None:
 class TestProvisionCA(unittest.TestCase):
     """provision_ca selects the right CA cert (egress when the
     bottle has routes, else pipelock) and dispatches
-    machine_cp + machine_exec in the right order."""
+    cp_in + exec in the right order."""
 
     def setUp(self):
         self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-ca.")
@@ -566,10 +504,10 @@ class TestProvisionCA(unittest.TestCase):
     def tearDown(self):
         self._tmp.cleanup()
 
-    # provision_ca dies hard if update-ca-certificates' stdout
-    # doesn't include "1 added"; supply a stock success return
-    # so the bulk of the tests below exercise the happy path.
-    _UPDATE_OK = SmolvmRunResult(
+    # provision_ca dies hard if update-ca-certificates' exit
+    # is non-zero; supply a stock success return so the bulk of
+    # the tests below exercise the happy path.
+    _UPDATE_OK = ExecResult(
         returncode=0,
         stdout="Updating certificates in /etc/ssl/certs...\n1 added, 0 removed; done.\n",
         stderr="",
@@ -577,27 +515,20 @@ class TestProvisionCA(unittest.TestCase):
 
     def test_pipelock_path_when_no_routes(self):
         plan = _plan(pipelock_ca_path=self.pipelock_ca)
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec",
-            return_value=self._UPDATE_OK,
-        ) as ex:
-            _ca.provision_ca(plan, "bot-bottle-demo-abc12")
-        cp.assert_called_once_with(
+        bottle = _make_bottle(exec_result=self._UPDATE_OK)
+        _ca.provision_ca(plan, bottle)
+        bottle.cp_in.assert_called_once_with(
             str(self.pipelock_ca),
-            "bot-bottle-demo-abc12:" + _ca.AGENT_CA_PATH,
+            _ca.AGENT_CA_PATH,
         )
-        # chmod + chown + update-ca-certificates are now folded
-        # into one `sh -c` invocation (working around a smolvm
-        # exec warm-up SIGKILL race), so we look at the single
-        # exec's argv rather than expecting separate calls.
-        ex.assert_called_once()
-        argv = ex.call_args.args[1]
-        self.assertEqual("sh", argv[0])
-        self.assertEqual("-c", argv[1])
-        self.assertIn("chmod 644", argv[2])
-        self.assertIn("update-ca-certificates", argv[2])
+        # chmod + chown + update-ca-certificates are folded into
+        # one exec invocation; look at the single exec's script
+        # rather than expecting separate calls.
+        bottle.exec.assert_called_once()
+        script = bottle.exec.call_args.args[0]
+        self.assertIn("chmod 644", script)
+        self.assertIn("update-ca-certificates", script)
+        self.assertEqual("root", bottle.exec.call_args.kwargs.get("user"))
 
     def test_egress_path_when_routes_declared(self):
         plan = _plan(
@@ -605,51 +536,39 @@ class TestProvisionCA(unittest.TestCase):
             egress_ca_path=self.egress_ca,
             pipelock_ca_path=self.pipelock_ca,
         )
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec",
-            return_value=self._UPDATE_OK,
-        ):
-            _ca.provision_ca(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle(exec_result=self._UPDATE_OK)
+        _ca.provision_ca(plan, bottle)
         # When routes are declared, egress is the agent's first hop,
         # so egress's CA is the one that gets installed.
-        cp.assert_called_once_with(
+        bottle.cp_in.assert_called_once_with(
             str(self.egress_ca),
-            "bot-bottle-demo-abc12:" + _ca.AGENT_CA_PATH,
+            _ca.AGENT_CA_PATH,
         )
 
     def test_retries_smolvm_sigkill_during_update_ca(self):
         plan = _plan(pipelock_ca_path=self.pipelock_ca)
-        killed = SmolvmRunResult(
+        killed = ExecResult(
             returncode=137,
             stdout="Updating certificates in /etc/ssl/certs...\n",
             stderr="",
         )
+        bottle = _make_bottle()
+        bottle.exec.side_effect = [killed, self._UPDATE_OK]
         with patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
-        ), patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec",
-            side_effect=[killed, self._UPDATE_OK],
-        ) as ex, patch(
             "bot_bottle.backend.smolmachines.provision.ca.time.sleep"
         ) as sleep:
-            _ca.provision_ca(plan, "bot-bottle-demo-abc12")
+            _ca.provision_ca(plan, bottle)
 
-        self.assertEqual(2, ex.call_count)
+        self.assertEqual(2, bottle.exec.call_count)
         sleep.assert_called_once_with(1.0)
 
     def test_dies_when_selected_cert_missing(self):
         # Plan claims a pipelock cert at a path that doesn't exist —
         # something went wrong in launch's pipelock_tls_init.
         plan = _plan(pipelock_ca_path=self.tmp / "does-not-exist.pem")
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
-        ), patch(
-            "bot_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec"
-        ):
-            with self.assertRaises(SystemExit):
-                _ca.provision_ca(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle()
+        with self.assertRaises(SystemExit):
+            _ca.provision_ca(plan, bottle)
 
 
 class TestProvisionGit(unittest.TestCase):
@@ -665,16 +584,10 @@ class TestProvisionGit(unittest.TestCase):
         self._tmp.cleanup()
 
     def test_noop_when_no_cwd_and_no_git_entries(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git.provision_git(
-                _plan(stage_dir=self.stage), "bot-bottle-demo-abc12",
-            )
-        cp.assert_not_called()
-        ex.assert_not_called()
+        bottle = _make_bottle()
+        _git.provision_git(_plan(stage_dir=self.stage), bottle)
+        bottle.cp_in.assert_not_called()
+        bottle.exec.assert_not_called()
 
     def test_copies_cwd_git_when_copy_cwd_and_git_present(self):
         # Stage a fake host .git dir under user_cwd so the path-
@@ -684,33 +597,25 @@ class TestProvisionGit(unittest.TestCase):
         plan = _plan(
             copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage,
         )
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git.provision_git(plan, "bot-bottle-demo-abc12")
-        cp.assert_called_once_with(
+        bottle = _make_bottle()
+        _git.provision_git(plan, bottle)
+        bottle.cp_in.assert_called_once_with(
             f"{cwd}/.git",
-            "bot-bottle-demo-abc12:/home/node/workspace/.git",
+            "/home/node/workspace/.git",
         )
-        argvs = [c.args[1] for c in ex.call_args_list]
-        self.assertIn(["mkdir", "-p", "/home/node/workspace"], argvs)
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(any("mkdir -p" in s and "/home/node/workspace" in s for s in scripts))
         # chown the workspace tree so the agent (node) owns it.
-        self.assertIn(
-            ["chown", "-R", "node:node", "/home/node/workspace/.git"],
-            argvs,
+        self.assertTrue(
+            any("chown -R" in s and "node:node" in s and "/home/node/workspace/.git" in s
+                for s in scripts)
         )
 
     def test_skips_cwd_when_copy_cwd_false(self):
         plan = _plan(copy_cwd=False, stage_dir=self.stage)
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ):
-            _git.provision_git(plan, "bot-bottle-demo-abc12")
-        cp.assert_not_called()
+        bottle = _make_bottle()
+        _git.provision_git(plan, bottle)
+        bottle.cp_in.assert_not_called()
 
     def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self):
         # Smolmachines's TSI-allowlisted guest dials git-gate via
@@ -726,15 +631,11 @@ class TestProvisionGit(unittest.TestCase):
             stage_dir=self.stage,
             agent_git_gate_host="127.0.0.1:9418",
         )
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ):
-            _git.provision_git(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle()
+        _git.provision_git(plan, bottle)
         # The staged gitconfig path is whatever NamedTemporaryFile
         # picked; we read its contents.
-        cp_call = cp.call_args
+        cp_call = bottle.cp_in.call_args
         staged_path = Path(cp_call.args[0])
         self.assertEqual(self.stage, staged_path.parent)
         content = staged_path.read_text()
@@ -776,71 +677,63 @@ class TestBundleLaunchSpec(unittest.TestCase):
 
 class TestProvisionGitUser(unittest.TestCase):
     """`_provision_git_user` runs `git config --global` inside the
-    guest as the node user with HOME forced via `smolvm -e`
-    (otherwise --global lands in /root/.gitconfig). No-op when the
-    bottle didn't declare git_user (issue #86)."""
+    guest as the node user. SmolmachinesBottle.exec sets HOME and
+    USER automatically for the requested user, so --global lands
+    in /home/node/.gitconfig. No-op when the bottle didn't declare
+    git_user (issue #86)."""
 
-    def _git_config_calls(self, mock_exec):
-        """Filter machine_exec calls down to git-config invocations,
-        return list of (argv, env-dict) tuples."""
+    def _git_config_calls(self, bottle: MagicMock) -> list[tuple[str, str]]:
+        """Filter bottle.exec calls down to git-config invocations,
+        return list of (script, user) tuples."""
         out = []
-        for c in mock_exec.call_args_list:
-            argv = c.args[1] if len(c.args) > 1 else c.kwargs.get("argv", [])
-            if "git" in argv and "config" in argv:
-                out.append((argv, c.kwargs.get("env") or {}))
+        for c in bottle.exec.call_args_list:
+            script = c.args[0] if c.args else ""
+            user = c.kwargs.get("user", "node")
+            if "git config" in script:
+                out.append((script, user))
         return out
 
     def test_noop_when_no_git_user(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git._provision_git_user(_plan(), "bot-bottle-demo-abc12")
-        self.assertEqual([], self._git_config_calls(ex))
+        bottle = _make_bottle()
+        _git._provision_git_user(_plan(), bottle)
+        self.assertEqual([], self._git_config_calls(bottle))
 
     def test_sets_name_and_email_as_node(self):
         plan = _plan(git_user={
             "name": "Eric Bauerfeld",
             "email": "eric@dideric.is",
         })
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = self._git_config_calls(ex)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = self._git_config_calls(bottle)
         self.assertEqual(2, len(calls))
-        # Both go through `runuser -u node --` so they run as node;
-        # HOME is forced via smolvm -e so --global writes to
-        # /home/node/.gitconfig and not /root/.gitconfig.
-        for argv, env in calls:
-            self.assertEqual(
-                ["runuser", "-u", "node", "--",
-                 "git", "config", "--global"],
-                argv[:7],
-            )
-            self.assertEqual("/home/node", env.get("HOME"))
-            self.assertEqual("node", env.get("USER"))
-        self.assertEqual(["user.name", "Eric Bauerfeld"], calls[0][0][7:])
-        self.assertEqual(["user.email", "eric@dideric.is"], calls[1][0][7:])
+        # Both run as node so SmolmachinesBottle.exec sets HOME=/home/node
+        # automatically, ensuring --global writes to /home/node/.gitconfig.
+        for script, user in calls:
+            self.assertEqual("node", user)
+            self.assertIn("git config --global", script)
+        self.assertIn("user.name", calls[0][0])
+        self.assertIn("Eric Bauerfeld", calls[0][0])
+        self.assertIn("user.email", calls[1][0])
+        self.assertIn("eric@dideric.is", calls[1][0])
 
     def test_name_only(self):
         plan = _plan(git_user={"name": "Bot"})
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = self._git_config_calls(ex)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = self._git_config_calls(bottle)
         self.assertEqual(1, len(calls))
-        self.assertEqual(["user.name", "Bot"], calls[0][0][7:])
+        self.assertIn("user.name", calls[0][0])
+        self.assertIn("Bot", calls[0][0])
 
     def test_email_only(self):
         plan = _plan(git_user={"email": "bot@example.com"})
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
-        ) as ex:
-            _git._provision_git_user(plan, "bot-bottle-demo-abc12")
-        calls = self._git_config_calls(ex)
+        bottle = _make_bottle()
+        _git._provision_git_user(plan, bottle)
+        calls = self._git_config_calls(bottle)
         self.assertEqual(1, len(calls))
-        self.assertEqual(["user.email", "bot@example.com"], calls[0][0][7:])
+        self.assertIn("user.email", calls[0][0])
+        self.assertIn("bot@example.com", calls[0][0])
 
 
 class TestProvisionWorkspace(unittest.TestCase):
@@ -853,94 +746,68 @@ class TestProvisionWorkspace(unittest.TestCase):
 
     def test_noop_when_copy_cwd_false(self):
         plan = _plan(copy_cwd=False, stage_dir=self.stage)
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_exec"
-        ) as ex:
-            _workspace.provision_workspace(plan, "bot-bottle-demo-abc12")
-        cp.assert_not_called()
-        ex.assert_not_called()
+        bottle = _make_bottle()
+        _workspace.provision_workspace(plan, bottle)
+        bottle.cp_in.assert_not_called()
+        bottle.exec.assert_not_called()
 
     def test_copies_workspace_to_plan_path_and_chowns(self):
         cwd = self.stage / "cwd"
         cwd.mkdir()
         plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage)
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_cp"
-        ) as cp, patch(
-            "bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_exec"
-        ) as ex:
-            _workspace.provision_workspace(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle()
+        _workspace.provision_workspace(plan, bottle)
 
-        cp.assert_called_once_with(
+        bottle.cp_in.assert_called_once_with(
             str(cwd),
-            "bot-bottle-demo-abc12:/home/node/workspace",
+            "/home/node/workspace",
         )
-        argvs = [c.args[1] for c in ex.call_args_list]
-        self.assertIn(
-            ["sh", "-c", "rm -rf /home/node/workspace && mkdir -p /home/node"],
-            argvs,
+        scripts = _exec_scripts(bottle)
+        self.assertTrue(
+            any("rm -rf /home/node/workspace" in s and "mkdir -p /home/node" in s
+                for s in scripts)
         )
-        self.assertIn(
-            [
-                "sh", "-c",
-                "chown -R node:node /home/node/workspace && "
-                "chmod 755 /home/node/workspace",
-            ],
-            argvs,
+        self.assertTrue(
+            any("chown -R node:node /home/node/workspace" in s
+                and "chmod 755 /home/node/workspace" in s
+                for s in scripts)
         )
 
 
 class TestProvisionSupervise(unittest.TestCase):
     def test_noop_when_supervise_not_enabled(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec"
-        ) as ex:
-            _supervise.provision_supervise(_plan(), "bot-bottle-demo-abc12")
-        ex.assert_not_called()
+        bottle = _make_bottle()
+        _supervise.provision_supervise(_plan(), bottle)
+        bottle.exec.assert_not_called()
 
     def test_calls_claude_mcp_add_when_supervise_enabled(self):
         plan = _plan(
             supervise=True,
             agent_supervise_url="http://127.0.0.1:9100/",
         )
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec",
-            return_value=SmolvmRunResult(returncode=0, stdout="", stderr=""),
-        ) as ex:
-            _supervise.provision_supervise(plan, "bot-bottle-demo-abc12")
-        ex.assert_called_once()
-        argv = ex.call_args.args[1]
-        # `claude mcp add --scope user` writes to ~/.claude.json,
-        # and the agent is the `node` user — switch UID + set
-        # HOME so the config lands in /home/node/.claude.json,
-        # not root's. URL is the agent-side endpoint (host
-        # loopback + discovered port), not the docker bridge IP.
-        self.assertEqual(
-            [
-                "runuser", "-u", "node", "--",
-                "env", "HOME=/home/node",
-                "claude", "mcp", "add",
-                "--scope", "user",
-                "--transport", "http",
-                "supervise",
-                "http://127.0.0.1:9100/",
-            ],
-            argv,
-        )
+        bottle = _make_bottle()
+        _supervise.provision_supervise(plan, bottle)
+        bottle.exec.assert_called_once()
+        script = bottle.exec.call_args.args[0]
+        user = bottle.exec.call_args.kwargs.get("user")
+        self.assertEqual("node", user)
+        # SmolmachinesBottle.exec(user="node") handles uid switch +
+        # HOME setup automatically — the script itself is just the
+        # claude command.
+        self.assertIn("claude mcp add", script)
+        self.assertIn("--scope user", script)
+        self.assertIn("--transport http", script)
+        self.assertIn("supervise", script)
+        self.assertIn("http://127.0.0.1:9100/", script)
 
     def test_non_zero_exit_logs_warning_but_does_not_raise(self):
         plan = _plan(supervise=True)
-        with patch(
-            "bot_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec",
-            return_value=SmolvmRunResult(
-                returncode=1, stdout="", stderr="boom",
-            ),
-        ):
-            # No raise — the bottle still works without the MCP
-            # entry, so we log and move on.
-            _supervise.provision_supervise(plan, "bot-bottle-demo-abc12")
+        bottle = _make_bottle(
+            exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
+        )
+        # No raise — the bottle still works without the MCP
+        # entry, so we log and move on.
+        _supervise.provision_supervise(plan, bottle)
 
 
 if __name__ == "__main__":