bot_bottle/_provision_apply.py

@@ -0,0 +1,117 @@
+"""Shared apply primitives used by every `AgentProvider` (PRD 0050).
+
+The skills/prompt/provision loops here are backend-agnostic — they
+only touch the `Bottle.exec`/`Bottle.cp_in` primitives that every
+backend implements. Each provider's `provision_skills` /
+`provision_prompt` / `provision` defaults delegate here so the
+business logic lives in exactly one place."""
+
+from __future__ import annotations
+
+import os
+import shlex
+from typing import TYPE_CHECKING
+
+from .agent_provider import GUEST_HOME
+from .log import die, info
+
+
+if TYPE_CHECKING:
+    from .backend import Bottle, BottlePlan
+
+
+# In-guest paths agents look at. These are baked into the bot-bottle
+# Dockerfile and the smolmachines guest image; no env knob exposed,
+# per the PRD 0050 issue feedback.
+SKILLS_DIR = f"{GUEST_HOME}/.claude/skills"
+PROMPT_PATH = f"{GUEST_HOME}/.bot-bottle-prompt.txt"
+
+
+def apply_skills(plan: "BottlePlan", bottle: "Bottle") -> None:
+    """Copy each named skill tree from the host into the guest.
+    No-op when the agent has no skills.
+
+    `cp_in` lands files as root in every backend; chown back to
+    node:node so the agent can read its own skills."""
+    from .backend.util import host_skill_dir
+
+    agent = plan.spec.manifest.agents[plan.spec.agent_name]
+    if not agent.skills:
+        return
+
+    bottle.exec(f"mkdir -p {SKILLS_DIR}", user="root")
+
+    for name in agent.skills:
+        src = host_skill_dir(name)
+        if not os.path.isdir(src):
+            die(
+                f"skill {name!r} disappeared from host between "
+                f"validation and copy at {src}."
+            )
+        dst = f"{SKILLS_DIR}/{name}"
+        info(f"copying skill {name} into {bottle.name}:{dst}")
+        bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
+        bottle.cp_in(f"{src}/.", f"{dst}/")
+        bottle.exec(f"chown -R node:node {dst}", user="root")
+
+
+def apply_prompt(plan: "BottlePlan", bottle: "Bottle") -> str | None:
+    """Copy the prompt file into the guest, fix ownership/mode.
+    Returns the in-guest path iff the agent has a non-empty prompt
+    (drives `--append-system-prompt-file`); the file is copied either
+    way so the path always exists."""
+    bottle.cp_in(str(plan.prompt_file), PROMPT_PATH)
+    bottle.exec(
+        f"chown node:node {PROMPT_PATH} && chmod 600 {PROMPT_PATH}",
+        user="root",
+    )
+    agent = plan.spec.manifest.agents[plan.spec.agent_name]
+    return PROMPT_PATH if agent.prompt else None
+
+
+def apply_provision(plan: "BottlePlan", bottle: "Bottle") -> None:
+    """Apply the declarative `dirs`/`pre_copy`/`files`/`verify` steps
+    from the provider's `AgentProvisionPlan`. This is the loop that
+    used to live in
+    `backend/<name>/provision/provider_auth.py:provision_provider_auth`
+    pre-PRD-0050."""
+    provision = plan.agent_provision
+    for d in provision.dirs:
+        path = shlex.quote(d.guest_path)
+        _exec(bottle, f"mkdir -p {path}", f"could not create {d.guest_path}")
+        _exec(
+            bottle,
+            f"chown {shlex.quote(d.owner)} {path}",
+            f"could not chown {d.guest_path}",
+        )
+        _exec(
+            bottle,
+            f"chmod {shlex.quote(d.mode)} {path}",
+            f"could not chmod {d.guest_path}",
+        )
+    for command in provision.pre_copy:
+        _exec(bottle, shlex.join(command.argv), command.error)
+    for f in provision.files:
+        bottle.cp_in(str(f.host_path), f.guest_path)
+        path = shlex.quote(f.guest_path)
+        _exec(
+            bottle,
+            f"chown {shlex.quote(f.owner)} {path}",
+            f"could not chown {f.guest_path}",
+        )
+        _exec(
+            bottle,
+            f"chmod {shlex.quote(f.mode)} {path}",
+            f"could not chmod {f.guest_path}",
+        )
+    for command in provision.verify:
+        _exec(bottle, shlex.join(command.argv), command.error)
+
+
+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/agent_provider.py

