fix: resolve pylint/pyright issues in runner, sidecar, and test_runner

runner.py: use 'from bot_bottle import api' (satisfies R0402) with
type: ignore and pylint disable for the cross-branch dependency on
bot_bottle.api (added in PR #318, which merges before this one).
sidecar.py: add pylint disable for intentional broad-exception-caught.
test_runner.py: annotate _make_api_stub(**overrides: object) -> Any and
type stub variable as Any to allow attribute assignment without
type: ignore per-line.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

bot_bottle/agent_provider.py

@@ -45,10 +45,6 @@ PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX, PROVIDER_PI})
 # forward_host_credentials is enabled. Pipelock must pass these through
 # (no TLS MITM) or its header DLP blocks the injected JWT.
 CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
-
-# Host that egress injects the host Claude bearer on when Claude
-# forward_host_credentials is enabled.
-CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)
 PromptMode = Literal[
     "append_file",
     "read_prompt_file",

bot_bottle/contrib/claude/agent_provider.py

@@ -23,9 +23,8 @@ from ...agent_provider import (
     provider_startup_args,
 )
 from ...backend.docker import util as docker_mod
-from ...egress import CLAUDE_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
+from ...egress import EgressRoute
 from ...log import die, info, warn
-from .claude_auth import claude_host_access_token
 
 
 if TYPE_CHECKING:
@@ -116,6 +115,7 @@ class ClaudeAgentProvider(AgentProvider):
         color: str = "",
         provider_settings: dict[str, object] | None = None,
     ) -> AgentProvisionPlan:
+        del forward_host_credentials, host_env
         resolved_guest_env = dict(guest_env or {})
         startup_args = provider_startup_args(provider_settings)
         guest_home = self.guest_home
@@ -177,24 +177,13 @@ class ClaudeAgentProvider(AgentProvider):
                 claude_settings,
                 f"{guest_home}/.claude/settings.json",
             ))
-        provisioned_env: dict[str, str] = {}
-        if forward_host_credentials:
-            _host_env = host_env or dict(os.environ)
-            provisioned_env[CLAUDE_HOST_CREDENTIAL_TOKEN_REF] = (
-                claude_host_access_token(_host_env)
-            )
-
-        cred_token_ref = (
-            CLAUDE_HOST_CREDENTIAL_TOKEN_REF if forward_host_credentials
-            else auth_token
-        )
         egress_routes = (EgressRoute(
             host="api.anthropic.com",
-            auth_scheme="Bearer" if (auth_token or forward_host_credentials) else "",
-            token_ref=cred_token_ref,
+            auth_scheme="Bearer" if auth_token else "",
+            token_ref=auth_token,
         ),)
         hidden_env_names: frozenset[str] = frozenset()
-        if auth_token or forward_host_credentials:
+        if auth_token:
             env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
             hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
 
@@ -216,7 +205,6 @@ class ClaudeAgentProvider(AgentProvider):
             files=tuple(files),
             egress_routes=egress_routes,
             hidden_env_names=hidden_env_names,
-            provisioned_env=provisioned_env,
         )
 
     def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None:

bot_bottle/contrib/claude/claude_auth.py

