didericis
a0e17d56de
Create bot_bottle/paths.py as the canonical home for the app-root path helpers (bot_bottle_root, host_db_path, HOST_DB_FILENAME) — foundational, not supervise- or db-specific. `bot_bottle_root()` now honours a BOT_BOTTLE_ROOT env override. Repoint every consumer (supervise, supervise_types, db_store, queue_store, audit_store, store_manager, bottle_state, cli/supervise, docker/cleanup, orchestrator/registry) at paths; remove the definitions (and supervise's duplicate host_db_path) and the now-dead `import sys`. Add paths.py to the sidecar bundle (Dockerfile.sidecars) for the flat-import copies. Tests: replace ~12 files' monkeypatching of supervise.bot_bottle_root (and the flat/pkg/supervise_types triple-patch dance) with a single `use_bottle_root()` helper that sets BOT_BOTTLE_ROOT — every module and flat/package copy reads the same env var, so one override covers them all. Net -97 lines. Behaviour-preserving: full unit suite unchanged (only the pre-existing /bin/sleep sidecar-init errors remain). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
"""Unit: cli/start.py session-end state capture (crash preservation).
|
|
|
|
The launch-context machinery is covered by integration; this isolates
|
|
the post-exec_agent decision: snapshot transcript + mark for
|
|
preservation if non-zero exit, no-op for clean exit."""
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from tests.unit import use_bottle_root
|
|
from bot_bottle import bottle_state
|
|
from bot_bottle.cli import start as start_mod
|
|
|
|
|
|
class _FakeHomeMixin:
|
|
def _setup_fake_home(self):
|
|
self._tmp = tempfile.TemporaryDirectory(prefix="cli-start-settle.")
|
|
self._restore = use_bottle_root(Path(self._tmp.name) / ".bot-bottle")
|
|
|
|
def _teardown_fake_home(self):
|
|
self._restore()
|
|
self._tmp.cleanup()
|
|
|
|
|
|
class TestCaptureSessionState(_FakeHomeMixin, unittest.TestCase):
|
|
# capture_claude_session_state handles the preserve marker for
|
|
# non-zero agent exits.
|
|
def setUp(self):
|
|
self._setup_fake_home()
|
|
|
|
def tearDown(self):
|
|
self._teardown_fake_home()
|
|
|
|
def test_clean_exit_does_not_mark(self):
|
|
start_mod.capture_claude_session_state("dev-abc", exit_code=0)
|
|
self.assertFalse(bottle_state.is_preserved("dev-abc"))
|
|
|
|
def test_crash_marks_preserved(self):
|
|
start_mod.capture_claude_session_state("dev-abc", exit_code=137)
|
|
self.assertTrue(bottle_state.is_preserved("dev-abc"))
|
|
|
|
def test_ctrl_c_treated_as_crash(self):
|
|
# SIGINT delivers exit 130; the operator may have Ctrl-C'd
|
|
# because something went wrong, so we preserve.
|
|
start_mod.capture_claude_session_state("dev-abc", exit_code=130)
|
|
self.assertTrue(bottle_state.is_preserved("dev-abc"))
|
|
|
|
def test_empty_identity_is_noop(self):
|
|
# Backends without an identity field shouldn't crash this
|
|
# path (the _identity_from_plan helper falls back to "").
|
|
start_mod.capture_claude_session_state("", exit_code=137)
|
|
self.assertFalse(bottle_state.is_preserved(""))
|
|
|
|
|
|
class TestSettleState(_FakeHomeMixin, unittest.TestCase):
|
|
def setUp(self):
|
|
self._setup_fake_home()
|
|
|
|
def tearDown(self):
|
|
self._teardown_fake_home()
|
|
|
|
def test_preserved_state_survives(self):
|
|
bottle_state.write_per_bottle_dockerfile("dev-abc", "FROM x\n")
|
|
bottle_state.mark_preserved("dev-abc")
|
|
start_mod.settle_state("dev-abc")
|
|
self.assertTrue(bottle_state.bottle_state_dir("dev-abc").is_dir())
|
|
|
|
def test_unpreserved_state_is_cleaned(self):
|
|
bottle_state.write_per_bottle_dockerfile("dev-abc", "FROM x\n")
|
|
start_mod.settle_state("dev-abc")
|
|
self.assertFalse(bottle_state.bottle_state_dir("dev-abc").exists())
|
|
|
|
def test_empty_identity_is_noop(self):
|
|
start_mod.settle_state("") # should not raise
|
|
|
|
|
|
class TestAttachAgent(unittest.TestCase):
|
|
def test_passes_provider_startup_args(self):
|
|
class Bottle:
|
|
argv: list[str] = []
|
|
|
|
def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
|
|
self.argv = list(argv)
|
|
return 0
|
|
|
|
bottle = Bottle()
|
|
exit_code = start_mod.attach_agent(
|
|
bottle, # type: ignore[arg-type]
|
|
agent_provider_template="pi",
|
|
startup_args=("--models", "openrouter/google/gemma"),
|
|
)
|
|
|
|
self.assertEqual(0, exit_code)
|
|
self.assertEqual(
|
|
["--models", "openrouter/google/gemma"],
|
|
bottle.argv,
|
|
)
|
|
|
|
def test_remote_control_is_provider_startup_arg(self):
|
|
class Bottle:
|
|
argv: list[str] = []
|
|
|
|
def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
|
|
self.argv = list(argv)
|
|
return 0
|
|
|
|
bottle = Bottle()
|
|
exit_code = start_mod.attach_agent(
|
|
bottle, # type: ignore[arg-type]
|
|
agent_provider_template="codex",
|
|
startup_args=("remote-control",),
|
|
)
|
|
|
|
self.assertEqual(0, exit_code)
|
|
self.assertEqual(
|
|
["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
|
|
bottle.argv,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|