@@ -3,18 +3,32 @@
 The manifest owns the user-facing AgentProvider shape. This module is
 the launch-time table that turns a provider template into an executable
 command, default image, and prompt/auth behavior.
+
+Per PRD 0050 the per-provider implementations live under
+`bot_bottle/contrib/<template>/agent_provider.py`. This module exposes:
+
+  - `AgentProvider` (ABC) — the contract each plugin implements.
+  - `get_provider(template)` — lazy-imported registry; the analogue
+    of `bot_bottle/deploy_key_provisioner.get_provisioner`.
+  - `AgentProvisionPlan` (+ helper dataclasses) — declarative shape
+    each provider produces and the backends consume unchanged.
+  - `agent_provision_plan` / `runtime_for` — thin wrappers around the
+    registry kept so existing callers keep working without per-call
+    edits.
 """
 
 from __future__ import annotations
 
-import json
+from abc import ABC, abstractmethod
-import os
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
 
-from .codex_auth import codex_host_access_token, write_codex_dummy_auth_file
+from .egress import EgressRoute
-from .egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
+
+
+if TYPE_CHECKING:
+    from .backend import Bottle, BottlePlan
 
 
 PROVIDER_CLAUDE = "claude"
@@ -27,6 +41,8 @@ PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX})
 CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
 PromptMode = Literal["append_file", "read_prompt_file"]
 
+GUEST_HOME = "/home/node"
+
 
 @dataclass(frozen=True)
 class AgentProviderRuntime:
@@ -96,35 +112,96 @@ class AgentProvisionPlan:
     provisioned_env: dict[str, str] = field(default_factory=dict)
 
 
-_REPO_ROOT = Path(__file__).resolve().parent.parent
+class AgentProvider(ABC):
+    """Per-template plugin: produces the provision plan and applies
+    the provider-specific in-guest setup steps (skills, prompt, the
+    declarative `dirs`/`files`/`pre_copy`/`verify` apply loop, and
+    supervise MCP registration). Concrete subclasses live under
+    `bot_bottle/contrib/<template>/agent_provider.py`."""
+
+    @property
+    @abstractmethod
+    def runtime(self) -> AgentProviderRuntime:
+        """The static command / image / prompt-mode table for this
+        template."""
+
+    @abstractmethod
+    def provision_plan(
+        self,
+        *,
+        dockerfile: str,
+        state_dir: Path,
+        guest_home: str = GUEST_HOME,
+        guest_env: dict[str, str] | None = None,
+        auth_token: str = "",
+        forward_host_credentials: bool = False,
+        host_env: dict[str, str] | None = None,
+        trusted_project_path: str = "",
+    ) -> AgentProvisionPlan:
+        """Build the declarative AgentProvisionPlan for one launch.
+        Backends call this during `prepare` and consume the result as
+        before."""
+
+    def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None:
+        """Copy each of the agent's named skills from
+        ~/.claude/skills/<name>/ on the host into
+        /home/node/.claude/skills/<name>/ in the guest. No-op when
+        the agent has no skills.
+
+        Default implementation matches the legacy backend-side
+        modules; providers override only if the in-guest layout
+        differs."""
+        from . import _provision_apply
+        _provision_apply.apply_skills(plan, bottle)
+
+    def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
+        """Copy the prompt file into the guest, fix ownership/mode,
+        and return the in-guest path iff the agent has a non-empty
+        prompt (drives the `--append-system-prompt-file` flag).
+
+        The file is copied either way so the path always exists."""
+        from . import _provision_apply
+        return _provision_apply.apply_prompt(plan, bottle)
+
+    def provision(self, plan: "BottlePlan", bottle: "Bottle") -> None:
+        """Apply the declarative `dirs`/`pre_copy`/`files`/`verify`
+        steps from `plan.agent_provision`. Default implementation
+        works for any provider that produces a standard plan; was
+        called `provision_provider_auth` on `BottleBackend` before
+        PRD 0050."""
+        from . import _provision_apply
+        _provision_apply.apply_provision(plan, bottle)
+
+    @abstractmethod
+    def provision_supervise_mcp(
+        self,
+        plan: "BottlePlan",
+        bottle: "Bottle",
+        supervise_url: str,
+    ) -> None:
+        """Register the per-bottle supervise sidecar as an MCP server
+        in the provider's in-guest config. Called by the backend after
+        the supervise sidecar is reachable. No-op when
+        `plan.supervise_plan is None`."""
 
 
-_RUNTIMES = {
+def get_provider(template: str) -> AgentProvider:
-    PROVIDER_CLAUDE: AgentProviderRuntime(
+    """Resolve a provider template name to its plugin instance.
-        template=PROVIDER_CLAUDE,
+
-        command="claude",
+    Lazy-imports the contrib module so importing this module doesn't
-        image="bot-bottle-claude:latest",
+    pull provider-specific code paths in. Mirrors the contrib
-        dockerfile=str(_REPO_ROOT / "Dockerfile.claude"),
+    convention PRD 0048 established for deploy key provisioners."""
-        prompt_mode="append_file",
+    if template == PROVIDER_CLAUDE:
-        bypass_args=("--dangerously-skip-permissions",),
+        from .contrib.claude.agent_provider import ClaudeAgentProvider
-        resume_args=("--continue",),
+        return ClaudeAgentProvider()
-        remote_control_args=("--remote-control",),
+    if template == PROVIDER_CODEX:
-    ),
+        from .contrib.codex.agent_provider import CodexAgentProvider
-    PROVIDER_CODEX: AgentProviderRuntime(
+        return CodexAgentProvider()
-        template=PROVIDER_CODEX,
+    raise ValueError(f"unknown agent provider template: {template!r}")
-        command="codex",
-        image="bot-bottle-codex:latest",
-        dockerfile=str(_REPO_ROOT / "Dockerfile.codex"),
-        prompt_mode="read_prompt_file",
-        bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
-        resume_args=("resume", "--last"),
-        remote_control_args=(),
-    ),
-}
 
 
 def runtime_for(template: str) -> AgentProviderRuntime:
-    return _RUNTIMES[template]
+    return get_provider(template).runtime
 
 
 def agent_provision_plan(
@@ -132,118 +209,24 @@ def agent_provision_plan(
     template: str,
     dockerfile: str,
     state_dir: Path,
-    guest_home: str = "/home/node",
+    guest_home: str = GUEST_HOME,
     guest_env: dict[str, str] | None = None,
     auth_token: str = "",
     forward_host_credentials: bool = False,
     host_env: dict[str, str] | None = None,
     trusted_project_path: str = "",
 ) -> AgentProvisionPlan:
-    runtime = runtime_for(template)
+    """Back-compat shim — `prepare` callers stay the same; the work
-    resolved_guest_env = dict(guest_env or {})
+    now lives on the provider plugin."""
-    trusted_path = trusted_project_path or guest_home
+    return get_provider(template).provision_plan(
-    env_vars: dict[str, str] = {}
-    provisioned_env: dict[str, str] = {}
-    dirs: list[AgentProvisionDir] = []
-    files: list[AgentProvisionFile] = []
-    pre_copy: list[AgentProvisionCommand] = []
-    verify: list[AgentProvisionCommand] = []
-    egress_routes: list[EgressRoute] = []
-    hidden_env_names: frozenset[str] = frozenset()
-
-    if template == PROVIDER_CODEX:
-        env_vars["CODEX_CA_CERTIFICATE"] = "/etc/ssl/certs/ca-certificates.crt"
-        auth_dir = resolved_guest_env.get("CODEX_HOME", f"{guest_home}/.codex")
-        if forward_host_credentials:
-            env_vars["CODEX_HOME"] = auth_dir
-        dirs.append(AgentProvisionDir(auth_dir))
-        config_path = f"{auth_dir}/config.toml"
-        config_file = state_dir / "codex-config.toml"
-        toml_path = trusted_path.replace("\\", "\\\\").replace('"', '\\"')
-        config_file.write_text(
-            f'[projects."{toml_path}"]\n'
-            'trust_level = "trusted"\n'
-        )
-        config_file.chmod(0o600)
-        files.append(AgentProvisionFile(config_file, config_path))
-
-        for host in CODEX_HOST_CREDENTIAL_HOSTS:
-            egress_routes.append(EgressRoute(
-                host=host,
-                auth_scheme="Bearer" if forward_host_credentials else "",
-                token_ref=CODEX_HOST_CREDENTIAL_TOKEN_REF if forward_host_credentials else "",
-                tls_passthrough=True,
-            ))
-        if forward_host_credentials:
-            _host_env = host_env or dict(os.environ)
-            provisioned_env[CODEX_HOST_CREDENTIAL_TOKEN_REF] = codex_host_access_token(
-                _host_env,
-            )
-            auth_file = state_dir / "codex-auth.json"
-            write_codex_dummy_auth_file(auth_file, _host_env)
-            files.append(AgentProvisionFile(auth_file, f"{auth_dir}/auth.json"))
-            pre_copy.append(AgentProvisionCommand((
-                "find", auth_dir,
-                "-maxdepth", "1",
-                "-type", "f",
-                "(",
-                "-name", "*.sqlite",
-                "-o", "-name", "*.sqlite-*",
-                "-o", "-name", "*.codex-repair-*.bak",
-                ")",
-                "-delete",
-            ), "codex host credentials: could not reset runtime db files"))
-            verify.append(AgentProvisionCommand((
-                "runuser", "-u", "node", "--",
-                "env",
-                f"HOME={guest_home}",
-                f"CODEX_HOME={auth_dir}",
-                "codex", "login", "status",
-            ), (
-                "codex host credentials: dummy auth was copied into the "
-                "guest, but Codex did not accept it"
-            )))
-    if template == PROVIDER_CLAUDE:
-        env_vars["CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC"] = "1"
-        env_vars["DISABLE_ERROR_REPORTING"] = "1"
-        claude_config = state_dir / "claude.json"
-        claude_projects = {
-            guest_home: {"hasTrustDialogAccepted": True},
-        }
-        claude_projects[trusted_path] = {"hasTrustDialogAccepted": True}
-        claude_config.write_text(json.dumps({
-            "hasCompletedOnboarding": True,
-            "theme": "dark",
-            "bypassPermissionsModeAccepted": True,
-            "projects": claude_projects,
-        }, indent=2) + "\n")
-        claude_config.chmod(0o600)
-        files.append(AgentProvisionFile(claude_config, f"{guest_home}/.claude.json"))
-        egress_routes.append(EgressRoute(
-            host="api.anthropic.com",
-            auth_scheme="Bearer" if auth_token else "",
-            token_ref=auth_token,
-            tls_passthrough=True,
-        ))
-        if auth_token:
-            env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
-            hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
-
-    return AgentProvisionPlan(
-        template=template,
-        command=runtime.command,
-        prompt_mode=runtime.prompt_mode,
-        image=runtime.image,
         dockerfile=dockerfile,
-        env_vars=env_vars,
+        state_dir=state_dir,
-        guest_env=resolved_guest_env,
+        guest_home=guest_home,
-        dirs=tuple(dirs),
+        guest_env=guest_env,
-        files=tuple(files),
+        auth_token=auth_token,
-        pre_copy=tuple(pre_copy),
+        forward_host_credentials=forward_host_credentials,
-        verify=tuple(verify),
+        host_env=host_env,
-        egress_routes=tuple(egress_routes),
+        trusted_project_path=trusted_project_path,
-        hidden_env_names=hidden_env_names,
-        provisioned_env=provisioned_env,
     )
 
 

bot_bottle/contrib/claude/agent_provider.py

@@ -0,0 +1,133 @@
+"""Claude agent provider plugin (PRD 0050, contrib).
+
+The Claude-specific behavior previously inlined under
+`agent_provider.agent_provision_plan` (claude.json trust marker,
+api.anthropic.com egress route, OAuth-token placeholder), plus
+the `claude mcp add` invocation that registers the supervise
+sidecar in claude-code's user config (PRD 0013)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from ...agent_provider import (
+    GUEST_HOME,
+    AgentProvider,
+    AgentProviderRuntime,
+    AgentProvisionFile,
+    AgentProvisionPlan,
+)
+from ...egress import EgressRoute
+from ...log import info, warn
+
+
+if TYPE_CHECKING:
+    from ...backend import Bottle, BottlePlan
+
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+
+_SUPERVISE_MCP_NAME = "supervise"
+
+_RUNTIME = AgentProviderRuntime(
+    template="claude",
+    command="claude",
+    image="bot-bottle-claude:latest",
+    dockerfile=str(_REPO_ROOT / "Dockerfile.claude"),
+    prompt_mode="append_file",
+    bypass_args=("--dangerously-skip-permissions",),
+    resume_args=("--continue",),
+    remote_control_args=("--remote-control",),
+)
+
+
+class ClaudeAgentProvider(AgentProvider):
+    @property
+    def runtime(self) -> AgentProviderRuntime:
+        return _RUNTIME
+
+    def provision_plan(
+        self,
+        *,
+        dockerfile: str,
+        state_dir: Path,
+        guest_home: str = GUEST_HOME,
+        guest_env: dict[str, str] | None = None,
+        auth_token: str = "",
+        forward_host_credentials: bool = False,
+        host_env: dict[str, str] | None = None,
+        trusted_project_path: str = "",
+    ) -> AgentProvisionPlan:
+        del forward_host_credentials, host_env  # Codex-only knobs
+        resolved_guest_env = dict(guest_env or {})
+        trusted_path = trusted_project_path or guest_home
+
+        env_vars: dict[str, str] = {
+            "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
+            "DISABLE_ERROR_REPORTING": "1",
+        }
+        claude_config = state_dir / "claude.json"
+        claude_projects = {guest_home: {"hasTrustDialogAccepted": True}}
+        claude_projects[trusted_path] = {"hasTrustDialogAccepted": True}
+        claude_config.write_text(json.dumps({
+            "hasCompletedOnboarding": True,
+            "theme": "dark",
+            "bypassPermissionsModeAccepted": True,
+            "projects": claude_projects,
+        }, indent=2) + "\n")
+        claude_config.chmod(0o600)
+        files = (
+            AgentProvisionFile(claude_config, f"{guest_home}/.claude.json"),
+        )
+        egress_routes = (EgressRoute(
+            host="api.anthropic.com",
+            auth_scheme="Bearer" if auth_token else "",
+            token_ref=auth_token,
+            tls_passthrough=True,
+        ),)
+        hidden_env_names: frozenset[str] = frozenset()
+        if auth_token:
+            env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
+            hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
+
+        return AgentProvisionPlan(
+            template=_RUNTIME.template,
+            command=_RUNTIME.command,
+            prompt_mode=_RUNTIME.prompt_mode,
+            image=_RUNTIME.image,
+            dockerfile=dockerfile,
+            env_vars=env_vars,
+            guest_env=resolved_guest_env,
+            files=files,
+            egress_routes=egress_routes,
+            hidden_env_names=hidden_env_names,
+        )
+
+    def provision_supervise_mcp(
+        self,
+        plan: "BottlePlan",
+        bottle: "Bottle",
+        supervise_url: str,
+    ) -> None:
+        """Run `claude mcp add` inside the agent guest to register the
+        supervise sidecar in claude-code's user config (~/.claude.json).
+
+        Failure is logged but not fatal — the bottle still works without
+        the entry; the operator can register it manually."""
+        if plan.supervise_plan is None:
+            return
+        info(f"registering supervise MCP server in agent claude config → {supervise_url}")
+        r = bottle.exec(
+            f"claude mcp add --scope user --transport http "
+            f"{_SUPERVISE_MCP_NAME} {supervise_url}",
+            user="node",
+        )
+        if r.returncode != 0:
+            warn(
+                f"`claude mcp add supervise` failed (exit {r.returncode}): "
+                f"{(r.stderr or r.stdout or '').strip()}. Inside the bottle, "
+                f"register manually with: "
+                f"claude mcp add --scope user --transport http supervise {supervise_url}"
+            )

