didericis-claude
18e610c7a8
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>
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""Bottle runner: drive bot_bottle to manage a bottle's life.
|
|
|
|
`BottleRunner` is the interface the lifecycle depends on;
|
|
`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.
|
|
|
|
`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
|
|
from collections.abc import Sequence
|
|
from typing import Protocol
|
|
|
|
|
|
class BottleRunner(Protocol):
|
|
def start(
|
|
self,
|
|
*,
|
|
agent: str,
|
|
bottles: Sequence[str],
|
|
label: str,
|
|
prompt: str,
|
|
forge_env: dict[str, str],
|
|
) -> str: ...
|
|
|
|
def freeze(self, slug: str) -> None: ...
|
|
|
|
def resume(self, slug: str, prompt: str) -> None: ...
|
|
|
|
def destroy(self, slug: str) -> None: ...
|
|
|
|
|
|
_SLUG_RE = re.compile(r"[^a-z0-9]+")
|
|
|
|
|
|
def slugify(label: str) -> str:
|
|
"""Lowercase, collapse non-alphanumerics to single hyphens, strip
|
|
leading/trailing hyphens — matches bot-bottle's slug rule."""
|
|
return _SLUG_RE.sub("-", label.lower()).strip("-")
|
|
|
|
|
|
class ProgrammaticBottleRunner:
|
|
"""Calls into the bot_bottle Python API directly — no subprocess.
|
|
|
|
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,
|
|
*,
|
|
agent: str,
|
|
bottles: Sequence[str],
|
|
label: str,
|
|
prompt: str,
|
|
forge_env: dict[str, str],
|
|
) -> 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,
|
|
)
|
|
|
|
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) -> 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) -> None:
|
|
from bot_bottle import api # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
|
api.destroy(slug)
|