feat: expose stable Python API for programmatic bottle orchestration

Add bot_bottle/api.py with four public functions the orchestrator uses:
start_headless, resume_headless, freeze, and destroy. These let a
ProgrammaticBottleRunner call directly into bot_bottle instead of
shelling out to the CLI; call sites in lifecycle.py stay unchanged.

Key changes:
- BottleSpec gains forge_env field for forge sidecar credentials
- _launch_bottle returns (slug, exit_code) instead of int so start_headless
  can return the slug to callers
- All four API functions convert Die and non-zero exits to BottleError
- 27 new unit tests; existing tests updated for the new return type

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

bot_bottle/__init__.py

@@ -1 +1,11 @@
 """bot-bottle: Python implementation of the agent container launcher."""
+
+from .api import BottleError, destroy, freeze, resume_headless, start_headless
+
+__all__ = [
+    "BottleError",
+    "destroy",
+    "freeze",
+    "resume_headless",
+    "start_headless",
+]

bot_bottle/api.py

@@ -0,0 +1,258 @@
+"""Public Python API for programmatic bottle orchestration.
+
+Stable surface for bot-bottle-orchestrator (and other Python callers) to
+drive bottles without invoking the CLI as a subprocess.  Every function
+converts ``Die`` and non-zero agent exit codes to ``BottleError`` so
+callers use exception handling rather than inspecting return values.
+
+The Protocol the orchestrator's ``BottleRunner`` targets looks like::
+
+    class BottleRunner(Protocol):
+        def start(self, agent: str, *, prompt: str, ...) -> str: ...
+        def resume(self, slug: str, *, prompt: str) -> None: ...
+        def freeze(self, slug: str) -> None: ...
+        def destroy(self, slug: str) -> None: ...
+
+A ``SubprocessBottleRunner`` calls ``./cli.py`` for each operation.  A
+``ProgrammaticBottleRunner`` calls these functions directly; the Protocol
+call sites in ``lifecycle.py`` are unchanged.
+"""
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from .backend import BottleSpec
+from .backend.freeze import CommitCancelled, get_freezer
+from .bottle_state import cleanup_state, clear_preserve_marker, read_metadata
+from .cli._common import USER_CWD
+from .cli.start import _launch_bottle, _peek_agent_bottle, _uniquify_label_headless
+from .log import Die
+from .manifest import ManifestError, ManifestIndex
+
+
+class BottleError(Exception):
+    """Raised when a bottle operation fails.
+
+    ``exit_code`` carries the agent process's exit code when the failure is
+    a non-zero agent exit; 1 for all other failure modes (missing state,
+    backend errors, etc.)."""
+
+    def __init__(self, message: str, *, exit_code: int = 1) -> None:
+        super().__init__(message)
+        self.exit_code = exit_code
+
+
+def start_headless(
+    agent_name: str,
+    *,
+    prompt: str,
+    bottles: Sequence[str] | None = None,
+    label: str | None = None,
+    color: str | None = None,
+    backend_name: str | None = None,
+    copy_cwd: bool = False,
+    forge_env: dict[str, str] | None = None,
+    user_cwd: str | None = None,
+) -> str:
+    """Launch a new bottle headlessly.  Returns the bottle slug.
+
+    ``forge_env`` is passed through to the forge sidecar (not the agent)
+    when the bottle is forge-targeted; it carries the credentials and
+    context the sidecar needs to call the forge API.
+
+    Raises ``BottleError`` on configuration errors or if the agent exits
+    non-zero.  The returned slug can be passed to ``freeze()``,
+    ``resume_headless()``, or ``destroy()`` for subsequent lifecycle
+    operations."""
+    cwd = user_cwd or USER_CWD
+    try:
+        manifest = ManifestIndex.resolve(cwd)
+        manifest.require_agent(agent_name)
+    except (Die, ManifestError) as exc:
+        raise BottleError(str(exc)) from exc
+
+    if bottles:
+        bottle_names: tuple[str, ...] = tuple(bottles)
+    else:
+        default_bottle = _peek_agent_bottle(manifest, agent_name)
+        if not default_bottle:
+            raise BottleError(
+                f"agent '{agent_name}' has no default bottle; "
+                f"pass bottles=[...]"
+            )
+        bottle_names = (default_bottle,)
+
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=agent_name,
+        copy_cwd=copy_cwd,
+        user_cwd=cwd,
+        label=_uniquify_label_headless(label or agent_name),
+        color=color or "",
+        bottle_names=bottle_names,
+        forge_env=dict(forge_env) if forge_env else {},
+    )
+    try:
+        slug, exit_code = _launch_bottle(
+            spec,
+            dry_run=False,
+            backend_name=backend_name,
+            assume_yes=True,
+            headless_prompt_text=prompt,
+        )
+    except Die as exc:
+        raise BottleError(exc.message, exit_code=exc.code) from exc
+    if exit_code != 0:
+        raise BottleError(
+            f"agent exited {exit_code} (slug={slug!r})", exit_code=exit_code
+        )
+    return slug
+
+
+def resume_headless(
+    slug: str,
+    *,
+    prompt: str,
+    backend_name: str | None = None,
+    forge_env: dict[str, str] | None = None,
+) -> None:
+    """Resume a frozen bottle headlessly with ``prompt``.
+
+    ``forge_env`` re-supplies forge context for the new session (the
+    sidecar is relaunched alongside the agent on resume).
+
+    Raises ``BottleError`` on missing state, backend errors, or non-zero
+    agent exit."""
+    metadata = read_metadata(slug)
+    if metadata is None:
+        raise BottleError(
+            f"no state recorded for slug {slug!r}; "
+            f"check ~/.bot-bottle/state/ or call start_headless() to create a new bottle"
+        )
+
+    try:
+        manifest = ManifestIndex.resolve(metadata.cwd or USER_CWD)
+        manifest.require_agent(metadata.agent_name)
+    except (Die, ManifestError) as exc:
+        raise BottleError(str(exc)) from exc
+
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=metadata.agent_name,
+        copy_cwd=metadata.copy_cwd,
+        user_cwd=metadata.cwd or USER_CWD,
+        identity=metadata.identity,
+        bottle_names=tuple(metadata.bottle_names),
+        forge_env=dict(forge_env) if forge_env else {},
+    )
+    try:
+        _, exit_code = _launch_bottle(
+            spec,
+            dry_run=False,
+            backend_name=backend_name or metadata.backend or None,
+            assume_yes=True,
+            headless_prompt_text=prompt,
+        )
+    except Die as exc:
+        raise BottleError(exc.message, exit_code=exc.code) from exc
+    if exit_code != 0:
+        raise BottleError(
+            f"agent exited {exit_code} resuming {slug!r}", exit_code=exit_code
+        )
+
+
+def freeze(slug: str, *, backend_name: str | None = None) -> None:
+    """Freeze the named bottle to a resumable artifact.
+
+    Reads the bottle's backend from its metadata when ``backend_name`` is
+    not supplied.  Raises ``BottleError`` if the freeze fails."""
+    metadata = read_metadata(slug)
+    resolved_backend = backend_name or (metadata.backend if metadata else "") or "docker"
+    try:
+        get_freezer(resolved_backend).commit_slug(slug)
+    except CommitCancelled as exc:
+        raise BottleError(f"freeze cancelled for {slug!r}") from exc
+    except Die as exc:
+        raise BottleError(exc.message, exit_code=exc.code) from exc
+
+
+def destroy(slug: str, *, backend_name: str | None = None) -> None:
+    """Destroy the named bottle, removing all resources and state.
+
+    Brings down any running resources for ``slug``, then removes the
+    per-bottle state directory.  Idempotent: a slug with no running
+    resources or no state directory is not an error."""
+    metadata = read_metadata(slug)
+    resolved_backend = backend_name or (metadata.backend if metadata else "") or "docker"
+    try:
+        if resolved_backend == "docker":
+            _destroy_docker(slug)
+        elif resolved_backend == "smolmachines":
+            _destroy_smolmachines(slug)
+        # macos-container: the container is torn down inside the launch
+        # context manager; no persistent VM survives, so nothing extra is
+        # needed at destroy time beyond the state-dir removal below.
+    except Die as exc:
+        raise BottleError(exc.message, exit_code=exc.code) from exc
+    clear_preserve_marker(slug)
+    cleanup_state(slug)
+
+
+# --- backend-specific helpers -----------------------------------------------
+
+
+def _destroy_docker(slug: str) -> None:
+    """Best-effort ``docker compose down`` for a Docker bottle.
+
+    No-op when the compose file is absent — the project was already
+    brought down (normal for a frozen bottle) or was never created."""
+    from .backend.docker.compose import (
+        compose_down,
+        compose_file_path,
+        compose_project_name,
+    )
+    from .bottle_state import bottle_state_dir
+
+    state_dir = bottle_state_dir(slug)
+    compose_file = compose_file_path(state_dir)
+    if compose_file.exists():
+        compose_down(compose_project_name(slug), compose_file)
+
+
+def _destroy_smolmachines(slug: str) -> None:
+    """Best-effort stop + delete for a smolmachines bottle.
+
+    Both steps are best-effort: a machine that is already gone does not
+    cause an error; partial failures are logged as warnings."""
+    import subprocess
+
+    from .log import warn
+
+    machine = f"bot-bottle-{slug}"
+    subprocess.run(
+        ["smolvm", "machine", "stop", "--name", machine],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        check=False,
+    )
+    r = subprocess.run(
+        ["smolvm", "machine", "delete", "-f", machine],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if r.returncode != 0:
+        warn(
+            f"smolvm machine delete -f {machine!r} failed "
+            f"(may already be gone): {(r.stderr or '').strip()}"
+        )
+
+
+__all__ = [
+    "BottleError",
+    "destroy",
+    "freeze",
+    "resume_headless",
+    "start_headless",
+]

bot_bottle/backend/__init__.py

@@ -37,7 +37,7 @@ import shlex
 import sys
 from abc import ABC, abstractmethod
 from contextlib import AbstractContextManager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Generic, Sequence, TypeVar
 
@@ -75,6 +75,11 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
+    # Forge sidecar env vars (PRD forge-native-integration, chunk 1).
+    # Passed by the orchestrator at launch time; the forge sidecar reads
+    # them to connect to Gitea. Empty for non-forge runs. The agent
+    # process itself does not receive these.
+    forge_env: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)