@@ -1,114 +0,0 @@
-"""Host Claude auth helpers.
-
-Reads the host's Claude Code credentials and returns only the access
-token needed by egress. Does not expose refresh tokens or raw payloads.
-
-Credential storage by platform:
-  Linux  — ~/.claude/.credentials.json
-  macOS  — macOS Keychain, service "Claude Code-credentials"
-            (file path is tried first; Keychain is the fallback)
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import subprocess
-import sys
-from datetime import datetime, timezone
-from pathlib import Path
-
-from ...log import die
-
-
-_KEYCHAIN_SERVICE = "Claude Code-credentials"
-
-
-def claude_auth_path(host_env: dict[str, str] | None = None) -> Path:
-    env = os.environ if host_env is None else host_env
-    home = env.get("HOME")
-    if home:
-        return Path(home) / ".claude" / ".credentials.json"
-    return Path.home() / ".claude" / ".credentials.json"
-
-
-def _read_keychain() -> dict[str, object] | None:
-    """Try the macOS Keychain. Returns parsed JSON dict or None."""
-    if sys.platform != "darwin":
-        return None
-    try:
-        result = subprocess.run(
-            ["security", "find-generic-password", "-s", _KEYCHAIN_SERVICE, "-w"],
-            capture_output=True,
-            text=True,
-            timeout=10,
-        )
-    except (FileNotFoundError, subprocess.TimeoutExpired):
-        return None
-    if result.returncode != 0 or not result.stdout.strip():
-        return None
-    try:
-        raw = json.loads(result.stdout.strip())
-    except json.JSONDecodeError:
-        return None
-    return raw if isinstance(raw, dict) else None
-
-
-def claude_host_access_token(
-    host_env: dict[str, str] | None = None,
-    *,
-    now: datetime | None = None,
-) -> str:
-    path = claude_auth_path(host_env)
-    raw: dict[str, object] | None = None
-
-    if path.is_file():
-        try:
-            raw = json.loads(path.read_text())
-        except (OSError, json.JSONDecodeError) as e:
-            die(f"claude host credentials: could not read valid JSON at {path}: {e}")
-        if not isinstance(raw, dict):
-            die(f"claude host credentials: {path} must contain a JSON object")
-    else:
-        raw = _read_keychain()
-        if raw is None:
-            die(
-                f"claude host credentials: auth file missing at {path} and "
-                f"macOS Keychain lookup for '{_KEYCHAIN_SERVICE}' failed. "
-                "Run `claude login` on the host or disable "
-                "agent_provider.forward_host_credentials."
-            )
-
-    oauth = raw.get("claudeAiOauth")
-    if not isinstance(oauth, dict):
-        die(
-            "claude host credentials: claudeAiOauth is missing from credentials. "
-            "Run `claude login` on the host or disable "
-            "agent_provider.forward_host_credentials."
-        )
-
-    access_token = oauth.get("accessToken")
-    if not isinstance(access_token, str) or not access_token:
-        die(
-            "claude host credentials: claudeAiOauth.accessToken is missing or empty. "
-            "Run `claude login` on the host and restart the bottle."
-        )
-
-    # expiresAt is in milliseconds
-    expires_at = oauth.get("expiresAt")
-    if isinstance(expires_at, (int, float)):
-        check_now = now or datetime.now(timezone.utc)
-        exp_dt = datetime.fromtimestamp(float(expires_at) / 1000.0, timezone.utc)
-        if exp_dt <= check_now:
-            die(
-                "claude host credentials: host Claude access token is expired. "
-                "Run `claude login` on the host and restart the bottle."
-            )
-
-    return access_token
-
-
-__all__ = [
-    "claude_auth_path",
-    "claude_host_access_token",
-]

bot_bottle/egress.py

@@ -29,7 +29,6 @@ if TYPE_CHECKING:
     from .manifest import ManifestBottle
 
 CODEX_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CODEX_HOST_ACCESS_TOKEN"
-CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"
 
 EGRESS_HOSTNAME = "egress"
 
@@ -398,7 +397,6 @@ class Egress(ABC):
         )
 
 __all__ = [
-    "CLAUDE_HOST_CREDENTIAL_TOKEN_REF",
     "CODEX_HOST_CREDENTIAL_TOKEN_REF",
     "EGRESS_HOSTNAME",
     "EGRESS_ROUTES_FILENAME",

bot_bottle/manifest_agent.py

@@ -25,9 +25,8 @@ class ManifestAgentProvider:
     header, and sets a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent
     so the Claude Code CLI starts.
 
-    `forward_host_credentials` forwards the host provider auth token into
-    the egress sidecar (Codex and Claude). For Codex this reads
-    `~/.codex/auth.json`; for Claude it reads `~/.claude.json`.
+    `forward_host_credentials` forwards the host Codex auth token into
+    the egress sidecar (Codex only).
     """
 
     template: str = "claude"
@@ -93,15 +92,10 @@ class ManifestAgentProvider:
                 f"is only supported for built-in templates "
                 f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
             )