bot_bottle/contrib/codex/agent_provider.py

@@ -0,0 +1,175 @@
+"""Codex agent provider plugin (PRD 0050, contrib).
+
+The Codex-specific behavior previously inlined under
+`agent_provider.agent_provision_plan` (config.toml trust marker,
+chatgpt.com / api.openai.com egress routes, optional host-credential
+forwarding with dummy-auth.json + verify), plus the `codex mcp add`
+invocation that registers the supervise sidecar in Codex's
+~/.codex/config.toml (PRD 0050)."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from ...agent_provider import (
+    CODEX_HOST_CREDENTIAL_HOSTS,
+    GUEST_HOME,
+    AgentProvider,
+    AgentProviderRuntime,
+    AgentProvisionCommand,
+    AgentProvisionDir,
+    AgentProvisionFile,
+    AgentProvisionPlan,
+)
+from ...codex_auth import codex_host_access_token, write_codex_dummy_auth_file
+from ...egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
+from ...log import info, warn
+
+
+if TYPE_CHECKING:
+    from ...backend import Bottle, BottlePlan
+
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+
+_SUPERVISE_MCP_NAME = "supervise"
+
+_RUNTIME = AgentProviderRuntime(
+    template="codex",
+    command="codex",
+    image="bot-bottle-codex:latest",
+    dockerfile=str(_REPO_ROOT / "Dockerfile.codex"),
+    prompt_mode="read_prompt_file",
+    bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
+    resume_args=("resume", "--last"),
+    remote_control_args=(),
+)
+
+
+class CodexAgentProvider(AgentProvider):
+    @property
+    def runtime(self) -> AgentProviderRuntime:
+        return _RUNTIME
+
+    def provision_plan(
+        self,
+        *,
+        dockerfile: str,
+        state_dir: Path,
+        guest_home: str = GUEST_HOME,
+        guest_env: dict[str, str] | None = None,
+        auth_token: str = "",
+        forward_host_credentials: bool = False,
+        host_env: dict[str, str] | None = None,
+        trusted_project_path: str = "",
+    ) -> AgentProvisionPlan:
+        del auth_token  # Claude-only knob
+        resolved_guest_env = dict(guest_env or {})
+        trusted_path = trusted_project_path or guest_home
+
+        env_vars: dict[str, str] = {
+            "CODEX_CA_CERTIFICATE": "/etc/ssl/certs/ca-certificates.crt",
+        }
+        auth_dir = resolved_guest_env.get("CODEX_HOME", f"{guest_home}/.codex")
+        if forward_host_credentials:
+            env_vars["CODEX_HOME"] = auth_dir
+
+        dirs = [AgentProvisionDir(auth_dir)]
+        files: list[AgentProvisionFile] = []
+        pre_copy: list[AgentProvisionCommand] = []
+        verify: list[AgentProvisionCommand] = []
+        provisioned_env: dict[str, str] = {}
+
+        config_path = f"{auth_dir}/config.toml"
+        config_file = state_dir / "codex-config.toml"
+        toml_path = trusted_path.replace("\\", "\\\\").replace('"', '\\"')
+        config_file.write_text(
+            f'[projects."{toml_path}"]\n'
+            'trust_level = "trusted"\n'
+        )
+        config_file.chmod(0o600)
+        files.append(AgentProvisionFile(config_file, config_path))
+
+        egress_routes: list[EgressRoute] = []
+        for host in CODEX_HOST_CREDENTIAL_HOSTS:
+            egress_routes.append(EgressRoute(
+                host=host,
+                auth_scheme="Bearer" if forward_host_credentials else "",
+                token_ref=CODEX_HOST_CREDENTIAL_TOKEN_REF if forward_host_credentials else "",
+                tls_passthrough=True,
+            ))
+
+        if forward_host_credentials:
+            _host_env = host_env or dict(os.environ)
+            provisioned_env[CODEX_HOST_CREDENTIAL_TOKEN_REF] = (
+                codex_host_access_token(_host_env)
+            )
+            auth_file = state_dir / "codex-auth.json"
+            write_codex_dummy_auth_file(auth_file, _host_env)
+            files.append(AgentProvisionFile(auth_file, f"{auth_dir}/auth.json"))
+            pre_copy.append(AgentProvisionCommand((
+                "find", auth_dir,
+                "-maxdepth", "1",
+                "-type", "f",
+                "(",
+                "-name", "*.sqlite",
+                "-o", "-name", "*.sqlite-*",
+                "-o", "-name", "*.codex-repair-*.bak",
+                ")",
+                "-delete",
+            ), "codex host credentials: could not reset runtime db files"))
+            verify.append(AgentProvisionCommand((
+                "runuser", "-u", "node", "--",
+                "env",
+                f"HOME={guest_home}",
+                f"CODEX_HOME={auth_dir}",
+                "codex", "login", "status",
+            ), (
+                "codex host credentials: dummy auth was copied into the "
+                "guest, but Codex did not accept it"
+            )))
+
+        return AgentProvisionPlan(
+            template=_RUNTIME.template,
+            command=_RUNTIME.command,
+            prompt_mode=_RUNTIME.prompt_mode,
+            image=_RUNTIME.image,
+            dockerfile=dockerfile,
+            env_vars=env_vars,
+            guest_env=resolved_guest_env,
+            dirs=tuple(dirs),
+            files=tuple(files),
+            pre_copy=tuple(pre_copy),
+            verify=tuple(verify),
+            egress_routes=tuple(egress_routes),
+            provisioned_env=provisioned_env,
+        )
+
+    def provision_supervise_mcp(
+        self,
+        plan: "BottlePlan",
+        bottle: "Bottle",
+        supervise_url: str,
+    ) -> None:
+        """Run `codex mcp add` inside the agent guest to register the
+        supervise sidecar in Codex's user config (~/.codex/config.toml).
+
+        Mirrors the Claude provider's `claude mcp add` flow — failure
+        is logged but not fatal."""
+        if plan.supervise_plan is None:
+            return
+        info(f"registering supervise MCP server in agent codex config → {supervise_url}")
+        r = bottle.exec(
+            f"codex mcp add --transport http "
+            f"{_SUPERVISE_MCP_NAME} {supervise_url}",
+            user="node",
+        )
+        if r.returncode != 0:
+            warn(
+                f"`codex mcp add supervise` failed (exit {r.returncode}): "
+                f"{(r.stderr or r.stdout or '').strip()}. Inside the bottle, "
+                f"register manually with: "
+                f"codex mcp add --transport http supervise {supervise_url}"
+            )