bot_bottle/cli/resume.py

@@ -74,10 +74,11 @@ def cmd_resume(argv: list[str]) -> int:
         bottle_names=tuple(metadata.bottle_names),
     )
     backend_name = metadata.backend or None
-    return _launch_bottle(
+    _, rc = _launch_bottle(
         spec,
         dry_run=args.dry_run,
         backend_name=backend_name,
         assume_yes=args.headless,
         headless_prompt_text=args.prompt or "",
     )
+    return rc

bot_bottle/cli/start.py

@@ -144,11 +144,12 @@ def cmd_start(argv: list[str]) -> int:
         color=color,
         bottle_names=bottle_names,
     )
-    return _launch_bottle(
+    _, rc = _launch_bottle(
         spec,
         dry_run=dry_run,
         backend_name=backend_name,
     )
+    return rc
 
 
 # --- Headless launch -----------------------------------------------------
@@ -203,13 +204,14 @@ def _start_headless(
         color=args.color or "",
         bottle_names=bottle_names,
     )
-    return _launch_bottle(
+    _, rc = _launch_bottle(
         spec,
         dry_run=dry_run,
         backend_name=backend_name,
         assume_yes=True,
         headless_prompt_text=prompt,
     )
+    return rc
 
 
 def _uniquify_label_headless(label: str) -> str:
@@ -497,11 +499,16 @@ def _launch_bottle(
     backend_name: str | None = None,
     assume_yes: bool = False,
     headless_prompt_text: str = "",
-) -> int:
+) -> tuple[str, int]:
     """Shared launch core for `start` and `resume`. Builds the plan,
     prints / dry-runs / prompts as appropriate, brings the bottle up,
     attaches claude, and prints the resume hint on session end.
 
+    Returns ``(slug, exit_code)`` where ``slug`` is the bottle identity
+    (empty string when the launch was aborted before a slug was minted)
+    and ``exit_code`` is the agent process's exit code (0 on clean exit
+    or when launch was aborted before the agent ran).
+
     `assume_yes` skips the interactive y/N confirmation (headless /
     orchestrator launches), where there is no human at the prompt.
 
@@ -510,6 +517,7 @@ def _launch_bottle(
     agent receives the initial task without interactive input."""
     stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
     identity = ""
+    exit_code = 0
     try:
         plan, identity = prepare_with_preflight(
             spec,
@@ -520,7 +528,7 @@ def _launch_bottle(
             backend_name=backend_name,
         )
         if plan is None:
-            return 0
+            return identity, 0
 
         backend = get_bottle_backend(backend_name)
         with backend.launch(plan) as bottle:
@@ -547,7 +555,7 @@ def _launch_bottle(
             # Ctrl-Cs / OOM kills before cleanup removes the state dir.
             if agent_provider_template == "claude":
                 capture_claude_session_state(identity, exit_code)
-        return 0
+        return identity, exit_code
     finally:
         # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount
         # sources under state/<slug>/. If we never reached the