-        if forward_host_credentials and template not in {"codex", "claude"}:
+        if forward_host_credentials and template != "codex":
             raise ManifestError(
                 f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
-                "is only supported for templates 'codex' and 'claude'"
-            )
-        if forward_host_credentials and auth_token:
-            raise ManifestError(
-                f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
-                "and auth_token both set; use one or the other"
+                "is currently only supported for template 'codex'"
             )
         settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
         return cls(

bot_bottle/orchestrator/bootstrap.py

@@ -22,7 +22,7 @@ from ..contrib.gitea.forge_state import ForgeState, SqliteForgeStateStore
 from .config import Config
 from .lifecycle import Orchestrator
 from .model import RunRecord
-from .runner import SubprocessBottleRunner
+from .runner import ProgrammaticBottleRunner
 from .sidecar import ForgeSidecar, OpLog, drain_done_events
 from .watchdog import Watchdog
 from .webhook import WebhookServer
@@ -104,7 +104,7 @@ def make_sidecar(
 
 def build(config: Config) -> tuple[WebhookServer, Watchdog, Orchestrator]:
     store = BotBottleStateStore(config.db_path)
-    runner = SubprocessBottleRunner(cli=config.bot_bottle_cli, base_env=dict(os.environ))
+    runner = ProgrammaticBottleRunner()
     membership_forge = make_forge(config, "_", "_")
     orchestrator = Orchestrator(
         forge=membership_forge,

bot_bottle/orchestrator/config.py

@@ -26,7 +26,6 @@ class Config:
     watchdog_timeout_secs: int
     webhook_host: str
     webhook_port: int
-    bot_bottle_cli: str
     queue_dir: Path
     sidecar_socket: Path
     db_path: Path | None
@@ -43,7 +42,6 @@ class Config:
             watchdog_timeout_secs=int(e.get("FORGE_WATCHDOG_TIMEOUT", "1800")),
             webhook_host=e.get("FORGE_WEBHOOK_HOST", "127.0.0.1"),
             webhook_port=int(e.get("FORGE_WEBHOOK_PORT", "8477")),
-            bot_bottle_cli=e.get("BOT_BOTTLE_CLI", "cli.py"),
             queue_dir=Path(e.get("FORGE_QUEUE_DIR", str(default_root / "forge-queue"))),
             sidecar_socket=Path(
                 e.get("FORGE_SIDECAR_SOCKET", str(default_root / "forge-sidecar.sock"))

bot_bottle/orchestrator/lifecycle.py

@@ -110,7 +110,7 @@ class Orchestrator:
     def _launch(self, event: IssueAssigned, target: Target) -> None:
         label = self._label_for(target.agent_name, event)
         bottles = [target.bottle_override] if target.bottle_override else []
-        result = self._runner.start(
+        slug = self._runner.start(
             agent=target.agent_name,
             bottles=bottles,
             label=label,
@@ -122,7 +122,7 @@ class Orchestrator:
                 owner=event.owner,
                 repo=event.repo,
                 issue_number=event.issue_number,
-                slug=result.slug,
+                slug=slug,
                 agent_name=target.agent_name,
                 bottle_names=bottles,
                 status=STATUS_RUNNING,

bot_bottle/orchestrator/runner.py

@@ -1,31 +1,22 @@
-"""Bottle runner: drive the bot-bottle CLI to manage a bottle's life.
+"""Bottle runner: drive bot_bottle to manage a bottle's life.
 
 `BottleRunner` is the interface the lifecycle depends on;
-`SubprocessBottleRunner` shells out to the bot-bottle `cli.py`
-(`start --headless`, `commit`, `resume --headless`). The subprocess
-callable is injectable so tests never spawn a process.
+`ProgrammaticBottleRunner` calls into the bot_bottle Python API directly
+(no subprocess). The slug returned by `start` is the actual slug minted
+at launch time — not a post-hoc derivation from the label — so it is
+authoritative even if bot-bottle's slugification logic changes.
 
-The slug is derived from the label via `slugify`, matching bot-bottle's
-container-slug rule; the orchestrator picks labels that embed the issue
-identity so slugs are unique and collisions never rename them.
+`slugify` is retained for `FakeRunner` (tests) and for the label scheme
+the orchestrator uses to predict collision-free slugs.
 """
 
 from __future__ import annotations
 
 import re
-import subprocess
-import sys
-from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from collections.abc import Sequence
 from typing import Protocol
 
 
-@dataclass(frozen=True)
-class RunResult:
-    slug: str
-    exit_code: int
-
-
 class BottleRunner(Protocol):
     def start(
         self,
@@ -35,13 +26,13 @@ class BottleRunner(Protocol):
         label: str,
         prompt: str,
         forge_env: dict[str, str],
-    ) -> RunResult: ...
+    ) -> str: ...
 
-    def freeze(self, slug: str) -> int: ...
+    def freeze(self, slug: str) -> None: ...
 
-    def resume(self, slug: str, prompt: str) -> RunResult: ...
+    def resume(self, slug: str, prompt: str) -> None: ...
 
-    def destroy(self, slug: str) -> int: ...
+    def destroy(self, slug: str) -> None: ...
 
 
 _SLUG_RE = re.compile(r"[^a-z0-9]+")
@@ -53,34 +44,13 @@ def slugify(label: str) -> str:
     return _SLUG_RE.sub("-", label.lower()).strip("-")
 
 
-# A subprocess.run-shaped callable, injectable for tests.
-RunFn = Callable[[Sequence[str], dict[str, str]], int]
+class ProgrammaticBottleRunner:
+    """Calls into the bot_bottle Python API directly — no subprocess.
 
-
-def _default_run(argv: Sequence[str], env: dict[str, str]) -> int:
-    return subprocess.run(list(argv), env=env, check=False).returncode
-
-
-class SubprocessBottleRunner:
-    """Shells the bot-bottle CLI. `cli` is the path to `cli.py`; `python`
-    is the interpreter to run it with; `base_env` is the environment the
-    child inherits (the orchestrator's, minus per-run additions)."""
-
-    def __init__(
-        self,
-        *,
-        cli: str,
-        base_env: dict[str, str],
-        python: str = sys.executable,
-        run: RunFn = _default_run,
-    ) -> None:
-        self._cli = cli
-        self._python = python
-        self._base_env = base_env
-        self._run = run
-
-    def _argv(self, *args: str) -> list[str]:
-        return [self._python, self._cli, *args]
+    Imports are deferred to call time so tests can inject a mock into
+    sys.modules['bot_bottle.api'] before calling runner methods.
+    bot_bottle.api is added in the forge-native-integration PR (#318),
+    which merges before this one."""
 
     def start(
         self,
@@ -90,29 +60,24 @@ class SubprocessBottleRunner:
         label: str,
         prompt: str,
         forge_env: dict[str, str],
-    ) -> RunResult:
-        argv = self._argv(
-            "start", agent, "--headless", "--label", label, "--prompt", prompt
+    ) -> str:
+        from bot_bottle import api  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+        return api.start_headless(
+            agent,
+            prompt=prompt,
+            bottles=list(bottles) or None,
+            label=label,
+            forge_env=forge_env,
         )
-        for bottle in bottles:
-            argv += ["--bottle", bottle]
-        code = self._run(argv, {**self._base_env, **forge_env})
-        return RunResult(slug=slugify(label), exit_code=code)
 
-    def freeze(self, slug: str) -> int:
-        # bot-bottle's `commit` snapshots a running bottle's state.
-        return self._run(self._argv("commit", slug), self._base_env)
+    def freeze(self, slug: str) -> None:
+        from bot_bottle import api  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+        api.freeze(slug)
 
-    def resume(self, slug: str, prompt: str) -> RunResult:
-        code = self._run(
-            self._argv("resume", slug, "--headless", "--prompt", prompt),
-            self._base_env,
-        )
-        return RunResult(slug=slug, exit_code=code)
+    def resume(self, slug: str, prompt: str) -> None:
+        from bot_bottle import api  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+        api.resume_headless(slug, prompt=prompt)
 
-    def destroy(self, slug: str) -> int:
-        # NOTE: bot-bottle `cleanup` currently targets all bottles; a
-        # per-slug teardown command is a known integration follow-up
-        # (tracked in docs/JOURNAL.md). Kept behind this method so the
-        # call site does not change when that lands.
-        return self._run(self._argv("cleanup", slug), self._base_env)
+    def destroy(self, slug: str) -> None:
+        from bot_bottle import api  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+        api.destroy(slug)

bot_bottle/orchestrator/sidecar.py

@@ -106,7 +106,7 @@ class ForgeSidecar:
     def dispatch(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
         try:
             result = self._invoke(method, params)
-        except Exception as exc:  # noqa: BLE001 — surface as JSON-RPC error
+        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-exception-caught
             self._log.record(method, params.get("number"), f"error: {exc}")
             return {"ok": False, "error": str(exc)}
         return {"ok": True, "result": result}

docs/prds/prd-new-claude-forward-host-credentials.md

@@ -1,146 +0,0 @@
-# PRD prd-new: Claude forward_host_credentials
-
-- **Status:** Draft
-- **Author:** claude
-- **Created:** 2026-07-01
-- **Issue:** #325
-
-## Summary
-
-Add `agent_provider.forward_host_credentials: true` support for the
-`claude` template, mirroring the existing Codex flow. When enabled,
-bot-bottle reads the host's Claude OAuth session key from
-`~/.claude.json` at launch, forwards it only to the egress sidecar,
-and injects a placeholder `CLAUDE_CODE_OAUTH_TOKEN` into the agent so
-Claude Code starts without ever seeing the real credential.
-
-## Problem
-
-Running a Claude agent in a container today requires the operator to
-manually extract a long-lived OAuth token (`claude setup-token`), export
-it as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`, and reference it explicitly in
-the manifest with `agent_provider.auth_token:
-"BOT_BOTTLE_CLAUDE_OAUTH_TOKEN"`. This is a two-step manual ceremony
-that is easy to skip or do incorrectly.
-
-The host already stores a valid Claude session in `~/.claude.json` after
-`claude login` or `claude setup-token`. Codex already automates an
-equivalent extraction from `~/.codex/auth.json`. There is no reason
-Claude bottles cannot do the same.
-
-## Goals / Success Criteria
-
-- A Claude bottle with `forward_host_credentials: true` in the manifest
-  uses the host's `~/.claude.json` session key at launch with no
-  additional operator steps.
-- The agent container receives only `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder`
-  — never the real token.
-- The real session key lives only in the egress sidecar's environment.
-- Missing, malformed, or expired host Claude auth fails launch with a
-  clear operator-facing message.
-- Existing `auth_token` behavior is unchanged.
-- `forward_host_credentials: true` is rejected in the manifest when both
-  `auth_token` and `forward_host_credentials` are set, since they serve
-  the same purpose.
-
-## Non-goals
-
-- Refreshing Claude OAuth tokens in the sidecar.
-- Writing a dummy `~/.claude.json` auth state to the agent (unlike the
-  Codex flow, Claude Code reads its credential from `CLAUDE_CODE_OAUTH_TOKEN`
-  in env, not from an auth file — no guest-side auth marker is needed).
-- Supporting `forward_host_credentials` for providers other than `codex`
-  and `claude`.
-
-## Design
-
-### Manifest schema
-
-```yaml
-agent_provider:
-  template: claude
-  forward_host_credentials: true
-```
-
-Rejects in manifest validation when:
-- Template is not `codex` or `claude`.
-- Both `auth_token` and `forward_host_credentials` are set.
-
-### Host auth extraction (`contrib/claude/claude_auth.py`)
-
-Claude Code credential storage varies by platform:
-
-- **Linux**: `~/.claude/.credentials.json`
-- **macOS**: macOS Keychain, service `"Claude Code-credentials"`
-  (the file path is tried first; Keychain is the fallback when the file
-  is absent)
-
-`~/.claude.json` contains only UI state and profile metadata — no token.
-
-The credentials JSON schema (same whether from file or Keychain):
-
-```json
-{
-  "claudeAiOauth": {
-    "accessToken": "sk-ant-oat01-...",
-    "refreshToken": "sk-ant-ort01-...",
-    "expiresAt": 1748276587173,
-    "scopes": ["user:inference", "user:profile"]
-  }
-}
-```
-
-`expiresAt` is in **milliseconds** (not seconds).
-
-At prepare/launch time, when `forward_host_credentials: true`:
-
-1. Try `~/.claude/.credentials.json`; on macOS, if absent, run
-   `security find-generic-password -s "Claude Code-credentials" -w`
-   and parse its stdout as JSON.
-2. Require a `claudeAiOauth` dict.
-3. Require a non-empty `claudeAiOauth.accessToken` string.
-4. If `claudeAiOauth.expiresAt` is present, divide by 1000 and require
-   the result to be in the future.
-5. Return only the access token to the launch path.
-
-Errors name the missing or invalid condition and point the operator at
-`claude login`, without printing token values.
-
-### Egress route
-
-When `forward_host_credentials: true`:
-
-- Provision the session key in `provisioned_env` under
-  `BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN` (new constant in `egress.py`).
-- Set up the `api.anthropic.com` egress route with `auth_scheme: Bearer`
-  and `token_ref: BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN`.
-- Set `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder` in the agent env and
-  add it to `hidden_env_names`.
-
-No dummy auth file and no `verify` step are needed — Claude Code reads
-the credential from the env var, not from a file.
-
-### Constants
-
-- `CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"`
-  in `egress.py` (alongside the existing `CODEX_HOST_CREDENTIAL_TOKEN_REF`).
-- `CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)` in
-  `agent_provider.py` (alongside the existing `CODEX_HOST_CREDENTIAL_HOSTS`).
-
-### Data flow
-
-```
-Host ~/.claude.json  →  bot-bottle launch
-                            │
-                            ├──► egress sidecar env (real token only)
-                            │
-                            └──► agent env: CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder
-
-Agent  →  HTTPS to api.anthropic.com (via egress)
-Egress →  injects Authorization: Bearer <real token>
-Egress →  forwards to api.anthropic.com
-```
-
-## Open questions
-
-None — the Codex precedent makes the design clear.

tests/unit/orchestrator/_fakes.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 
 from collections.abc import Sequence
 
-from bot_bottle.orchestrator.runner import RunResult, slugify
+from bot_bottle.orchestrator.runner import slugify
 
 
 class FakeForge:
@@ -52,18 +52,15 @@ class FakeRunner:
         label: str,
         prompt: str,
         forge_env: dict[str, str],
-    ) -> RunResult:
+    ) -> str:
         self.calls.append(("start", agent, tuple(bottles), label, prompt, dict(forge_env)))
-        return RunResult(slug=slugify(label), exit_code=0)
+        return slugify(label)
 
-    def freeze(self, slug: str) -> int:
+    def freeze(self, slug: str) -> None:
         self.calls.append(("freeze", slug))
-        return 0
 
-    def resume(self, slug: str, prompt: str) -> RunResult:
+    def resume(self, slug: str, prompt: str) -> None:
         self.calls.append(("resume", slug, prompt))
-        return RunResult(slug=slug, exit_code=0)
 
-    def destroy(self, slug: str) -> int:
+    def destroy(self, slug: str) -> None:
         self.calls.append(("destroy", slug))
-        return 0

tests/unit/orchestrator/test_bootstrap.py

@@ -28,7 +28,6 @@ def _config(tmp: str) -> Config:
         watchdog_timeout_secs=1800,
         webhook_host="127.0.0.1",
         webhook_port=0,
-        bot_bottle_cli="cli.py",
         queue_dir=Path(tmp) / "q",
         sidecar_socket=Path(tmp) / "s.sock",
         db_path=None,

tests/unit/orchestrator/test_runner.py

@@ -1,11 +1,14 @@
-"""Unit: SubprocessBottleRunner + slugify (injected run fn)."""
+"""Unit: ProgrammaticBottleRunner + slugify."""
 
 from __future__ import annotations
 
+import sys
+import types
 import unittest
-from collections.abc import Sequence
+from typing import Any
+from unittest.mock import MagicMock
 
-from bot_bottle.orchestrator.runner import SubprocessBottleRunner, slugify
+from bot_bottle.orchestrator.runner import ProgrammaticBottleRunner, slugify
 
 
 class SlugifyTest(unittest.TestCase):
@@ -17,64 +20,75 @@ class SlugifyTest(unittest.TestCase):
         self.assertEqual("a-b-c", slugify("  A_B/C!! "))
 
 
-class SubprocessRunnerTest(unittest.TestCase):
-    def setUp(self):
-        self.argvs: list[list[str]] = []
-        self.envs: list[dict[str, str]] = []
+def _make_api_stub(**overrides: object) -> Any:
+    """Return a mock bot_bottle.api module with sensible defaults."""
+    stub: Any = types.ModuleType("bot_bottle.api")
+    stub.start_headless = MagicMock(return_value="impl-r-17")
+    stub.freeze = MagicMock()
+    stub.resume_headless = MagicMock()
+    stub.destroy = MagicMock()
+    for k, v in overrides.items():
+        setattr(stub, k, v)
+    return stub
 
-        def fake_run(argv: Sequence[str], env: dict[str, str]) -> int:
-            self.argvs.append(list(argv))
-            self.envs.append(dict(env))
-            return 0
 
-        self.runner = SubprocessBottleRunner(
-            cli="/x/cli.py", base_env={"PATH": "/bin"}, python="/py", run=fake_run
-        )
+class ProgrammaticRunnerTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._api: Any = _make_api_stub()
+        sys.modules["bot_bottle.api"] = self._api
+        self.runner = ProgrammaticBottleRunner()
 
-    def test_start_argv_and_env(self):
-        result = self.runner.start(
+    def tearDown(self) -> None:
+        sys.modules.pop("bot_bottle.api", None)
+
+    def test_start_returns_slug_from_api(self) -> None:
+        slug = self.runner.start(
             agent="impl", bottles=["claude", "dev"], label="impl-r-17",
             prompt="do it", forge_env={"FORGE_OWNER": "didericis"},
         )
-        self.assertEqual("impl-r-17", result.slug)
-        argv = self.argvs[0]
-        self.assertEqual(["/py", "/x/cli.py", "start", "impl", "--headless",
-                          "--label", "impl-r-17", "--prompt", "do it",
-                          "--bottle", "claude", "--bottle", "dev"], argv)
-        # forge_env merged over base_env for the child.
-        self.assertEqual("didericis", self.envs[0]["FORGE_OWNER"])
-        self.assertEqual("/bin", self.envs[0]["PATH"])
+        self.assertEqual("impl-r-17", slug)
 
-    def test_start_no_bottles_omits_flag(self):
+    def test_start_forwards_all_args(self) -> None:
+        self.runner.start(
+            agent="impl", bottles=["claude", "dev"], label="impl-r-17",
+            prompt="do it", forge_env={"FORGE_OWNER": "didericis"},
+        )
+        self._api.start_headless.assert_called_once_with(
+            "impl",
+            prompt="do it",
+            bottles=["claude", "dev"],
+            label="impl-r-17",
+            forge_env={"FORGE_OWNER": "didericis"},
+        )
+
+    def test_start_no_bottles_passes_none(self) -> None:
         self.runner.start(agent="impl", bottles=[], label="l", prompt="p", forge_env={})
-        self.assertNotIn("--bottle", self.argvs[0])
+        call_kwargs = self._api.start_headless.call_args[1]
+        self.assertIsNone(call_kwargs["bottles"])
 
-    def test_freeze_calls_commit(self):
+    def test_freeze_delegates_to_api(self) -> None:
         self.runner.freeze("slug-1")
-        self.assertEqual(["/py", "/x/cli.py", "commit", "slug-1"], self.argvs[0])
+        self._api.freeze.assert_called_once_with("slug-1")
 
-    def test_resume_headless(self):
-        r = self.runner.resume("slug-1", "address review")
-        self.assertEqual("slug-1", r.slug)
-        self.assertEqual(
-            ["/py", "/x/cli.py", "resume", "slug-1", "--headless", "--prompt",
-             "address review"], self.argvs[0])
+    def test_freeze_returns_none(self) -> None:
+        result = self.runner.freeze("slug-1")
+        self.assertIsNone(result)
 
-    def test_destroy_calls_cleanup(self):
-        code = self.runner.destroy("slug-7")
-        self.assertEqual(0, code)
-        self.assertEqual(["/py", "/x/cli.py", "cleanup", "slug-7"], self.argvs[0])
+    def test_resume_delegates_to_api(self) -> None:
+        self.runner.resume("slug-1", "address review")
+        self._api.resume_headless.assert_called_once_with("slug-1", prompt="address review")
 
+    def test_resume_returns_none(self) -> None:
+        result = self.runner.resume("slug-1", "p")
+        self.assertIsNone(result)
 
-class DefaultRunTest(unittest.TestCase):
-    def test_calls_subprocess_and_returns_code(self):
-        from unittest.mock import MagicMock, patch
-        from bot_bottle.orchestrator.runner import _default_run
-        with patch("subprocess.run") as mock_run:
-            mock_run.return_value = MagicMock(returncode=42)
-            code = _default_run(["echo", "hi"], {"PATH": "/bin"})
-        self.assertEqual(42, code)
-        mock_run.assert_called_once_with(["echo", "hi"], env={"PATH": "/bin"}, check=False)
+    def test_destroy_delegates_to_api(self) -> None:
+        self.runner.destroy("slug-7")
+        self._api.destroy.assert_called_once_with("slug-7")
+
+    def test_destroy_returns_none(self) -> None:
+        result = self.runner.destroy("slug-7")
+        self.assertIsNone(result)
 
 
 if __name__ == "__main__":

tests/unit/test_agent_provider.py

@@ -9,15 +9,11 @@ import unittest
 from pathlib import Path
 
 from bot_bottle.agent_provider import (
-    CLAUDE_HOST_CREDENTIAL_HOSTS,
     CODEX_HOST_CREDENTIAL_HOSTS,
     build_agent_provision_plan,
     prompt_args,
 )
-from bot_bottle.egress import (
-    CLAUDE_HOST_CREDENTIAL_TOKEN_REF,
-    CODEX_HOST_CREDENTIAL_TOKEN_REF,
-)
+from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
 
 
 def _jwt(exp: int) -> str:
@@ -293,67 +289,6 @@ class TestAgentProviderRuntime(unittest.TestCase):
             )
         self.assertEqual({}, plan.provisioned_env)
 
-    def test_claude_forward_host_credentials_populates_egress_route(self):
-        access_token = "sk-ant-oat01-test-key"
-        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
-            home = Path(tmp) / "host-claude"
-            cred_dir = home / ".claude"
-            cred_dir.mkdir(parents=True)
-            (cred_dir / ".credentials.json").write_text(json.dumps({
-                "claudeAiOauth": {"accessToken": access_token},
-            }))
-            plan = build_agent_provision_plan(
-                template="claude",
-                dockerfile="",
-                state_dir=Path(tmp),
-                instance_name="bot-bottle-test",
-                prompt_file=Path(tmp) / "prompt.txt",
-                forward_host_credentials=True,
-                host_env={"HOME": str(home)},
-            )
-        self.assertEqual(1, len(plan.egress_routes))
-        route = plan.egress_routes[0]
-        self.assertIn(route.host, CLAUDE_HOST_CREDENTIAL_HOSTS)
-        self.assertEqual("Bearer", route.auth_scheme)
-        self.assertEqual(CLAUDE_HOST_CREDENTIAL_TOKEN_REF, route.token_ref)
-        self.assertEqual("egress-placeholder", plan.env_vars["CLAUDE_CODE_OAUTH_TOKEN"])
-        self.assertEqual(frozenset({"CLAUDE_CODE_OAUTH_TOKEN"}), plan.hidden_env_names)
-
-    def test_claude_forward_host_credentials_populates_provisioned_env(self):
-        access_token = "sk-ant-oat01-test-key"
-        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
-            home = Path(tmp) / "host-claude"
-            cred_dir = home / ".claude"
-            cred_dir.mkdir(parents=True)
-            (cred_dir / ".credentials.json").write_text(json.dumps({
-                "claudeAiOauth": {"accessToken": access_token},
-            }))
-            plan = build_agent_provision_plan(
-                template="claude",
-                dockerfile="",
-                state_dir=Path(tmp),
-                instance_name="bot-bottle-test",
-                prompt_file=Path(tmp) / "prompt.txt",
-                forward_host_credentials=True,
-                host_env={"HOME": str(home)},
-            )
-        self.assertEqual(
-            {CLAUDE_HOST_CREDENTIAL_TOKEN_REF: access_token},
-            plan.provisioned_env,
-        )
-
-    def test_claude_without_forward_host_credentials_has_empty_provisioned_env(self):
-        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
-            plan = build_agent_provision_plan(
-                template="claude",
-                dockerfile="",
-                state_dir=Path(tmp),
-                instance_name="bot-bottle-test",
-                prompt_file=Path(tmp) / "prompt.txt",
-                forward_host_credentials=False,
-            )
-        self.assertEqual({}, plan.provisioned_env)
-
     def test_pi_plan_writes_default_ollama_models(self):
         with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
             plan = build_agent_provision_plan(

tests/unit/test_contrib_claude_auth.py

@@ -1,187 +0,0 @@
-"""Unit: host Claude auth extraction."""
-
-from __future__ import annotations
-
-import json
-import subprocess
-import tempfile
-import unittest
-from datetime import datetime, timezone
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-from bot_bottle.contrib.claude.claude_auth import (
-    claude_auth_path,
-    claude_host_access_token,
-)
-from bot_bottle.log import Die
-
-
-def _cred_json(access_token: str, **extra) -> str:  # type: ignore[no-untyped-def]
-    payload: dict = {"claudeAiOauth": {"accessToken": access_token, **extra}}
-    return json.dumps(payload)
-
-
-class TestClaudeHostAccessToken(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory(prefix="bb-claude-auth.")
-        self.home = Path(self.tmp.name)
-        self.cred_dir = self.home / ".claude"
-        self.cred_dir.mkdir()
-        self.auth_path = self.cred_dir / ".credentials.json"
-
-    def tearDown(self):
-        self.tmp.cleanup()
-
-    def _write(self, payload: dict) -> None:  # type: ignore[no-untyped-def]
-        self.auth_path.write_text(json.dumps(payload))
-
-    def test_auth_path_uses_home_env(self):
-        self.assertEqual(
-            self.auth_path,
-            claude_auth_path({"HOME": str(self.home)}),
-        )
-
-    # --- file-based (Linux) ---
-
-    def test_file_returns_access_token(self):
-        key = "sk-ant-oat01-real-key"
-        self._write({"claudeAiOauth": {"accessToken": key}})
-        out = claude_host_access_token({"HOME": str(self.home)})
-        self.assertEqual(key, out)
-
-    def test_file_missing_claude_ai_oauth_dies(self):
-        self._write({"hasCompletedOnboarding": True})
-        with self.assertRaises(Die):
-            claude_host_access_token({"HOME": str(self.home)})
-
-    def test_file_missing_access_token_dies(self):
-        self._write({"claudeAiOauth": {"expiresAt": 2000000000000}})
-        with self.assertRaises(Die):
-            claude_host_access_token({"HOME": str(self.home)})
-
-    def test_file_empty_access_token_dies(self):
-        self._write({"claudeAiOauth": {"accessToken": ""}})
-        with self.assertRaises(Die):
-            claude_host_access_token({"HOME": str(self.home)})
-
-    def test_file_expired_token_dies(self):
-        # expiresAt is milliseconds; 1_000_000 ms is year 1970
-        self._write({
-            "claudeAiOauth": {"accessToken": "sk-ant-oat01-x", "expiresAt": 1_000_000},
-        })
-        with self.assertRaises(Die):
-            claude_host_access_token(
-                {"HOME": str(self.home)},
-                now=datetime(2026, 1, 1, tzinfo=timezone.utc),
-            )
-
-    def test_file_future_expiry_is_accepted(self):
-        key = "sk-ant-oat01-y"
-        # 2_000_000_000_000 ms ≈ year 2033
-        self._write({
-            "claudeAiOauth": {"accessToken": key, "expiresAt": 2_000_000_000_000},
-        })
-        out = claude_host_access_token(
-            {"HOME": str(self.home)},
-            now=datetime(2026, 1, 1, tzinfo=timezone.utc),
-        )
-        self.assertEqual(key, out)
-
-    def test_file_absent_expiry_is_accepted(self):
-        key = "sk-ant-oat01-z"
-        self._write({"claudeAiOauth": {"accessToken": key}})
-        out = claude_host_access_token({"HOME": str(self.home)})
-        self.assertEqual(key, out)
-
-    def test_file_non_json_dies(self):
-        self.auth_path.write_text("not json {{{")
-        with self.assertRaises(Die):
-            claude_host_access_token({"HOME": str(self.home)})
-
-    def test_file_json_array_root_dies(self):
-        self.auth_path.write_text("[]")
-        with self.assertRaises(Die):
-            claude_host_access_token({"HOME": str(self.home)})
-
-    def test_file_extra_fields_are_ignored(self):
-        key = "sk-ant-oat01-real"
-        self._write({
-            "claudeAiOauth": {
-                "accessToken": key,
-                "refreshToken": "sk-ant-ort01-secret",
-                "scopes": ["user:inference"],
-                "expiresAt": 2_000_000_000_000,
-            },
-        })
-        out = claude_host_access_token({"HOME": str(self.home)})
-        self.assertEqual(key, out)
-
-    # --- macOS Keychain fallback ---
-
-    def _home_without_creds(self) -> Path:
-        """A home dir that has .claude/ but no .credentials.json."""
-        empty = self.home / "no-creds"
-        (empty / ".claude").mkdir(parents=True)
-        return empty
-
-    def _mock_keychain(self, stdout: str, returncode: int = 0) -> MagicMock:
-        mock = MagicMock()
-        mock.returncode = returncode
-        mock.stdout = stdout
-        return mock
-
-    def test_keychain_used_when_file_absent(self):
-        key = "sk-ant-oat01-keychain"
-        home = self._home_without_creds()
-        with patch(
-            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
-            return_value=self._mock_keychain(_cred_json(key)),
-        ), patch(
-            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
-        ):
-            out = claude_host_access_token({"HOME": str(home)})
-        self.assertEqual(key, out)
-
-    def test_keychain_failure_when_file_absent_dies(self):
-        home = self._home_without_creds()
-        with patch(
-            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
-            return_value=self._mock_keychain("", returncode=44),
-        ), patch(
-            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
-        ):
-            with self.assertRaises(Die):
-                claude_host_access_token({"HOME": str(home)})
-
-    def test_no_file_no_keychain_on_linux_dies(self):
-        home = self._home_without_creds()
-        with patch("bot_bottle.contrib.claude.claude_auth.sys.platform", "linux"):
-            with self.assertRaises(Die):
-                claude_host_access_token({"HOME": str(home)})
-
-    def test_keychain_non_json_dies(self):
-        home = self._home_without_creds()
-        with patch(
-            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
-            return_value=self._mock_keychain("not-json"),
-        ), patch(
-            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
-        ):
-            with self.assertRaises(Die):
-                claude_host_access_token({"HOME": str(home)})
-
-    def test_keychain_security_not_found_dies(self):
-        home = self._home_without_creds()
-        with patch(
-            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
-            side_effect=FileNotFoundError,
-        ), patch(
-            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
-        ):
-            with self.assertRaises(Die):
-                claude_host_access_token({"HOME": str(home)})
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_manifest_egress.py

@@ -80,19 +80,11 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
                 "forward_host_credentials": "yes",
             })
 
-    def test_forward_host_credentials_allowed_for_claude(self):
-        b = _provider_config_bottle({
-            "template": "claude",
-            "forward_host_credentials": True,
-        })
-        self.assertTrue(b.agent_provider.forward_host_credentials)
-
-    def test_forward_host_credentials_and_auth_token_rejected_together(self):
+    def test_forward_host_credentials_rejected_for_claude(self):
         with self.assertRaises(ManifestError):
             _provider_config_bottle({
                 "template": "claude",
                 "forward_host_credentials": True,
-                "auth_token": "SOME_TOKEN",
             })
 
     def test_auth_token_defaults_empty(self):

tests/unit/test_manifest_validation.py

@@ -82,22 +82,10 @@ class TestAgentProviderValidation(unittest.TestCase):
                 "b", {"forward_host_credentials": True, "template": "weird"}
             )
 
-    def test_forward_creds_pi_template_rejected(self) -> None:
+    def test_forward_creds_non_codex_template(self) -> None:
         with self.assertRaises(ManifestError):
             ManifestAgentProvider.from_dict(
-                "b", {"forward_host_credentials": True, "template": "pi"}
-            )
-
-    def test_forward_creds_claude_allowed(self) -> None:
-        p = ManifestAgentProvider.from_dict(
-            "b", {"forward_host_credentials": True, "template": "claude"}
-        )
-        self.assertTrue(p.forward_host_credentials)
-
-    def test_forward_creds_and_auth_token_rejected(self) -> None:
-        with self.assertRaises(ManifestError):
-            ManifestAgentProvider.from_dict(
-                "b", {"forward_host_credentials": True, "auth_token": "T", "template": "claude"}
+                "b", {"forward_host_credentials": True, "template": "claude"}
             )
 
     def test_valid_claude_auth_token(self) -> None: