diff --git a/lib/python/agentic_isolation/agentic_isolation/__init__.py b/lib/python/agentic_isolation/agentic_isolation/__init__.py index 76c45665..7eabceaf 100644 --- a/lib/python/agentic_isolation/agentic_isolation/__init__.py +++ b/lib/python/agentic_isolation/agentic_isolation/__init__.py @@ -51,7 +51,9 @@ WorkspaceConfig, ) from agentic_isolation.providers import ( + AwaitResult, ExecuteResult, + InteractiveSession, Workspace, WorkspaceDockerProvider, WorkspaceLocalProvider, @@ -93,6 +95,8 @@ "WorkspaceProvider", "Workspace", "ExecuteResult", + "InteractiveSession", + "AwaitResult", "WorkspaceLocalProvider", "WorkspaceDockerProvider", # Workspace file staging primitives diff --git a/lib/python/agentic_isolation/agentic_isolation/providers/__init__.py b/lib/python/agentic_isolation/agentic_isolation/providers/__init__.py index d2a7a805..9b516f8e 100644 --- a/lib/python/agentic_isolation/agentic_isolation/providers/__init__.py +++ b/lib/python/agentic_isolation/agentic_isolation/providers/__init__.py @@ -12,7 +12,9 @@ from typing import TYPE_CHECKING, Any from agentic_isolation.providers.base import ( + AwaitResult, ExecuteResult, + InteractiveSession, Workspace, WorkspaceProvider, ) @@ -46,6 +48,8 @@ def __getattr__(name: str) -> Any: "WorkspaceProvider", "Workspace", "ExecuteResult", + "InteractiveSession", + "AwaitResult", # Providers "WorkspaceLocalProvider", "WorkspaceDockerProvider", diff --git a/lib/python/agentic_isolation/agentic_isolation/providers/base.py b/lib/python/agentic_isolation/agentic_isolation/providers/base.py index 3944a382..df8fef54 100644 --- a/lib/python/agentic_isolation/agentic_isolation/providers/base.py +++ b/lib/python/agentic_isolation/agentic_isolation/providers/base.py @@ -44,6 +44,59 @@ def to_dict(self) -> dict[str, Any]: } +@runtime_checkable +class AwaitResult(Protocol): + """Structural read port for the result of waiting for an interactive + agent pane to reach a ready/idle state. + + This is a *structural* Protocol, not a concrete dataclass. The + interactive-tmux driver defines its own `AwaitResult` dataclass + (`providers/workspaces/interactive-tmux/driver/interactive_tmux.py`) + and that real object is what `interactive_session().await_completion()` + actually returns. Declaring this as a Protocol (rather than a duplicate + dataclass) means `isinstance(driver_result, AwaitResult)` is True at + runtime, the two definitions can never drift, and `agentic_isolation` + still never needs to import the driver (which is stdlib-only and lives + outside this package). + + Exposes only the read surface consumers need: + + - ready=True -> adapter is_ready() held stable + - timed_out=True, ready=False -> deadline hit before idle + - ready=False, reason="never_ready" -> never reached even one ready frame + - ready=False, reason="unstable" -> ready frames seen but pane kept changing + + `pane` carries the last captured pane text so callers don't have to + re-capture to inspect post-mortem; `success` mirrors `ready`. + Serialization (`to_dict()`) lives on the concrete driver dataclass, not + on this read port. + """ + + @property + def ready(self) -> bool: ... + + @property + def timed_out(self) -> bool: ... + + @property + def reason(self) -> str: ... + + @property + def duration_ms(self) -> float: ... + + @property + def stable_polls_observed(self) -> int: ... + + @property + def pane(self) -> str: ... + + @property + def error(self) -> str | None: ... + + @property + def success(self) -> bool: ... + + @dataclass class Workspace: """Represents an active isolated workspace.""" @@ -178,6 +231,48 @@ async def file_exists( ... +@runtime_checkable +class InteractiveSession(Protocol): + """Typed port for driving an interactive, prompt-based agent session + (e.g. a tmux-driven claude/codex/gemini TUI running in a container). + + This is a structural protocol: any object exposing these three + methods with compatible signatures satisfies it, including the + driver's `InteractiveTmuxWorkspace` instance today (no wrapper class + needed). It exists so consumers can type against a stable port + instead of reaching into a provider-specific `Workspace._handle: Any`. + """ + + def send_message(self, agent: str, text: str) -> None: + """Send a message/prompt to the named agent's pane.""" + ... + + def await_completion( + self, + agent: str, + *, + timeout: float = 60.0, + stable_polls: int = 4, + poll_interval: float = 0.5, + ) -> AwaitResult: + """Block until the named agent's pane reaches a stable ready state. + + Args: + agent: Agent name (e.g. "claude", "codex", "gemini"). + timeout: Overall deadline in seconds. + stable_polls: Number of consecutive "ready" polls required. + poll_interval: Seconds between polls. + + Returns: + AwaitResult describing whether/how readiness was reached. + """ + ... + + def capture_response(self, agent: str) -> str: + """Capture the current response text from the named agent's pane.""" + ... + + class BaseProvider(ABC): """Abstract base class for workspace providers. @@ -241,6 +336,16 @@ async def file_exists( """Check if file exists.""" ... + def interactive_session(self, workspace: Workspace) -> InteractiveSession | None: + """Return an `InteractiveSession` port for `workspace`, if supported. + + Default implementation returns `None`: most providers (docker, + local) don't support interactive prompt round-trips. Providers + that do (e.g. `InteractiveTmuxProvider`) should override this to + return an object satisfying the `InteractiveSession` protocol. + """ + return None + @staticmethod async def _terminate_process(proc: asyncio.subprocess.Process) -> None: """Ensure a subprocess is terminated.""" diff --git a/lib/python/agentic_isolation/agentic_isolation/providers/interactive_tmux/__init__.py b/lib/python/agentic_isolation/agentic_isolation/providers/interactive_tmux/__init__.py index f50c21ad..4ad5e13e 100644 --- a/lib/python/agentic_isolation/agentic_isolation/providers/interactive_tmux/__init__.py +++ b/lib/python/agentic_isolation/agentic_isolation/providers/interactive_tmux/__init__.py @@ -10,7 +10,7 @@ * `create()` → `InteractiveTmuxWorkspace.start_workspace(...)` (the running container exposes claude/codex/gemini panes - via the underlying `_handle` for orchestration code that + via `interactive_session()` for orchestration code that wants direct access). * `destroy()` → `InteractiveTmuxWorkspace.stop()` * `execute()`, `read_file()`, `write_file()`, `file_exists()` @@ -20,8 +20,10 @@ uses, returning a proper `ExecuteResult` so call sites that already speak the protocol need no translation. -The richer prompt round-trip API (send/await/capture) stays available on -the underlying workspace via `workspace._handle: InteractiveTmuxWorkspace`. +The richer prompt round-trip API (send/await/capture) is exposed as a +typed `agentic_isolation.providers.base.InteractiveSession` port via +`provider.interactive_session(workspace)` (structurally satisfied by the +underlying `InteractiveTmuxWorkspace` handle -- no wrapper class needed). """ from __future__ import annotations @@ -42,6 +44,7 @@ from agentic_isolation.providers.base import ( BaseProvider, ExecuteResult, + InteractiveSession, Workspace, ) @@ -181,11 +184,11 @@ class InteractiveTmuxProvider(BaseProvider): await provider.destroy(workspace) To reach the underlying claude/codex/gemini panes for prompt - round-trips, pull the driver workspace off the handle: - ws: InteractiveTmuxWorkspace = workspace._handle - ws.send_message("claude", "hello") - result: AwaitResult = ws.await_completion("claude") - print(ws.capture_response("claude")) + round-trips, get the typed `InteractiveSession` port: + session = provider.interactive_session(workspace) + session.send_message("claude", "hello") + result = session.await_completion("claude") + print(session.capture_response("claude")) """ def __init__( @@ -225,6 +228,18 @@ def __init__( self._startup_timeout_s = startup_timeout_s self._strict_startup = strict_startup self._workspaces: dict[str, Workspace] = {} + # Guards ONLY mutations of the `self._workspaces` registry dict (the + # sole shared state that needs it). It is deliberately NOT held + # across the blocking `start_workspace`/`stop` driver calls: doing so + # serialized every provision/teardown across all workspaces and made + # a `destroy()` block behind an in-flight `create()`, defeating the + # multi-agent concurrency this provider exists to enable. The + # blocking calls run under `asyncio.to_thread` WITHOUT this lock, so + # unrelated workspaces provision concurrently. (The previously feared + # child-watcher race between two threaded `subprocess.run` calls was + # never empirically reproduced, and the driver reaps its own child + # via `os.waitpid` in a blocking `subprocess.run`, not through + # asyncio's child watcher, so the race cannot occur here.) self._lock = asyncio.Lock() @property @@ -286,25 +301,32 @@ async def create(self, config: WorkspaceConfig) -> Workspace: workdir = config.working_dir or DEFAULT_CONTAINER_WORKDIR name = f"itws-{uuid.uuid4().hex[:8]}" - # The driver is blocking (subprocess + sleep loops). Calling it - # synchronously briefly blocks the event loop, but that's the - # right tradeoff: when an executor thread shells out via - # `subprocess.run` it races asyncio's child watcher and the docker - # run/exec calls become flaky (we saw 127 from docker exec right - # after a successful docker run -d). Holding the loop for the - # ~5s container-start window is acceptable; if a future caller - # needs concurrency, they can wrap `await provider.create(...)` - # in their own thread. - ws_handle: InteractiveTmuxWorkspace = driver.InteractiveTmuxWorkspace.start_workspace( - name=name, - host_auth=host_auth, - image=image, - workdir=workdir, - startup_timeout_s=self._startup_timeout_s, - strict_startup=self._strict_startup, - host_claude_dotjson=self._default_host_claude_dotjson, - claude_plugin_dirs=self._default_claude_plugin_dirs, + # The driver is blocking (subprocess + sleep loops), so offload it via + # `asyncio.to_thread` to keep the event loop free. It runs WITHOUT any + # cross-workspace lock, so N concurrent `create()` calls provision in + # parallel and a `destroy()` never waits behind an in-flight create. + # `asyncio.shield` keeps the start running if THIS create() is + # cancelled; the done callback then stops the late handle so nothing + # leaks (see `_schedule_create_cancellation_cleanup`). + loop = asyncio.get_running_loop() + start_task = asyncio.create_task( + asyncio.to_thread( + driver.InteractiveTmuxWorkspace.start_workspace, + name=name, + host_auth=host_auth, + image=image, + workdir=workdir, + startup_timeout_s=self._startup_timeout_s, + strict_startup=self._strict_startup, + host_claude_dotjson=self._default_host_claude_dotjson, + claude_plugin_dirs=self._default_claude_plugin_dirs, + ) ) + try: + ws_handle: InteractiveTmuxWorkspace = await asyncio.shield(start_task) + except asyncio.CancelledError: + self._schedule_create_cancellation_cleanup(start_task, name, loop) + raise # The container is now running with throwaway claude/codex/gemini # credentials mounted. Any failure between here and a successful @@ -325,6 +347,13 @@ async def create(self, config: WorkspaceConfig) -> Workspace: metadata={ "container": ws_handle.container, "workdir": workdir, + # `workspace_dir` is what downstream artifact collection + # (#225) reads to know where to copy files out of. This + # provider is exec-based with no host bind-mount, so it is + # the CONTAINER-side workspace path (identical to + # `workdir`) -- consumers resolve container-relative + # copies from it. It is intentionally NOT a host path. + "workspace_dir": workdir, "enabled_agents": list(ws_handle.enabled_agents), "startup_status": {a: r.to_dict() for a, r in ws_handle.startup_status.items()}, }, @@ -336,10 +365,13 @@ async def create(self, config: WorkspaceConfig) -> Workspace: except BaseException: # Best-effort teardown, then re-raise. BaseException so a # cancellation mid-setup also cleans up the credential-mounted - # container. stop() is synchronous and fast (<2s). + # container. Shield the stop() so it runs to completion even if + # THIS cleanup is itself cancelled (double-cancellation): the + # inner `except BaseException` catches that second CancelledError + # too, so no credential-seeded container can escape teardown. try: - ws_handle.stop() - except Exception: + await asyncio.shield(asyncio.to_thread(ws_handle.stop)) + except BaseException: logger.warning( "failed to stop workspace %s during create() cleanup", name, @@ -347,17 +379,74 @@ async def create(self, config: WorkspaceConfig) -> Workspace: ) raise + def _schedule_create_cancellation_cleanup( + self, + start_task: asyncio.Task[Any], + name: str, + loop: asyncio.AbstractEventLoop, + ) -> None: + """Stop a workspace that finishes starting after create() is cancelled. + + `asyncio.to_thread()` cannot kill the underlying worker thread. If + `create()` is cancelled while `start_workspace()` is still running, + the thread may still return a live workspace handle later. Attach a + done callback so that late handle is stopped and does not leak. + """ + + def _on_started(task: asyncio.Task[Any]) -> None: + try: + late_handle = task.result() + except BaseException: + return + loop.create_task(self._stop_late_started_workspace(late_handle, name)) + + start_task.add_done_callback(_on_started) + + async def _stop_late_started_workspace( + self, + ws_handle: InteractiveTmuxWorkspace, + name: str, + ) -> None: + try: + await asyncio.shield(asyncio.to_thread(ws_handle.stop)) + except BaseException: + logger.warning( + "failed to stop workspace %s after create() cancellation", + name, + exc_info=True, + ) + async def destroy(self, workspace: Workspace) -> None: ws_handle: InteractiveTmuxWorkspace | None = workspace._handle async with self._lock: self._workspaces.pop(workspace.id, None) if ws_handle is None: return - # Sync for the same reason as create(): the driver shells out - # via subprocess.run and the asyncio child-watcher race causes - # `docker rm -f` to return spurious non-zero exit codes from - # an executor thread. Stop is fast (<2s). - ws_handle.stop() + # Offloaded via `asyncio.to_thread` because `stop()` shells out via + # blocking `subprocess.run`. No cross-workspace lock is held, so a + # destroy() never blocks behind an in-flight create() (or another + # destroy) on an unrelated workspace. + await asyncio.to_thread(ws_handle.stop) + + def interactive_session(self, workspace: Workspace) -> InteractiveSession | None: + """Return the `InteractiveSession` port for `workspace`. + + The underlying driver's `InteractiveTmuxWorkspace` handle already + exposes `send_message`/`await_completion`/`capture_response` with + signatures compatible with `InteractiveSession`; structural typing + means it satisfies the protocol as-is, so no wrapper is needed. + Returns `None` if the workspace has no live handle (e.g. already + destroyed). + """ + handle = workspace._handle + if handle is None: + return None + if not isinstance(handle, InteractiveSession): + raise TypeError( + f"workspace._handle ({type(handle)!r}) does not satisfy " + "InteractiveSession — expected an InteractiveTmuxWorkspace" + ) + return handle async def execute( self, diff --git a/lib/python/agentic_isolation/tests/test_interactive_session_protocol.py b/lib/python/agentic_isolation/tests/test_interactive_session_protocol.py new file mode 100644 index 00000000..b18af7bb --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_session_protocol.py @@ -0,0 +1,231 @@ +"""Tests for the typed InteractiveSession port (Phase 1 of ADR/issue #225). + +Covers: + * AwaitResult is now a structural Protocol (not a duplicate dataclass); + a driver-shaped dataclass satisfies isinstance() against it, so the + two definitions can never drift. + * InteractiveSession protocol -- structural isinstance() checks against + a stub object, without needing the interactive-tmux driver. + * BaseProvider.interactive_session() default returns None. + * InteractiveTmuxProvider.interactive_session() returns workspace._handle. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from agentic_isolation.config import WorkspaceConfig +from agentic_isolation.providers.base import ( + AwaitResult, + BaseProvider, + ExecuteResult, + InteractiveSession, + Workspace, +) + + +@dataclass +class _DriverShapedAwaitResult: + """Field-for-field copy of the interactive-tmux driver's own + `AwaitResult` dataclass -- used to prove the real driver result + satisfies the base `AwaitResult` Protocol structurally, without + importing the driver.""" + + ready: bool + timed_out: bool + reason: str + duration_ms: float + stable_polls_observed: int + pane: str = "" + error: str | None = None + + @property + def success(self) -> bool: + return self.ready + + def to_dict(self) -> dict[str, object]: + return { + "ready": self.ready, + "timed_out": self.timed_out, + "reason": self.reason, + "duration_ms": self.duration_ms, + "stable_polls_observed": self.stable_polls_observed, + "pane": self.pane, + "error": self.error, + } + + +class TestAwaitResultProtocol: + def test_driver_shaped_result_satisfies_protocol(self) -> None: + """A driver-shaped dataclass satisfies the base `AwaitResult` + Protocol structurally -- so the isinstance() check that used to + always be False (two drifting dataclasses) now holds.""" + result = _DriverShapedAwaitResult( + ready=True, + timed_out=False, + reason="ready", + duration_ms=123.4, + stable_polls_observed=4, + ) + assert isinstance(result, AwaitResult) + + def test_read_surface_is_accessible(self) -> None: + result = _DriverShapedAwaitResult( + ready=False, + timed_out=True, + reason="timeout_never_ready", + duration_ms=60000.0, + stable_polls_observed=0, + pane="last frame", + error="boom", + ) + # Every member the port promises is readable off the concrete result. + assert result.ready is False + assert result.timed_out is True + assert result.reason == "timeout_never_ready" + assert result.duration_ms == 60000.0 + assert result.stable_polls_observed == 0 + assert result.pane == "last frame" + assert result.error == "boom" + assert result.success is False + + def test_plain_object_does_not_satisfy_protocol(self) -> None: + assert not isinstance(object(), AwaitResult) + + +class _StubInteractiveSession: + """Minimal object with the right method shapes but no relation to the + driver -- exercises the protocol structurally.""" + + def send_message(self, agent: str, text: str) -> None: + return None + + def await_completion( + self, + agent: str, + *, + timeout: float = 60.0, + stable_polls: int = 4, + poll_interval: float = 0.5, + ) -> AwaitResult: + return _DriverShapedAwaitResult( + ready=True, + timed_out=False, + reason="ready", + duration_ms=0.0, + stable_polls_observed=stable_polls, + ) + + def capture_response(self, agent: str) -> str: + return "" + + +class TestInteractiveSessionProtocol: + def test_stub_satisfies_protocol_structurally(self) -> None: + stub = _StubInteractiveSession() + assert isinstance(stub, InteractiveSession) + + def test_object_missing_methods_does_not_satisfy_protocol(self) -> None: + class NotASession: + def send_message(self, agent: str, text: str) -> None: + return None + + assert not isinstance(NotASession(), InteractiveSession) + + def test_plain_object_does_not_satisfy_protocol(self) -> None: + assert not isinstance(object(), InteractiveSession) + + +class _MinimalProvider(BaseProvider): + """Smallest possible BaseProvider subclass, to exercise the default + `interactive_session()` implementation without touching docker/local + providers.""" + + @property + def name(self) -> str: + return "minimal" + + async def create(self, config: WorkspaceConfig) -> Workspace: # pragma: no cover + raise NotImplementedError + + async def destroy(self, workspace: Workspace) -> None: # pragma: no cover + raise NotImplementedError + + async def execute( + self, + workspace: Workspace, + command: str, + *, + timeout: float | None = None, + cwd: str | None = None, + env: dict[str, str] | None = None, + ) -> ExecuteResult: # pragma: no cover + raise NotImplementedError + + async def write_file( + self, workspace: Workspace, path: str, content: str | bytes + ) -> None: # pragma: no cover + raise NotImplementedError + + async def read_file(self, workspace: Workspace, path: str) -> str: # pragma: no cover + raise NotImplementedError + + async def file_exists(self, workspace: Workspace, path: str) -> bool: # pragma: no cover + raise NotImplementedError + + +class TestBaseProviderDefault: + def test_interactive_session_defaults_to_none(self) -> None: + provider = _MinimalProvider() + config = WorkspaceConfig() + workspace = Workspace( + id="ws-1", + provider="minimal", + path=Path("/workspace"), + config=config, + ) + assert provider.interactive_session(workspace) is None + + +class TestInteractiveTmuxProviderOverride: + def test_interactive_session_returns_handle(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Without importing the real driver, verify the adapter's + `interactive_session()` returns `workspace._handle` when present.""" + from agentic_isolation.providers import interactive_tmux as adapter_module + + # Avoid constructing via __init__ (which loads the driver); use + # object.__new__ since interactive_session() only reads + # workspace._handle and doesn't touch provider state. + provider = object.__new__(adapter_module.InteractiveTmuxProvider) + + stub_handle = _StubInteractiveSession() + config = WorkspaceConfig() + workspace = Workspace( + id="ws-1", + provider="interactive-tmux", + path=Path("/workspace"), + config=config, + _handle=stub_handle, + ) + + session = provider.interactive_session(workspace) + assert session is stub_handle + assert isinstance(session, InteractiveSession) + + def test_interactive_session_none_when_no_handle(self) -> None: + from agentic_isolation.providers import interactive_tmux as adapter_module + + provider = object.__new__(adapter_module.InteractiveTmuxProvider) + config = WorkspaceConfig() + workspace = Workspace( + id="ws-1", + provider="interactive-tmux", + path=Path("/workspace"), + config=config, + _handle=None, + ) + + assert provider.interactive_session(workspace) is None diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_adapter_nonblocking.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_adapter_nonblocking.py new file mode 100644 index 00000000..f8628b6b --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_adapter_nonblocking.py @@ -0,0 +1,315 @@ +"""Regression tests for the async follow-up on issue #225 (phase 3). + +`InteractiveTmuxProvider.create()`/`.destroy()` used to call the blocking +driver (`start_workspace`/`stop`, both `subprocess.run`-backed) directly on +the caller's event loop, freezing every other task on that loop for the +full container startup/teardown window. They now offload the blocking +call via `asyncio.to_thread`. + +An earlier revision also serialized every offloaded call through a single +`asyncio.Lock`, but that made N concurrent `create()` calls run strictly +sequentially and a `destroy()` block behind an in-flight `create()` -- +defeating the multi-agent concurrency this provider exists to enable. The +lock is gone; only the `self._workspaces` registry dict is guarded. + +These tests prove two things: + + 1. `create()`/`destroy()` no longer block the event loop: a concurrently + scheduled `asyncio.sleep()` task completes *during* the call, not + only after it returns. + 2. Concurrent `create()` calls are NOT serialized: with a driver stub + whose `start_workspace` blocks, multiple creates overlap in time + (observed concurrency > 1) instead of running one at a time. +""" + +from __future__ import annotations + +import asyncio +import threading +import time +from pathlib import Path + +import pytest + +import agentic_isolation.providers.interactive_tmux as itm +from agentic_isolation.config import WorkspaceConfig +from agentic_isolation.providers.interactive_tmux import InteractiveTmuxProvider + + +def _provider() -> InteractiveTmuxProvider: + return InteractiveTmuxProvider( + default_host_auth={"claude": Path("/tmp/nowhere/.claude")}, + default_host_claude_dotjson=Path("/tmp/nowhere/.claude.json"), + default_claude_plugin_dirs=[], + ) + + +class _FakeHandle: + def __init__(self, sleep_s: float = 0.0) -> None: + self.container = "itws-fake" + self.enabled_agents = ("claude",) + self.startup_status: dict = {} + self.stopped = False + self._sleep_s = sleep_s + + def stop(self) -> None: + time.sleep(self._sleep_s) + self.stopped = True + + +class _SlowStartWorkspace: + """Driver stand-in whose `start_workspace` blocks synchronously.""" + + sleep_s = 0.3 + last_handle: _FakeHandle | None = None + + @classmethod + def start_workspace(cls, **kwargs) -> _FakeHandle: + time.sleep(cls.sleep_s) + handle = _FakeHandle() + cls.last_handle = handle + return handle + + +class _SlowStartDriver: + DEFAULT_IMAGE = "img:test" + InteractiveTmuxWorkspace = _SlowStartWorkspace + + +async def _noop_docker_exec(provider: InteractiveTmuxProvider) -> None: + async def _ok(*args, **kwargs): + return None + + provider._docker_exec = _ok # type: ignore[method-assign] + + +async def test_create_sets_workspace_dir_metadata(monkeypatch) -> None: + """#225 spec: metadata must carry a `workspace_dir` key so downstream + artifact collection knows where to copy files from. For this exec-based + provider it is the container workdir (no host bind-mount).""" + monkeypatch.setattr(itm, "_get_driver", lambda: _SlowStartDriver) + _SlowStartWorkspace.sleep_s = 0.0 + provider = _provider() + await _noop_docker_exec(provider) + + config = WorkspaceConfig(provider="interactive-tmux", working_dir="/workspace") + workspace = await provider.create(config) + + assert "workspace_dir" in workspace.metadata + assert workspace.metadata["workspace_dir"] == "/workspace" + # For this provider it mirrors the container workdir. + assert workspace.metadata["workspace_dir"] == workspace.metadata["workdir"] + + +async def test_create_does_not_block_event_loop(monkeypatch) -> None: + monkeypatch.setattr(itm, "_get_driver", lambda: _SlowStartDriver) + _SlowStartWorkspace.sleep_s = 0.3 + provider = _provider() + await _noop_docker_exec(provider) + + ticks = 0 + + async def _ticker() -> None: + nonlocal ticks + while True: + await asyncio.sleep(0.05) + ticks += 1 + + ticker_task = asyncio.ensure_future(_ticker()) + try: + await provider.create(WorkspaceConfig(provider="interactive-tmux")) + finally: + ticker_task.cancel() + + # `start_workspace` slept synchronously for 0.3s. If `create()` had + # blocked the event loop for that whole window, the ticker (0.05s + # cadence) would not have gotten a chance to run at all. Offloaded via + # `asyncio.to_thread`, the loop stays free to service it several times. + assert ticks >= 3, ( + f"expected the event loop to keep servicing other tasks while " + f"create() ran, but only {ticks} ticks completed" + ) + + +async def test_destroy_does_not_block_event_loop() -> None: + provider = _provider() + workspace = await _make_workspace(provider, sleep_s=0.3) + + ticks = 0 + + async def _ticker() -> None: + nonlocal ticks + while True: + await asyncio.sleep(0.05) + ticks += 1 + + ticker_task = asyncio.ensure_future(_ticker()) + try: + await provider.destroy(workspace) + finally: + ticker_task.cancel() + + assert ticks >= 3, ( + f"expected the event loop to keep servicing other tasks while " + f"destroy() ran, but only {ticks} ticks completed" + ) + assert workspace._handle.stopped is True + + +async def _make_workspace(provider: InteractiveTmuxProvider, *, sleep_s: float): + """Build a `Workspace` with a fake handle without going through + `create()` (keeps these tests independent of the driver stub above).""" + from datetime import UTC, datetime + + from agentic_isolation.providers.base import Workspace + + handle = _FakeHandle(sleep_s=sleep_s) + workspace = Workspace( + id="itws-fake", + provider=provider.name, + path=Path("/workspace"), + config=WorkspaceConfig(provider="interactive-tmux"), + created_at=datetime.now(UTC), + metadata={"container": handle.container, "workdir": "/workspace"}, + _handle=handle, + ) + provider._workspaces[workspace.id] = workspace + return workspace + + +class _ConcurrencyDriver: + """Driver stand-in that records how many `start_workspace` calls are + executing at the same time. + + Each call bumps a shared counter, sleeps (holding the overlap window + open), then decrements. If the provider serialized offloaded calls + (the old behavior), `max_observed_concurrency` would stay at 1. With + the cross-workspace lock removed, concurrent creates overlap and the + counter climbs above 1. + """ + + DEFAULT_IMAGE = "img:test" + max_observed_concurrency = 0 + current_concurrency = 0 + _guard = threading.Lock() + + class InteractiveTmuxWorkspace: + @classmethod + def start_workspace(cls, **kwargs) -> _FakeHandle: + return _ConcurrencyDriver._enter_blocking_section() + + @classmethod + def _enter_blocking_section(cls) -> _FakeHandle: + with cls._guard: + cls.current_concurrency += 1 + cls.max_observed_concurrency = max( + cls.max_observed_concurrency, cls.current_concurrency + ) + try: + # Hold the overlap window open long enough that sibling creates, + # now that they are not serialized, reliably run at the same time. + time.sleep(0.1) + finally: + with cls._guard: + cls.current_concurrency -= 1 + return _FakeHandle() + + +async def test_concurrent_creates_are_not_serialized(monkeypatch) -> None: + monkeypatch.setattr(itm, "_get_driver", lambda: _ConcurrencyDriver) + _ConcurrencyDriver.max_observed_concurrency = 0 + _ConcurrencyDriver.current_concurrency = 0 + + provider = _provider() + await _noop_docker_exec(provider) + + await asyncio.gather( + provider.create(WorkspaceConfig(provider="interactive-tmux")), + provider.create(WorkspaceConfig(provider="interactive-tmux")), + provider.create(WorkspaceConfig(provider="interactive-tmux")), + ) + + assert _ConcurrencyDriver.max_observed_concurrency > 1, ( + "concurrent create() calls did not overlap -- provisioning is still " + "serialized across workspaces, defeating multi-agent concurrency" + ) + + +async def test_cancelled_create_stops_late_started_workspace(monkeypatch) -> None: + monkeypatch.setattr(itm, "_get_driver", lambda: _SlowStartDriver) + _SlowStartWorkspace.sleep_s = 0.2 + _SlowStartWorkspace.last_handle = None + provider = _provider() + await _noop_docker_exec(provider) + + task = asyncio.create_task(provider.create(WorkspaceConfig(provider="interactive-tmux"))) + await asyncio.sleep(0.05) + task.cancel() + + try: + await task + except asyncio.CancelledError: + pass + + for _ in range(20): + handle = _SlowStartWorkspace.last_handle + if handle is not None and handle.stopped: + break + await asyncio.sleep(0.05) + + assert _SlowStartWorkspace.last_handle is not None + assert _SlowStartWorkspace.last_handle.stopped is True + + +class _StopSleepWorkspace: + """Driver whose `start_workspace` succeeds fast but whose handle's + `stop()` blocks briefly, giving a window to cancel `create()` *while + the cleanup stop() is in flight*.""" + + last_handle: _FakeHandle | None = None + + @classmethod + def start_workspace(cls, **kwargs) -> _FakeHandle: + handle = _FakeHandle(sleep_s=0.2) + cls.last_handle = handle + return handle + + +class _StopSleepDriver: + DEFAULT_IMAGE = "img:test" + InteractiveTmuxWorkspace = _StopSleepWorkspace + + +async def test_cleanup_stop_runs_when_cancelled_during_teardown(monkeypatch) -> None: + """Finding 3: post-start setup fails (entering create()'s teardown), + then a SECOND cancellation lands while the cleanup `stop()` is awaiting. + The shielded stop() must still run to completion so the + credential-seeded container is never leaked.""" + monkeypatch.setattr(itm, "_get_driver", lambda: _StopSleepDriver) + _StopSleepWorkspace.last_handle = None + provider = _provider() + + async def _boom(*args, **kwargs): + raise RuntimeError("post-start setup failed") + + provider._docker_exec = _boom # type: ignore[method-assign] + + task = asyncio.create_task(provider.create(WorkspaceConfig(provider="interactive-tmux"))) + # Let create() start, fail setup, and enter the shielded cleanup stop(). + await asyncio.sleep(0.05) + task.cancel() + + with pytest.raises((asyncio.CancelledError, RuntimeError)): + await task + + for _ in range(40): + handle = _StopSleepWorkspace.last_handle + if handle is not None and handle.stopped: + break + await asyncio.sleep(0.05) + + assert _StopSleepWorkspace.last_handle is not None + assert _StopSleepWorkspace.last_handle.stopped is True, ( + "cleanup stop() was skipped when create() was cancelled during " + "teardown -- the credential-seeded container leaked" + ) diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_cleanup.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_cleanup.py index 479fc1c0..dfe3e68a 100644 --- a/lib/python/agentic_isolation/tests/test_interactive_tmux_cleanup.py +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_cleanup.py @@ -94,7 +94,7 @@ def test_docker_run_failure_removes_throwaway_dir( no_real_subprocess: list[list[str]], monkeypatch: pytest.MonkeyPatch, ) -> None: - def failing_run(cmd, check=True, capture=True): + def failing_run(cmd, check=True, capture=True, timeout_s=None): if cmd[:2] == ["docker", "run"]: raise subprocess.CalledProcessError(125, cmd) return subprocess.CompletedProcess(cmd, 0, "", "") diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_environment.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_environment.py new file mode 100644 index 00000000..55ff4f3f --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_environment.py @@ -0,0 +1,614 @@ +"""Environment seam tests (issue #225 follow-up): provisioning becomes +pluggable, one layer above `CommandExecutor`. + +Covers: + * `DockerEnvironment.start()` runs the exact `docker run ...` shape + `start_workspace` always ran, and returns a `DockerExecExecutor` bound + to the container it created. + * `DockerEnvironment.stop()` runs `docker rm -f ` and tolerates a + wedged/timed-out `subprocess.run` the same way the old inline + `InteractiveTmuxWorkspace.stop()` logic did. + * `start_workspace(environment=None)` (the default) builds its own + `DockerEnvironment` from `image`/`workdir` — zero behavior change for + every existing caller. + * `start_workspace(environment=)` uses the injected `Environment` + instead of constructing a `DockerEnvironment` / shelling out to + `docker run` at all, and `ws.stop()` delegates to the injected + environment's `.stop()`. + +No real docker daemon or tmux required; subprocess calls are monkeypatched +at the driver module level, matching the convention of the sibling +interactive-tmux test files. +""" + +from __future__ import annotations + +import importlib.util +import subprocess +import sys +from pathlib import Path + +import pytest + + +def _load_driver_module(): + here = Path(__file__).resolve() + for ancestor in here.parents: + candidate = ( + ancestor + / "providers" + / "workspaces" + / "interactive-tmux" + / "driver" + / "interactive_tmux.py" + ) + if candidate.is_file(): + spec = importlib.util.spec_from_file_location("interactive_tmux", candidate) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules.setdefault("interactive_tmux", module) + spec.loader.exec_module(module) + return module + pytest.skip("interactive_tmux driver not found in repo layout") + + +driver = _load_driver_module() + + +@pytest.fixture +def fake_claude_home(tmp_path: Path) -> Path: + home = tmp_path / "home" + claude_dir = home / ".claude" + claude_dir.mkdir(parents=True) + (claude_dir / ".credentials.json").write_text("{}") + (home / ".claude.json").write_text("{}") + return claude_dir + + +class _FakeEnvironment: + """Records lifecycle calls instead of touching docker at all.""" + + def __init__(self, executor=None) -> None: + self.started = False + self.stopped = False + self._executor = executor or _FakeExecutor() + + def start(self): + self.started = True + return self._executor + + def stop(self) -> None: + self.stopped = True + + +class _FakeExecutor: + def __init__(self) -> None: + self.calls: list[list[str]] = [] + + def exec(self, command, *, timeout_s=None, stdin=None): # noqa: ARG002 + self.calls.append(list(command)) + return driver.ExecResult(exit_code=0, stdout="", stderr="") + + +class TestEnvironmentProtocol: + def test_docker_environment_satisfies_protocol(self) -> None: + env = driver.DockerEnvironment(name="c", image="img", workdir="/workspace") + assert isinstance(env, driver.Environment) + + def test_fake_environment_satisfies_protocol(self) -> None: + assert isinstance(_FakeEnvironment(), driver.Environment) + + +class TestDockerEnvironmentStart: + def test_start_runs_expected_docker_run_shape(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, check=True, capture=True, timeout_s=None): + captured["cmd"] = cmd + captured["timeout_s"] = timeout_s + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver, "_run", fake_run) + + env = driver.DockerEnvironment( + name="my-container", image="my-image:latest", workdir="/workspace" + ) + executor = env.start() + + assert captured["cmd"] == [ + "docker", + "run", + "-d", + "--name", + "my-container", + "--workdir", + "/workspace", + "my-image:latest", + "sleep", + "infinity", + ] + assert isinstance(executor, driver.DockerExecExecutor) + assert executor.target == "my-container" + assert captured["timeout_s"] == driver.DEFAULT_RUN_TIMEOUT_S + + def test_start_honors_custom_run_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, check=True, capture=True, timeout_s=None): + captured["timeout_s"] = timeout_s + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver, "_run", fake_run) + + env = driver.DockerEnvironment( + name="my-container", + image="my-image:latest", + workdir="/workspace", + run_timeout_s=2.5, + ) + env.start() + + assert captured["timeout_s"] == 2.5 + + def test_start_propagates_run_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + def failing_run(cmd, **kwargs): + raise subprocess.CalledProcessError(125, cmd) + + monkeypatch.setattr(driver, "_run", failing_run) + env = driver.DockerEnvironment(name="c", image="img", workdir="/workspace") + with pytest.raises(subprocess.CalledProcessError): + env.start() + + +class TestDockerEnvironmentStop: + def test_stop_runs_docker_rm_dash_f(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + env = driver.DockerEnvironment(name="my-container", image="img", workdir="/workspace") + env.stop() + + assert captured["cmd"] == ["docker", "rm", "-f", "my-container"] + assert captured["kwargs"]["check"] is False + + def test_stop_tolerates_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + def timing_out_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout")) + + monkeypatch.setattr(driver.subprocess, "run", timing_out_run) + env = driver.DockerEnvironment(name="c", image="img", workdir="/workspace") + # Must not raise. + env.stop() + + +class TestStartWorkspaceEnvironmentInjection: + def test_default_environment_is_docker_environment( + self, fake_claude_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """`environment=None` (the default) must build a `DockerEnvironment` + from `image`/`workdir` exactly as `start_workspace` always did — + zero behavior change for every existing caller.""" + captured: dict = {} + + def fake_run(cmd, check=True, capture=True, timeout_s=None): + captured["run_cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver, "_run", fake_run) + monkeypatch.setattr( + driver.InteractiveTmuxWorkspace, + "_bootstrap_tmux_and_launch", + lambda self, *a, **k: None, + ) + monkeypatch.setattr( + driver.subprocess, + "run", + lambda cmd, **kwargs: subprocess.CompletedProcess(cmd, 0, "", ""), + ) + + ws = driver.InteractiveTmuxWorkspace.start_workspace( + name="envdefault", + host_auth={"claude": fake_claude_home}, + image="custom-image:latest", + workdir="/custom-workdir", + ) + try: + assert isinstance(ws.environment, driver.DockerEnvironment) + assert ws.environment.image == "custom-image:latest" + assert ws.environment.workdir == "/custom-workdir" + assert captured["run_cmd"][:2] == ["docker", "run"] + assert "--name" in captured["run_cmd"] + finally: + ws.stop() + + def test_injected_environment_is_used_instead_of_docker_run( + self, fake_claude_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """When `environment=` is passed, `start_workspace` must not shell + out to `docker run` at all — it uses the injected `Environment`.""" + fake_executor = _FakeExecutor() + fake_env = _FakeEnvironment(executor=fake_executor) + + def unexpected_run(cmd, **kwargs): + raise AssertionError( + f"start_workspace should not call _run() when environment= is injected: {cmd}" + ) + + monkeypatch.setattr(driver, "_run", unexpected_run) + monkeypatch.setattr( + driver.InteractiveTmuxWorkspace, + "_bootstrap_tmux_and_launch", + lambda self, *a, **k: None, + ) + + ws = driver.InteractiveTmuxWorkspace.start_workspace( + name="envinjected", + host_auth={"claude": fake_claude_home}, + environment=fake_env, + ) + + assert fake_env.started is True + assert ws.environment is fake_env + assert ws.executor is fake_executor + # Credential bytes must have been transferred over the injected + # executor (not a freshly-constructed DockerExecExecutor). + assert fake_executor.calls + + ws.stop() + assert fake_env.stopped is True + + def test_injected_environment_stop_called_on_bootstrap_failure( + self, fake_claude_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A failure after provisioning (e.g. tmux bootstrap) must still + tear down the injected environment, not just a DockerEnvironment.""" + fake_env = _FakeEnvironment() + + def failing_bootstrap(self, *a, **k): + raise RuntimeError("boom") + + monkeypatch.setattr( + driver.InteractiveTmuxWorkspace, "_bootstrap_tmux_and_launch", failing_bootstrap + ) + + with pytest.raises(RuntimeError, match="boom"): + driver.InteractiveTmuxWorkspace.start_workspace( + name="envfail", + host_auth={"claude": fake_claude_home}, + environment=fake_env, + ) + + assert fake_env.started is True + assert fake_env.stopped is True + + def test_bootstrap_passes_injected_executor_to_agent_launch( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fake_executor = _FakeExecutor() + ws = driver.InteractiveTmuxWorkspace( + name="envbootstrap", + container="not-a-docker-container", + image="img", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/envbootstrap"), + enabled_agents=("claude",), + executor=fake_executor, + ) + captured: dict = {} + + def fake_launch(container, workdir, plugin_dirs=None, *, executor=None, timeout_s=None): + captured["container"] = container + captured["workdir"] = workdir + captured["plugin_dirs"] = plugin_dirs + captured["executor"] = executor + captured["timeout_s"] = timeout_s + + monkeypatch.setattr( + driver._ADAPTERS["claude"], + "launch_in_window", + staticmethod(fake_launch), + ) + monkeypatch.setattr( + ws, + "_wait_for_started", + lambda agent, timeout_s: driver.AwaitResult( + ready=True, + timed_out=False, + reason="ready", + duration_ms=0.0, + stable_polls_observed=1, + ), + ) + + ws._bootstrap_tmux_and_launch(startup_timeout_s=1.0, strict_startup=True) + + assert captured["container"] == "not-a-docker-container" + assert captured["workdir"] == "/workspace" + assert captured["executor"] is fake_executor + + +class TestLocalExecutor: + def test_exec_happy_path_runs_argv_directly_no_docker_prefix( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "out", "err") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + executor = driver.LocalExecutor(workdir=tmp_path) + result = executor.exec(["echo", "hi"], timeout_s=5) + + assert captured["cmd"] == ["echo", "hi"] + assert captured["kwargs"]["cwd"] == tmp_path + assert captured["kwargs"]["timeout"] == 5 + assert captured["kwargs"]["capture_output"] is True + assert captured["kwargs"]["text"] is True + assert result == driver.ExecResult(exit_code=0, stdout="out", stderr="err") + + def test_exec_timeout_returns_timed_out_exec_result( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + def timing_out_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout")) + + monkeypatch.setattr(driver.subprocess, "run", timing_out_run) + + executor = driver.LocalExecutor(workdir=tmp_path) + result = executor.exec(["sleep", "99"], timeout_s=1) + + assert result.timed_out is True + assert result.exit_code == -1 + assert "timed out after 1s" in result.stderr + + def test_exec_cwd_passed_through(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cwd"] = kwargs.get("cwd") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + executor = driver.LocalExecutor(workdir=tmp_path) + executor.exec(["pwd"]) + + assert captured["cwd"] == tmp_path + + +class TestLocalEnvironment: + def test_satisfies_environment_protocol(self, tmp_path: Path) -> None: + env = driver.LocalEnvironment(workdir=tmp_path) + assert isinstance(env, driver.Environment) + + def test_start_raises_runtime_error_when_tmux_missing( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(driver.shutil, "which", lambda tool: None) + env = driver.LocalEnvironment(workdir=tmp_path) + + with pytest.raises(RuntimeError, match="tmux"): + env.start() + + def test_start_returns_local_executor_when_tmux_present( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(driver.shutil, "which", lambda tool: "/usr/bin/tmux") + env = driver.LocalEnvironment(workdir=tmp_path) + + executor = env.start() + + assert isinstance(executor, driver.LocalExecutor) + assert executor.workdir == tmp_path + + def test_stop_is_a_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(driver.shutil, "which", lambda tool: "/usr/bin/tmux") + env = driver.LocalEnvironment(workdir=tmp_path) + env.start() + # Must not raise, must not touch subprocess at all. + env.stop() + + +class TestSSHExecutor: + def test_exec_happy_path_builds_expected_argv(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "out", "err") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + base_argv = [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=10", + "-i", + "/home/me/.ssh/id_ed25519", + "-p", + "22", + "--", + "me@example.com", + ] + executor = driver.SSHExecutor(base_argv) + result = executor.exec(["echo", "hi there"], timeout_s=5) + + assert captured["cmd"] == [*base_argv, "echo 'hi there'"] + assert captured["kwargs"]["timeout"] == 5 + assert captured["kwargs"]["capture_output"] is True + assert captured["kwargs"]["text"] is True + assert result == driver.ExecResult(exit_code=0, stdout="out", stderr="err") + + def test_exec_timeout_returns_timed_out_exec_result( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + def timing_out_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout")) + + monkeypatch.setattr(driver.subprocess, "run", timing_out_run) + + executor = driver.SSHExecutor(["ssh", "me@example.com"]) + result = executor.exec(["sleep", "99"], timeout_s=1) + + assert result.timed_out is True + assert result.exit_code == -1 + assert "timed out after 1s" in result.stderr + + def test_exec_prefixes_cd_when_workdir_set(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + base_argv = ["ssh", "me@example.com"] + executor = driver.SSHExecutor(base_argv, workdir="/remote/work") + executor.exec(["pwd"]) + + assert captured["cmd"] == [*base_argv, "cd /remote/work && pwd"] + + def test_exec_omits_cd_when_no_workdir(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + base_argv = ["ssh", "me@example.com"] + executor = driver.SSHExecutor(base_argv) + executor.exec(["pwd"]) + + assert captured["cmd"] == [*base_argv, "pwd"] + + +class TestSSHEnvironment: + def test_satisfies_environment_protocol(self) -> None: + env = driver.SSHEnvironment(host="example.com", user="me") + assert isinstance(env, driver.Environment) + + def test_start_runs_reachability_check_with_expected_argv( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + env = driver.SSHEnvironment( + host="example.com", + user="me", + key_path="/home/me/.ssh/id_ed25519", + port=2222, + workdir="/remote/work", + connect_timeout_s=7.0, + ) + executor = env.start() + + assert captured["cmd"] == [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=7", + "-i", + "/home/me/.ssh/id_ed25519", + "-p", + "2222", + "--", + "me@example.com", + "true", + ] + assert captured["kwargs"]["timeout"] == 7.0 + assert isinstance(executor, driver.SSHExecutor) + assert executor.workdir == "/remote/work" + assert executor.base_argv == [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=7", + "-i", + "/home/me/.ssh/id_ed25519", + "-p", + "2222", + "--", + "me@example.com", + ] + + def test_base_argv_uses_option_terminator_before_destination(self) -> None: + env = driver.SSHEnvironment(host="-oProxyCommand=bad", user="-l bad") + + argv = env._base_argv() + + assert argv[-2] == "--" + assert argv[-1] == "-l bad@-oProxyCommand=bad" + + def test_start_without_key_path_omits_dash_i(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + env = driver.SSHEnvironment(host="example.com", user="me") + env.start() + + assert "-i" not in captured["cmd"] + + def test_start_raises_runtime_error_with_stderr_on_failure( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + def fake_run(cmd, **kwargs): + return subprocess.CompletedProcess( + cmd, 255, "", "ssh: connect to host example.com port 22: Connection refused" + ) + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + env = driver.SSHEnvironment(host="example.com", user="me") + + with pytest.raises(RuntimeError, match="Connection refused"): + env.start() + + def test_start_raises_runtime_error_on_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + def timing_out_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout")) + + monkeypatch.setattr(driver.subprocess, "run", timing_out_run) + + env = driver.SSHEnvironment(host="example.com", user="me", connect_timeout_s=3.0) + + with pytest.raises(RuntimeError, match="timed out"): + env.start() + + def test_stop_is_a_noop(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(cmd, **kwargs): + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + env = driver.SSHEnvironment(host="example.com", user="me") + env.start() + env.stop() # must not raise diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_executor.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_executor.py new file mode 100644 index 00000000..21368691 --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_executor.py @@ -0,0 +1,385 @@ +"""Phase 2 tests: `CommandExecutor` seam + exec-based credential seeding. + +Covers the docker-out-of-docker (DooD) fix: the driver used to stage +credentials into a host tempdir and bind-mount them into the container at +`docker run` time (`-v host:container`). That breaks when the driver itself +runs inside a container, because the *driver's* host paths aren't visible to +the sibling `docker run` on the outer host. + +Phase 2 replaces that with: run the container WITHOUT credential mounts, +then push each credential file's bytes into the container over an injected +`CommandExecutor` (base64-chunked `docker exec` writes). These tests verify: + + * `ExecResult` / `CommandExecutor` / `DockerExecExecutor` shapes. + * `_docker_exec` routes through an injected executor when given one, and + constructs a default `DockerExecExecutor` when not. + * `_write_bytes_to_container` / `_transfer_path_to_container` reconstruct + file bytes and directory trees faithfully via a fake executor (no real + docker daemon). + * `start_workspace` no longer passes `-v` bind-mount flags to `docker run`, + and instead transfers credential file bytes over the executor. + +No real docker daemon or tmux is required; `subprocess.run` and the driver's +`_run`/`_docker_exec` seams are monkeypatched. +""" + +from __future__ import annotations + +import base64 +import importlib.util +import subprocess +import sys +from pathlib import Path + +import pytest + + +def _load_driver_module(): + here = Path(__file__).resolve() + for ancestor in here.parents: + candidate = ( + ancestor + / "providers" + / "workspaces" + / "interactive-tmux" + / "driver" + / "interactive_tmux.py" + ) + if candidate.is_file(): + spec = importlib.util.spec_from_file_location("interactive_tmux", candidate) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules.setdefault("interactive_tmux", module) + spec.loader.exec_module(module) + return module + pytest.skip("interactive_tmux driver not found in repo layout") + + +driver = _load_driver_module() + + +class _FakeExecutor: + """Records every `exec()` call; replays a scripted (exit_code) per call. + + Also incrementally "executes" `sh -c` file-write commands against an + in-memory filesystem so directory/file transfer tests can assert on the + reconstructed bytes without any real docker/subprocess involvement. + """ + + def __init__(self) -> None: + self.calls: list[list[str]] = [] + self.fs: dict[str, bytes] = {} + self.chowned: list[str] = [] + self.stdins: list[bytes | None] = [] + self.last_stdin: bytes | None = None + + def exec(self, command, *, timeout_s=None, stdin=None): # noqa: ARG002 + self.calls.append(list(command)) + self.stdins.append(stdin) + self.last_stdin = stdin + # Emulate the subset of shell behavior the driver's helpers rely on: + # `mkdir -p`, `> path` (truncate/create), `base64 -d >> path` fed the + # payload over STDIN (credential-leak fix), `chown ...`, + # `find ... chmod ...`. + if command[:1] == ["mkdir"]: + return driver.ExecResult(exit_code=0, stdout="", stderr="") + if command[:1] == ["sh"] and len(command) >= 3 and command[1] == "-c": + script = command[2] + if script.startswith(">"): + # `> 'path'` truncate/create. + path = script[1:].strip() + path = path.strip("'\"") + self.fs[path] = b"" + return driver.ExecResult(exit_code=0, stdout="", stderr="") + if "base64 -d >>" in script: + # `base64 -d >> 'path'` with the base64 payload on STDIN. + path = script.split("base64 -d >>")[1].strip().strip("'\"") + assert stdin is not None, "credential payload must arrive via stdin, not argv" + self.fs[path] = self.fs.get(path, b"") + base64.b64decode(stdin) + return driver.ExecResult(exit_code=0, stdout="", stderr="") + if "chown" in script: + self.chowned.append(script) + return driver.ExecResult(exit_code=0, stdout="", stderr="") + return driver.ExecResult(exit_code=0, stdout="", stderr="") + + +class TestExecResultShape: + def test_fields(self) -> None: + result = driver.ExecResult(exit_code=0, stdout="out", stderr="err") + assert result.exit_code == 0 + assert result.stdout == "out" + assert result.stderr == "err" + + +class TestDockerExecExecutor: + def test_exec_shells_out_via_docker_exec(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "hello\n", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + executor = driver.DockerExecExecutor("my-container") + result = executor.exec(["echo", "hello"]) + + assert captured["cmd"] == ["docker", "exec", "my-container", "echo", "hello"] + assert result.exit_code == 0 + assert result.stdout == "hello\n" + + def test_nonzero_exit_does_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(cmd, **kwargs): + return subprocess.CompletedProcess(cmd, 7, "", "boom") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + executor = driver.DockerExecExecutor("c") + result = executor.exec(["false"]) + + assert result.exit_code == 7 + assert result.stderr == "boom" + + def test_timeout_s_forwarded_to_subprocess_run(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + driver.DockerExecExecutor("c").exec(["true"], timeout_s=5.0) + + assert captured["kwargs"].get("timeout") == 5.0 + + +class TestDockerExecRoutesThroughExecutor: + def test_default_constructs_docker_exec_executor(self, monkeypatch: pytest.MonkeyPatch) -> None: + constructed: list[str] = [] + real_cls = driver.DockerExecExecutor + + class RecordingExecutor(real_cls): + def __init__(self, container): + constructed.append(container) + super().__init__(container) + + monkeypatch.setattr(driver, "DockerExecExecutor", RecordingExecutor) + monkeypatch.setattr( + driver.subprocess, + "run", + lambda cmd, **kwargs: subprocess.CompletedProcess(cmd, 0, "", ""), + ) + + driver._docker_exec("some-container", "true") + + assert constructed == ["some-container"] + + def test_injected_executor_is_used_instead_of_default(self) -> None: + fake = _FakeExecutor() + driver._docker_exec("ignored-container", "mkdir", "-p", "/x", executor=fake) + assert fake.calls == [["mkdir", "-p", "/x"]] + + def test_check_true_raises_on_nonzero_from_injected_executor(self) -> None: + class FailingExecutor: + def exec(self, command, *, timeout_s=None): + return driver.ExecResult(exit_code=1, stdout="", stderr="nope") + + with pytest.raises(subprocess.CalledProcessError): + driver._docker_exec("c", "false", executor=FailingExecutor()) + + def test_check_false_swallows_nonzero(self) -> None: + class FailingExecutor: + def exec(self, command, *, timeout_s=None): + return driver.ExecResult(exit_code=1, stdout="", stderr="nope") + + result = driver._docker_exec("c", "false", check=False, executor=FailingExecutor()) + assert result.returncode == 1 + + +class TestWriteBytesToContainer: + def test_small_payload_round_trips(self) -> None: + fake = _FakeExecutor() + driver._write_bytes_to_container(fake, "/home/agent/.credentials.json", b'{"token": "abc"}') + assert fake.fs["/home/agent/.credentials.json"] == b'{"token": "abc"}' + + def test_mkdir_dash_p_called_for_parent(self) -> None: + fake = _FakeExecutor() + driver._write_bytes_to_container(fake, "/home/agent/.claude/deep/file.json", b"{}") + assert ["mkdir", "-p", "/home/agent/.claude/deep"] in fake.calls + + def test_large_payload_round_trips_in_a_single_stdin_write(self) -> None: + fake = _FakeExecutor() + payload = bytes(range(256)) * 200 # 51200 bytes of arbitrary/binary data + driver._write_bytes_to_container(fake, "/home/agent/.codex/auth.json", payload) + assert fake.fs["/home/agent/.codex/auth.json"] == payload + # stdin has no argv length limit, so the whole file is one exec (no + # chunking loop): exactly one `base64 -d >>` write call. + b64_calls = [c for c in fake.calls if c[:1] == ["sh"] and "base64 -d >>" in c[2]] + assert len(b64_calls) == 1 + + def test_payload_never_appears_in_argv(self) -> None: + """Credential-leak fix: the base64 payload must travel over stdin, so + it must NOT appear in any command's argv (world-readable via `ps` / + `/proc//cmdline`).""" + fake = _FakeExecutor() + secret = b"super-secret-oauth-token-value-1234567890" + driver._write_bytes_to_container(fake, "/home/agent/.codex/auth.json", secret) + + b64_secret = base64.b64encode(secret).decode("ascii") + for call in fake.calls: + for arg in call: + assert secret.decode() not in arg + assert b64_secret not in arg + # It DID arrive over stdin instead. + assert any(s is not None and base64.b64decode(s) == secret for s in fake.stdins) + + +class TestTransferPathToContainer: + def test_single_file_transferred(self, tmp_path: Path) -> None: + src = tmp_path / "creds.json" + src.write_text('{"a": 1}') + fake = _FakeExecutor() + + driver._transfer_path_to_container(fake, src, "/home/agent/.claude.json") + + assert fake.fs["/home/agent/.claude.json"] == b'{"a": 1}' + + def test_directory_tree_transferred_preserving_relative_paths(self, tmp_path: Path) -> None: + src_dir = tmp_path / "codex-home" + (src_dir / "sessions").mkdir(parents=True) + (src_dir / "auth.json").write_text('{"auth": true}') + (src_dir / "sessions" / "s1.json").write_text('{"session": 1}') + fake = _FakeExecutor() + + driver._transfer_path_to_container(fake, src_dir, "/home/agent/.codex") + + assert fake.fs["/home/agent/.codex/auth.json"] == b'{"auth": true}' + assert fake.fs["/home/agent/.codex/sessions/s1.json"] == b'{"session": 1}' + + +class TestSecureContainerPath: + def test_file_chown_and_chmod_issued(self) -> None: + fake = _FakeExecutor() + driver._secure_container_path(fake, "/home/agent/.claude/.credentials.json", is_dir=False) + assert any("chown 1000:1000" in c and "chmod 600" in c for c in fake.chowned) + + def test_directory_chown_recursive_and_find_chmod(self) -> None: + fake = _FakeExecutor() + driver._secure_container_path(fake, "/home/agent/.codex", is_dir=True) + assert any("chown -R 1000:1000" in c and "find" in c for c in fake.chowned) + + +class TestStartWorkspaceNoLongerBindMounts: + """`start_workspace` must not bind-mount credentials at `docker run` time; + it transfers them into the running container over the executor instead. + """ + + @pytest.fixture + def fake_claude_home(self, tmp_path: Path) -> Path: + home = tmp_path / "home" + claude_dir = home / ".claude" + claude_dir.mkdir(parents=True) + (claude_dir / ".credentials.json").write_text('{"token": "tok"}') + (home / ".claude.json").write_text('{"oauthAccount": {"email": "a@b.com"}}') + return claude_dir + + def test_docker_run_has_no_dash_v_flags( + self, + fake_claude_home: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + run_calls: list[list[str]] = [] + + def fake_run(cmd, check=True, capture=True, timeout_s=None): # noqa: ARG001 + run_calls.append(list(cmd)) + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver, "_run", fake_run) + monkeypatch.setattr( + driver.subprocess, + "run", + lambda cmd, **kwargs: subprocess.CompletedProcess(cmd, 0, "", ""), + ) + # Avoid launching real tmux/agents during bootstrap. + monkeypatch.setattr( + driver.InteractiveTmuxWorkspace, + "_bootstrap_tmux_and_launch", + lambda self, *a, **k: None, + ) + + ws = driver.InteractiveTmuxWorkspace.start_workspace( + name="noboundtest", + host_auth={"claude": fake_claude_home}, + ) + + docker_run_calls = [c for c in run_calls if c[:2] == ["docker", "run"]] + assert len(docker_run_calls) == 1 + assert "-v" not in docker_run_calls[0] + ws.stop() + + def test_credential_bytes_transferred_over_executor( + self, + fake_claude_home: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setattr( + driver, + "_run", + lambda cmd, check=True, capture=True, timeout_s=None: subprocess.CompletedProcess( + cmd, 0, "", "" + ), + ) + monkeypatch.setattr( + driver.InteractiveTmuxWorkspace, + "_bootstrap_tmux_and_launch", + lambda self, *a, **k: None, + ) + + fake = _FakeExecutor() + monkeypatch.setattr(driver, "DockerExecExecutor", lambda container: fake) # noqa: ARG005 + monkeypatch.setattr( + driver.subprocess, + "run", + lambda cmd, **kwargs: subprocess.CompletedProcess(cmd, 0, "", ""), + ) + + ws = driver.InteractiveTmuxWorkspace.start_workspace( + name="transfertest", + host_auth={"claude": fake_claude_home}, + ) + + assert fake.fs["/home/agent/.claude/.credentials.json"] == b'{"token": "tok"}' + assert b"oauthAccount" in fake.fs["/home/agent/.claude.json"] + assert ws.executor is fake + ws.stop() + + +class TestWorkspaceExecutorField: + def test_defaults_to_docker_exec_executor_for_its_container(self) -> None: + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + ) + assert isinstance(ws.executor, driver.DockerExecExecutor) + assert ws.executor.target == "test-container" + + def test_explicit_executor_is_kept(self) -> None: + fake = _FakeExecutor() + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + executor=fake, + ) + assert ws.executor is fake diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_hardening.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_hardening.py index 7fc1d81c..cf404b73 100644 --- a/lib/python/agentic_isolation/tests/test_interactive_tmux_hardening.py +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_hardening.py @@ -107,3 +107,42 @@ def test_literal_payload_redacted(self) -> None: def test_non_literal_command_unchanged(self) -> None: cmd = ["tmux", "capture-pane", "-p", "-t", "agents:claude"] assert driver._redact_cmd(cmd) == "tmux capture-pane -p -t agents:claude" + + +def test_ignore_uncopyable_skips_special_files(tmp_path: Path) -> None: + """Auth staging must skip non-regular files (sockets/FIFOs) so a special + file in a .git tree (e.g. a git fsmonitor .ipc socket) cannot abort the + credential copytree. A FIFO exercises the same non-regular-file path as a + socket without the macOS AF_UNIX path-length limit.""" + import os as _os + + (tmp_path / "config.toml").write_text("ok") + (tmp_path / "sub").mkdir() + _os.mkfifo(tmp_path / "fsmonitor--daemon.ipc") + + names = [p.name for p in tmp_path.iterdir()] + skipped = driver._ignore_uncopyable(str(tmp_path), names) + assert "fsmonitor--daemon.ipc" in skipped + assert "config.toml" not in skipped + assert "sub" not in skipped + + +def test_ignore_uncopyable_allows_regular_tree(tmp_path: Path) -> None: + """With no special/heavy entries, nothing is skipped (copytree normal).""" + (tmp_path / "auth.json").write_text("{}") + (tmp_path / "sessions").mkdir() + names = [p.name for p in tmp_path.iterdir()] + assert driver._ignore_uncopyable(str(tmp_path), names) == set() + + +def test_ignore_uncopyable_skips_heavy_non_auth_dirs(tmp_path: Path) -> None: + """node_modules/.git are never auth material and balloon the stage; skip + them at any depth so credential staging stays fast and socket-free.""" + (tmp_path / "auth.json").write_text("{}") + (tmp_path / "node_modules").mkdir() + (tmp_path / ".git").mkdir() + names = [p.name for p in tmp_path.iterdir()] + skipped = driver._ignore_uncopyable(str(tmp_path), names) + assert "node_modules" in skipped + assert ".git" in skipped + assert "auth.json" not in skipped diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_pane_tail.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_pane_tail.py index a3d51687..a7e578a2 100644 --- a/lib/python/agentic_isolation/tests/test_interactive_tmux_pane_tail.py +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_pane_tail.py @@ -238,7 +238,7 @@ def test_capture_response_byte_equal_to_capture( monkeypatch.setattr( driver, "_tmux_capture", - lambda container, window: long_buffer, # type: ignore[arg-type] + lambda container, window, **kwargs: long_buffer, # type: ignore[arg-type] ) ws = driver.InteractiveTmuxWorkspace( diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_plugin_dirs.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_plugin_dirs.py index fd87cb19..9f0b2d6c 100644 --- a/lib/python/agentic_isolation/tests/test_interactive_tmux_plugin_dirs.py +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_plugin_dirs.py @@ -144,10 +144,10 @@ def test_no_plugin_dirs_preserves_keystroke_sequence( # this exact keystroke shape. calls: list[tuple[str, str, tuple[str, ...]]] = [] - def fake_send_keys(container: str, window: str, *keys: str) -> None: + def fake_send_keys(container: str, window: str, *keys: str, **kwargs) -> None: calls.append((container, window, keys)) - def fake_send_literal(container: str, window: str, text: str) -> None: # noqa: ARG001 + def fake_send_literal(container: str, window: str, text: str, **kwargs) -> None: # noqa: ARG001 pytest.fail("send_literal should not be called without plugin_dirs") monkeypatch.setattr(driver, "_tmux_send_keys", fake_send_keys) @@ -164,10 +164,10 @@ def test_with_plugin_dirs_sends_literal_then_enter( send_keys_calls: list[tuple[str, str, tuple[str, ...]]] = [] send_literal_calls: list[tuple[str, str, str]] = [] - def fake_send_keys(container: str, window: str, *keys: str) -> None: + def fake_send_keys(container: str, window: str, *keys: str, **kwargs) -> None: send_keys_calls.append((container, window, keys)) - def fake_send_literal(container: str, window: str, text: str) -> None: + def fake_send_literal(container: str, window: str, text: str, **kwargs) -> None: send_literal_calls.append((container, window, text)) monkeypatch.setattr(driver, "_tmux_send_keys", fake_send_keys) diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_reliability.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_reliability.py new file mode 100644 index 00000000..60a3ea50 --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_reliability.py @@ -0,0 +1,469 @@ +"""Phase 3 tests: bounded timeouts + tmux send-keys payload batching. + +Covers the reliability hardening pass: + + * `DockerExecExecutor.exec()` catches `subprocess.TimeoutExpired` and + returns a `timed_out=True` `ExecResult` instead of letting the + exception hang the caller. + * `_docker_exec` / `_run` forward a bounded default `timeout_s` to every + subprocess call so nothing in the driver blocks forever. + * `send_message`/`await_completion`/`capture_response` each pass a + bounded per-call timeout distinct from `await_completion`'s overall + deadline, and a single failed/timed-out poll doesn't abort the whole + `await_completion` call. + * Payloads over `TMUX_SEND_KEYS_MAX_BYTES` are staged via tmux + `load-buffer`/`paste-buffer` instead of raw `send-keys -l`; small + payloads keep using the raw path unchanged. + +No real docker daemon or tmux required; `subprocess.run` and the driver's +executor seam are monkeypatched/faked. +""" + +from __future__ import annotations + +import importlib.util +import subprocess +import sys +from pathlib import Path + +import pytest + + +def _load_driver_module(): + here = Path(__file__).resolve() + for ancestor in here.parents: + candidate = ( + ancestor + / "providers" + / "workspaces" + / "interactive-tmux" + / "driver" + / "interactive_tmux.py" + ) + if candidate.is_file(): + spec = importlib.util.spec_from_file_location("interactive_tmux", candidate) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules.setdefault("interactive_tmux", module) + spec.loader.exec_module(module) + return module + pytest.skip("interactive_tmux driver not found in repo layout") + + +driver = _load_driver_module() + + +class _FakeExecutor: + """Records every `exec()` call; can be scripted to raise/time out. + + Also emulates the subset of shell behavior `_write_bytes_to_container` + relies on (`mkdir -p`, `> path` truncate, `printf | base64 -d >>`) so + payload-staging tests can assert on reconstructed bytes without any + real docker/subprocess involvement — mirrors the fake in + `test_interactive_tmux_executor.py`. + """ + + def __init__(self) -> None: + self.calls: list[tuple[list[str], float | None]] = [] + self.fs: dict[str, bytes] = {} + + def exec(self, command, *, timeout_s=None, stdin=None): + self.calls.append((list(command), timeout_s)) + if command[:1] == ["mkdir"]: + return driver.ExecResult(exit_code=0, stdout="", stderr="") + if command[:1] == ["sh"] and len(command) >= 3 and command[1] == "-c": + script = command[2] + if script.startswith(">"): + path = script[1:].strip().strip("'\"") + self.fs[path] = b"" + return driver.ExecResult(exit_code=0, stdout="", stderr="") + if "base64 -d >>" in script: + import base64 + + # Payload now arrives over STDIN, not in argv (leak fix). + path = script.split("base64 -d >>")[1].strip().strip("'\"") + assert stdin is not None + self.fs[path] = self.fs.get(path, b"") + base64.b64decode(stdin) + return driver.ExecResult(exit_code=0, stdout="", stderr="") + return driver.ExecResult(exit_code=0, stdout="", stderr="") + + +class TestDockerExecExecutorTimeout: + def test_timeout_expired_becomes_timed_out_exec_result( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + def fake_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd=cmd, timeout=kwargs.get("timeout")) + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + executor = driver.DockerExecExecutor("c") + result = executor.exec(["tmux", "capture-pane"], timeout_s=2.0) + + assert result.timed_out is True + assert result.exit_code != 0 + assert "timed out" in result.stderr + + def test_no_timeout_expired_is_not_timed_out(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + driver.subprocess, + "run", + lambda cmd, **kwargs: subprocess.CompletedProcess(cmd, 0, "ok", ""), + ) + result = driver.DockerExecExecutor("c").exec(["true"], timeout_s=5.0) + assert result.timed_out is False + + +class TestBoundedDefaultTimeouts: + def test_docker_exec_forwards_default_timeout_to_injected_executor(self) -> None: + fake = _FakeExecutor() + driver._docker_exec("c", "tmux", "capture-pane", executor=fake) + (_cmd, timeout_s) = fake.calls[0] + assert timeout_s == driver.DEFAULT_EXEC_TIMEOUT_S + + def test_docker_exec_honors_explicit_timeout_override(self) -> None: + fake = _FakeExecutor() + driver._docker_exec("c", "tmux", "capture-pane", executor=fake, timeout_s=3.0) + (_cmd, timeout_s) = fake.calls[0] + assert timeout_s == 3.0 + + def test_run_forwards_default_run_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_subprocess_run(cmd, **kwargs): + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_subprocess_run) + driver._run(["docker", "run", "-d", "x"]) + assert captured["kwargs"].get("timeout") == driver.DEFAULT_RUN_TIMEOUT_S + + def test_tmux_capture_forwards_timeout_to_executor(self) -> None: + fake = _FakeExecutor() + driver._tmux_capture("c", "claude", executor=fake, timeout_s=7.0) + (_cmd, timeout_s) = fake.calls[0] + assert timeout_s == 7.0 + + def test_tmux_send_keys_forwards_timeout_to_executor(self) -> None: + fake = _FakeExecutor() + driver._tmux_send_keys("c", "claude", "Enter", executor=fake, timeout_s=4.0) + (_cmd, timeout_s) = fake.calls[0] + assert timeout_s == 4.0 + + +class TestSendKeysPayloadBatching: + def test_small_payload_uses_raw_send_keys(self) -> None: + fake = _FakeExecutor() + driver._tmux_send_literal("c", "claude", "hello world", executor=fake) + + assert len(fake.calls) == 1 + (cmd, _timeout) = fake.calls[0] + assert cmd[:2] == ["tmux", "send-keys"] + assert "-l" in cmd + + def test_payload_at_threshold_uses_raw_send_keys(self) -> None: + fake = _FakeExecutor() + text = "a" * driver.TMUX_SEND_KEYS_MAX_BYTES + driver._tmux_send_literal("c", "claude", text, executor=fake) + + send_keys_calls = [c for c, _ in fake.calls if c[:2] == ["tmux", "send-keys"]] + assert len(send_keys_calls) == 1 + + def test_large_payload_is_staged_via_load_and_paste_buffer(self) -> None: + fake = _FakeExecutor() + text = "x" * (driver.TMUX_SEND_KEYS_MAX_BYTES + 1000) + driver._tmux_send_literal("c", "claude", text, executor=fake) + + commands = [c for c, _ in fake.calls] + # No raw send-keys -l for the oversized payload. + assert not any(c[:2] == ["tmux", "send-keys"] and "-l" in c for c in commands) + assert any(c[:2] == ["tmux", "load-buffer"] for c in commands) + paste_calls = [c for c in commands if c[:2] == ["tmux", "paste-buffer"]] + assert paste_calls + # Finding 3: paste-buffer MUST use `-p` (bracketed paste) so a + # multiline payload's embedded newlines don't dispatch as individual + # Enter presses and submit the prompt early. + assert all("-p" in c for c in paste_calls) + + def test_large_payload_bytes_round_trip_through_write_bytes_to_container(self) -> None: + fake = _FakeExecutor() + text = "y" * (driver.TMUX_SEND_KEYS_MAX_BYTES + 500) + driver._tmux_send_literal("c", "claude", text, executor=fake) + + # Exactly one staged buffer file was written with the full payload. + written = [v for v in fake.fs.values() if v == text.encode("utf-8")] + assert len(written) == 1 + + def test_large_payload_cleans_up_staged_buffer_file(self) -> None: + fake = _FakeExecutor() + text = "z" * (driver.TMUX_SEND_KEYS_MAX_BYTES + 200) + driver._tmux_send_literal("c", "claude", text, executor=fake) + + commands = [c for c, _ in fake.calls] + assert any(c[:2] == ["rm", "-f"] for c in commands) + + +class TestAwaitCompletionResilientToTransientPollFailure: + def test_single_failed_poll_does_not_abort_await(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A poll that raises must be swallowed and retried, not propagated, + so a transient wedged `docker exec` can't crash `await_completion`. + """ + calls = {"n": 0} + ready_tail = "❯ \n? for shortcuts" + + def flaky_capture(container, window, **kwargs): + calls["n"] += 1 + if calls["n"] <= 2: + raise subprocess.CalledProcessError(1, ["docker", "exec"]) + return ready_tail + + monkeypatch.setattr(driver, "_tmux_capture", flaky_capture) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + ) + + result = ws.await_completion("claude", timeout=5.0, stable_polls=1, warmup=0.0) + + assert calls["n"] > 2 + assert result.ready is True + + def test_await_completion_accepts_poll_timeout_s(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured_timeouts: list[float | None] = [] + + def capture(container, window, **kwargs): + captured_timeouts.append(kwargs.get("timeout_s")) + return "❯ \n? for shortcuts" + + monkeypatch.setattr(driver, "_tmux_capture", capture) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + ) + + ws.await_completion("claude", timeout=5.0, stable_polls=1, warmup=0.0, poll_timeout_s=3.0) + + assert captured_timeouts + assert all(t == 3.0 for t in captured_timeouts) + # The per-poll timeout must be strictly smaller than the overall + # await deadline it was called with, so a wedged poll can't eat + # the whole budget silently. + assert 3.0 < 5.0 + + +class TestDockerExecStdin: + """Finding 1: credential payloads travel over STDIN, not argv. The + executor must add `-i` (keep stdin open) and forward the bytes as + `subprocess.run(input=...)`. + """ + + def test_stdin_adds_dash_i_and_forwards_input(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["input"] = kwargs.get("input") + captured["text"] = kwargs.get("text") + return subprocess.CompletedProcess(cmd, 0, b"", b"") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + driver.DockerExecExecutor("c").exec(["sh", "-c", "base64 -d >> /x"], stdin=b"payload") + + assert captured["cmd"] == ["docker", "exec", "-i", "c", "sh", "-c", "base64 -d >> /x"] + assert captured["input"] == b"payload" + # bytes input requires text=False; outputs are decoded via `_decode`. + assert captured["text"] is False + + def test_no_stdin_omits_dash_i_and_keeps_text_mode( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["text"] = kwargs.get("text") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + + driver.DockerExecExecutor("c").exec(["true"]) + + assert "-i" not in captured["cmd"] + assert captured["text"] is True + + +class TestRunExecCheckedRedactsCredentials: + """Finding 2: a failing credential-seeding exec must not leak the payload + OR the raw command into the raised error — only the redacted `label`. + """ + + def test_label_used_instead_of_raw_command_on_failure(self) -> None: + class FailingExecutor: + def exec(self, command, *, timeout_s=None, stdin=None): # noqa: ARG002 + return driver.ExecResult(exit_code=1, stdout="", stderr="boom") + + with pytest.raises(RuntimeError) as excinfo: + driver._run_exec_checked( + FailingExecutor(), + ["sh", "-c", "base64 -d >> /home/agent/.claude/.credentials.json"], + stdin=b"c2VjcmV0", + label="write bytes to /home/agent/.claude/.credentials.json", + ) + + msg = str(excinfo.value) + assert "write bytes to /home/agent/.claude/.credentials.json" in msg + # Neither the raw command list nor the base64 payload appears. + assert "base64 -d" not in msg + assert "c2VjcmV0" not in msg + + def test_write_bytes_failure_message_carries_no_payload(self) -> None: + secret = b"top-secret-token" + + class FailingExecutor: + def exec(self, command, *, timeout_s=None, stdin=None): # noqa: ARG002 + # Fail only on the actual base64 write (not mkdir / truncate). + if command[:1] == ["sh"] and "base64 -d >>" in command[2]: + return driver.ExecResult(exit_code=1, stdout="", stderr="disk full") + return driver.ExecResult(exit_code=0, stdout="", stderr="") + + import base64 as _b64 + + with pytest.raises(RuntimeError) as excinfo: + driver._write_bytes_to_container(FailingExecutor(), "/home/agent/.claude.json", secret) + + msg = str(excinfo.value) + assert secret.decode() not in msg + assert _b64.b64encode(secret).decode() not in msg + + +class TestReadinessBreaksFastOnDeadContainer: + """Finding 4: a container that is GONE must break the readiness poll + immediately (naming the death) instead of spinning the full deadline and + then reporting a misleading generic timeout. + """ + + @staticmethod + def _ws(): + return driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + ) + + def test_await_completion_breaks_immediately_on_dead_container( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls = {"n": 0} + + def dead_capture(container, window, **kwargs): # noqa: ARG001 + calls["n"] += 1 + raise subprocess.CalledProcessError( + 1, ["docker", "exec"], "", "Error: No such container: test-container" + ) + + monkeypatch.setattr(driver, "_tmux_capture", dead_capture) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + # Generous overall timeout: if the fix regressed, this would spin + # (many capture calls) instead of breaking after the first. + result = self._ws().await_completion("claude", timeout=100.0, stable_polls=1, warmup=0.0) + + assert result.ready is False + assert result.timed_out is False + assert result.reason == "container_dead" + assert "No such container" in (result.error or "") + assert calls["n"] == 1 + + def test_wait_for_started_breaks_immediately_on_dead_container( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls = {"n": 0} + + def dead_capture(container, window, **kwargs): # noqa: ARG001 + calls["n"] += 1 + raise subprocess.CalledProcessError( + 1, ["docker", "exec"], "", "Error response from daemon: Container is not running" + ) + + monkeypatch.setattr(driver, "_tmux_capture", dead_capture) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + result = self._ws()._wait_for_started("claude", 100.0) + + assert result.ready is False + assert result.timed_out is False + assert result.reason == "container_dead" + assert "is not running" in (result.error or "") + assert calls["n"] == 1 + + def test_transient_capture_failure_still_retries_while_alive( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A CalledProcessError whose stderr does NOT name a dead container is + a genuine transient hiccup and must stay on the retry path.""" + calls = {"n": 0} + + def flaky(container, window, **kwargs): # noqa: ARG001 + calls["n"] += 1 + if calls["n"] == 1: + raise subprocess.CalledProcessError(1, ["docker", "exec"], "", "temporary glitch") + return "❯ \n? for shortcuts" + + monkeypatch.setattr(driver, "_tmux_capture", flaky) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + result = self._ws().await_completion("claude", timeout=100.0, stable_polls=1, warmup=0.0) + + assert result.ready is True + assert calls["n"] >= 2 + + def test_docker_daemon_outage_is_transient_not_container_dead( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A daemon/socket connectivity error must NOT be read as container death. + + "Cannot connect to the Docker daemon" means the daemon blipped, not + that the container is gone; classifying it as container_dead would + abort a still-alive workspace. It must stay on the retry path and + recover once the daemon responds again. + """ + calls = {"n": 0} + + def daemon_blip(container, window, **kwargs): # noqa: ARG001 + calls["n"] += 1 + if calls["n"] == 1: + raise subprocess.CalledProcessError( + 1, + ["docker", "exec"], + "", + "Cannot connect to the Docker daemon at unix:///var/run/docker.sock.", + ) + return "❯ \n? for shortcuts" + + monkeypatch.setattr(driver, "_tmux_capture", daemon_blip) + monkeypatch.setattr(driver.time, "sleep", lambda *_a, **_k: None) + + result = self._ws().await_completion("claude", timeout=100.0, stable_polls=1, warmup=0.0) + + assert result.ready is True + assert result.reason != "container_dead" + assert calls["n"] >= 2 diff --git a/lib/python/agentic_isolation/tests/test_interactive_tmux_session.py b/lib/python/agentic_isolation/tests/test_interactive_tmux_session.py new file mode 100644 index 00000000..4d874778 --- /dev/null +++ b/lib/python/agentic_isolation/tests/test_interactive_tmux_session.py @@ -0,0 +1,262 @@ +"""Phase 4 tests: agent-agnostic `TmuxSession` + adapter-registry dispatch. + +Covers the session/agent split from issue #225 phase 4: + + * `TmuxSession` (send_keys/send_literal/capture_pane/get_incremental_output/ + is_alive/start/stop) is agent-agnostic and routes through the injected + `CommandExecutor` (via the module-level `_tmux_*`/`_docker_exec` seams, + so monkeypatching those still works whether callers go through + `TmuxSession` or the free functions directly). + * `InteractiveTmuxWorkspace` builds one `TmuxSession` per enabled agent in + `__post_init__` and delegates `capture_response`/`await_completion`'s + polling/the startup wait to it — a pure refactor, not a behavior change + (those call paths are also covered by the pre-existing reliability/ + pane-tail test files). + * `start_workspace` enables agents by iterating the `_ADAPTERS` registry + (not the closed `AGENTS` literal tuple), so a 4th agent becomes + enable-able by registering an adapter object. + +No real docker daemon or tmux required. +""" + +from __future__ import annotations + +import importlib.util +import sys +from dataclasses import dataclass +from pathlib import Path + +import pytest + + +def _load_driver_module(): + here = Path(__file__).resolve() + for ancestor in here.parents: + candidate = ( + ancestor + / "providers" + / "workspaces" + / "interactive-tmux" + / "driver" + / "interactive_tmux.py" + ) + if candidate.is_file(): + spec = importlib.util.spec_from_file_location("interactive_tmux", candidate) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules.setdefault("interactive_tmux", module) + spec.loader.exec_module(module) + return module + pytest.skip("interactive_tmux driver not found in repo layout") + + +driver = _load_driver_module() + + +@dataclass +class _FakeExecutor: + """Records every `exec()` call; returns a configurable canned result.""" + + exit_code: int = 0 + stdout: str = "" + stderr: str = "" + + def __post_init__(self) -> None: + self.calls: list[tuple[tuple, dict]] = [] + + def exec(self, command, *, timeout_s=None, stdin=None): + self.calls.append((tuple(command), {"timeout_s": timeout_s, "stdin": stdin})) + return driver.ExecResult(exit_code=self.exit_code, stdout=self.stdout, stderr=self.stderr) + + +class TestTmuxSessionIsAgentAgnostic: + def test_no_agent_name_referenced_in_class_source(self) -> None: + """`TmuxSession` must know nothing about claude/codex/gemini — only + generic pane/window operations, matching the phase-4 spec.""" + import inspect + + src = inspect.getsource(driver.TmuxSession) + for name in ("claude", "codex", "gemini"): + assert name not in src.lower(), f"TmuxSession source unexpectedly mentions {name!r}" + + def test_send_keys_delegates_to_module_tmux_send_keys( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls = [] + monkeypatch.setattr( + driver, + "_tmux_send_keys", + lambda container, window, *keys, **kw: calls.append((container, window, keys, kw)), + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + session.send_keys("Enter", timeout_s=9.0) + assert calls == [("c1", "win1", ("Enter",), {"executor": fake, "timeout_s": 9.0})] + + def test_send_literal_delegates_to_module_tmux_send_literal( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls = [] + monkeypatch.setattr( + driver, + "_tmux_send_literal", + lambda container, window, text, **kw: calls.append((container, window, text, kw)), + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + session.send_literal("hello world") + assert calls == [ + ( + "c1", + "win1", + "hello world", + {"executor": fake, "timeout_s": driver.DEFAULT_EXEC_TIMEOUT_S}, + ) + ] + + def test_capture_pane_delegates_to_module_tmux_capture( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + driver, + "_tmux_capture", + lambda container, window, **kw: f"{container}:{window}:{kw.get('timeout_s')}", + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + assert session.capture_pane(timeout_s=4.0) == "c1:win1:4.0" + + def test_is_alive_true_when_has_session_succeeds(self) -> None: + fake = _FakeExecutor(exit_code=0) + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + assert session.is_alive() is True + assert fake.calls[0][0] == ("tmux", "has-session", "-t", driver.TMUX_SESSION) + + def test_is_alive_false_when_has_session_fails(self) -> None: + fake = _FakeExecutor(exit_code=1) + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + assert session.is_alive() is False + + def test_start_new_session_for_first_window(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls = [] + monkeypatch.setattr( + driver, + "_docker_exec", + lambda container, *args, **kw: calls.append((container, args, kw)), + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="claude", executor=fake) + session.start(200, 50) + container, args, kw = calls[0] + assert container == "c1" + assert args == ( + "tmux", + "new-session", + "-d", + "-s", + driver.TMUX_SESSION, + "-n", + "claude", + "-x", + "200", + "-y", + "50", + ) + assert kw["executor"] is fake + + def test_start_new_window_for_subsequent_windows(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls = [] + monkeypatch.setattr( + driver, + "_docker_exec", + lambda container, *args, **kw: calls.append((container, args, kw)), + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="codex", executor=fake) + session.start(200, 50, as_new_window=True) + container, args, kw = calls[0] + assert args == ("tmux", "new-window", "-t", driver.TMUX_SESSION, "-n", "codex") + + def test_get_incremental_output_diffs_against_previous( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(driver, "_tmux_capture", lambda container, window, **kw: "abcXYZ") + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + new_text, full = session.get_incremental_output("abc") + assert new_text == "XYZ" + assert full == "abcXYZ" + + def test_get_incremental_output_falls_back_to_full_capture_when_no_common_prefix( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + driver, "_tmux_capture", lambda container, window, **kw: "fresh pane contents" + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + new_text, full = session.get_incremental_output("unrelated prior text") + assert new_text == "fresh pane contents" + assert full == "fresh pane contents" + + def test_get_incremental_output_with_no_previous_returns_full_capture_as_new( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + driver, "_tmux_capture", lambda container, window, **kw: "first capture" + ) + fake = _FakeExecutor() + session = driver.TmuxSession(target="c1", window="win1", executor=fake) + new_text, full = session.get_incremental_output(None) + assert new_text == "first capture" + assert full == "first capture" + + +class TestWorkspaceBuildsSessionsPerAgent: + def test_post_init_builds_one_session_per_enabled_agent(self) -> None: + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude", "codex"), + ) + assert set(ws._sessions) == {"claude", "codex"} + assert ws._sessions["claude"].target == "test-container" + assert ws._sessions["claude"].window == "claude" + assert ws._sessions["claude"].executor is ws.executor + + def test_capture_response_delegates_to_the_agent_session( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls = [] + monkeypatch.setattr( + driver, + "_tmux_capture", + lambda container, window, **kw: calls.append((container, window, kw)) or "pane text", + ) + ws = driver.InteractiveTmuxWorkspace( + name="test", + container="test-container", + image="test-image", + workdir="/workspace", + tmux_size=(200, 50), + host_throwaway_dir=Path("/tmp/test-throwaway"), + enabled_agents=("claude",), + ) + result = ws.capture_response("claude", timeout_s=6.0) + assert result == "pane text" + assert calls == [("test-container", "claude", {"executor": ws.executor, "timeout_s": 6.0})] + + +class TestAdapterRegistryDrivesEnablement: + def test_start_workspace_iterates_adapter_registry_not_closed_agents_tuple(self) -> None: + """A 4th agent becomes enable-able by registering it in `_ADAPTERS` + alone — `start_workspace`'s enablement loop must not be hardcoded + to the `AGENTS` literal tuple.""" + import inspect + + src = inspect.getsource(driver.InteractiveTmuxWorkspace.start_workspace.__func__) + assert "for agent in _ADAPTERS" in src diff --git a/providers/workspaces/interactive-tmux/driver/interactive_tmux.py b/providers/workspaces/interactive-tmux/driver/interactive_tmux.py index 7eb51568..615f1c9a 100644 --- a/providers/workspaces/interactive-tmux/driver/interactive_tmux.py +++ b/providers/workspaces/interactive-tmux/driver/interactive_tmux.py @@ -44,12 +44,15 @@ from __future__ import annotations import argparse +import base64 import json import logging import os +import posixpath import re import shlex import shutil +import stat import subprocess import sys import tempfile @@ -58,7 +61,7 @@ from collections.abc import Sequence from dataclasses import asdict, dataclass, field from pathlib import Path -from typing import Literal +from typing import Literal, Protocol, runtime_checkable logger = logging.getLogger(__name__) @@ -72,6 +75,21 @@ DEFAULT_TMUX_SIZE = (200, 50) TMUX_SESSION = "agents" +# --------------------------------------------------------------------------- +# Phase 3 (reliability) constants +# +# Every subprocess.run/docker-exec call used to be unbounded — a wedged +# container (or a docker daemon that stops responding) could hang the +# calling process forever. These bound every such call; `await_completion`'s +# own overall deadline stays separate (see `poll_timeout_s` below) so a +# single stuck poll can't eat the whole budget silently. +DEFAULT_EXEC_TIMEOUT_S = 15.0 # bound for one docker-exec/tmux operation +DEFAULT_RUN_TIMEOUT_S = 30.0 # bound for `docker run` / `docker rm -f` + +# tmux `send-keys -l` caps payloads around 16KB; above this we stage the +# text via `load-buffer` + `paste-buffer` instead (see `_tmux_send_literal`). +TMUX_SEND_KEYS_MAX_BYTES = 12_000 + # Claude readiness — empty `❯ ` prompt line (whitespace tolerated). Pre- # compiled because await_completion polls this 2x/sec per agent. _CLAUDE_EMPTY_PROMPT_RE = re.compile(r"^❯\s*$", re.MULTILINE) @@ -99,6 +117,7 @@ class AwaitResult: - timed_out=True, ready=False → deadline hit before idle - ready=False, reason="never_ready" → never reached even one ready frame - ready=False, reason="unstable" → ready frames seen but pane kept changing + - ready=False, reason="container_dead"→ target died mid-poll (not a timeout) - ready=True → adapter is_ready() held stable `pane` carries the last captured pane text so callers don't have to @@ -107,7 +126,7 @@ class AwaitResult: ready: bool timed_out: bool - reason: str # "ready" | "timeout_never_ready" | "timeout_unstable" | "error" + reason: str # "ready" | "timeout_never_ready" | "timeout_unstable" | "container_dead" | "error" duration_ms: float stable_polls_observed: int pane: str = "" @@ -139,6 +158,399 @@ def __init__(self, startup_status: dict[str, AwaitResult]): self.startup_status = startup_status +# --------------------------------------------------------------------------- +# Executor seam (Phase 2, ADR-driven docker-out-of-docker fix) +# +# Every tmux operation (send-keys, capture-pane) and every credential-seeding +# file write ultimately needs to run a command *inside* the workspace +# target (container, VM, SSH host, ...). Historically that meant +# `subprocess.run(["docker", "exec", ...])` sprinkled through this module. +# `CommandExecutor` pulls that behind a small Protocol so: +# 1. tests can inject a fake executor instead of monkeypatching subprocess +# at the module level; +# 2. a future transport (E2B, a remote agent, SSH/VPS, ...) can implement +# the same one-method surface without touching the tmux/adapter logic +# above. +# `DockerExecExecutor` is the only implementation today and preserves the +# exact `docker exec ` behavior this module always +# had. All identifiers threaded through the tmux/exec helpers below are +# named `target` (not `container`) so they read as backend-neutral: for +# Docker it holds the container name; for other backends (see the +# `Environment` seam further below) it holds whatever opaque label that +# backend uses for logging. + + +@dataclass +class ExecResult: + """Result of running one command inside a workspace container. + + `timed_out` (Phase 3) is set by `DockerExecExecutor.exec()` when the + underlying `subprocess.run(..., timeout=...)` call itself expired, + instead of letting `subprocess.TimeoutExpired` propagate and hang the + caller's stack. Defaults to `False` so existing call sites that only + ever cared about `exit_code`/`stdout`/`stderr` are unaffected. + """ + + exit_code: int + stdout: str + stderr: str + timed_out: bool = False + + +def _decode(stream: str | bytes | None) -> str: + """Normalize a subprocess stdout/stderr stream to `str`. + + When an executor pipes bytes over `stdin` it must run `subprocess.run` + with `text=False` (you cannot pass `bytes` as `input` while `text=True`), + so `stdout`/`stderr` come back as `bytes`. This decodes them (and the + `str` case is a passthrough) so every `ExecResult` carries text + regardless of whether the call used stdin. + """ + if stream is None: + return "" + if isinstance(stream, bytes): + return stream.decode("utf-8", errors="replace") + return stream + + +@runtime_checkable +class CommandExecutor(Protocol): + def exec( + self, + command: Sequence[str], + *, + timeout_s: float | None = None, + stdin: bytes | None = None, + ) -> ExecResult: + """Run `command` inside the workspace target. + + `stdin`, when given, is fed to the process as its standard input. + This is the credential-safe transfer path: bytes passed here never + appear in the process argv (which is world-readable via `ps` / + `/proc//cmdline`), unlike embedding them in a shell command. + """ + ... + + +@dataclass +class DockerExecExecutor: + """Default `CommandExecutor`: shells out to `docker exec ...`. + + Behavior-preserving extraction of what `_docker_exec` always did; the + only new capability is `timeout_s`, forwarded to `subprocess.run(..., + timeout=...)` so a hung `docker exec` can't block forever (a bare + `subprocess.run` with no timeout blocks indefinitely on a wedged + container). + """ + + target: str + + def exec( + self, + command: Sequence[str], + *, + timeout_s: float | None = None, + stdin: bytes | None = None, + ) -> ExecResult: + # `-i` (keep STDIN open) is required for `docker exec` to forward the + # piped `stdin` bytes to the in-container process; without it the + # payload is dropped. Only added when stdin is supplied so the + # non-stdin argv shape (and its tests) stay byte-for-byte identical. + interactive = ["-i"] if stdin is not None else [] + cmd = ["docker", "exec", *interactive, self.target, *command] + try: + proc = subprocess.run( + cmd, + capture_output=True, + # bytes `input` requires text=False; decode outputs via `_decode`. + text=stdin is None, + timeout=timeout_s, + input=stdin, + ) + except subprocess.TimeoutExpired as exc: + # Phase 3: a wedged `docker exec` used to hang the calling + # thread forever. Return a `timed_out` ExecResult instead of + # letting the exception propagate, so pollers (await_completion) + # can treat it as "not ready yet" and keep going within their + # own overall deadline. + partial_err = _decode(exc.stderr) + return ExecResult( + exit_code=-1, + stdout=_decode(exc.stdout), + stderr=(f"docker exec timed out after {timeout_s}s" + (f": {partial_err}" if partial_err else "")), + timed_out=True, + ) + return ExecResult( + exit_code=proc.returncode, stdout=_decode(proc.stdout), stderr=_decode(proc.stderr) + ) + + +# --------------------------------------------------------------------------- +# Environment seam (provisioning, one layer above `CommandExecutor`) +# +# `CommandExecutor` answers "how do I run one command against an already- +# running target?". `Environment` answers the layer above that: "how do I +# bring a target into existence, and get a `CommandExecutor` for it, and +# tear it down later?". Docker is the only implementation today +# (`DockerEnvironment`, extracted behavior-preserving from what +# `start_workspace`/`stop` always did), but the seam exists so Local and +# SSH/VPS backends can be added as day-one alternatives without touching +# any tmux/adapter/workspace logic — they only need to provision *something* +# that a `CommandExecutor` can run commands against. + + +@runtime_checkable +class Environment(Protocol): + def start(self) -> CommandExecutor: ... + def stop(self) -> None: ... + + +@dataclass +class DockerEnvironment: + """Default `Environment`: provisions a workspace via `docker run` / + `docker rm -f`. + + Behavior-preserving extraction of the `docker run ...` / `docker rm -f + ...` logic `start_workspace`/`stop` always ran inline. `start()` returns + a `DockerExecExecutor(target=self.name)` bound to the container it just + created; `stop()` best-effort removes that same container. + """ + + name: str + image: str + workdir: str + run_timeout_s: float | None = DEFAULT_RUN_TIMEOUT_S + + def start(self) -> CommandExecutor: + run_cmd = [ + "docker", + "run", + "-d", + "--name", + self.name, + "--workdir", + self.workdir, + self.image, + "sleep", + "infinity", + ] + _run(run_cmd, timeout_s=self.run_timeout_s) + return DockerExecExecutor(self.name) + + def stop(self) -> None: + try: + subprocess.run( + ["docker", "rm", "-f", self.name], + check=False, + capture_output=True, + timeout=self.run_timeout_s or DEFAULT_RUN_TIMEOUT_S, + ) + except subprocess.TimeoutExpired: + logger.warning("docker rm -f %s timed out during DockerEnvironment.stop()", self.name) + + +@dataclass +class LocalExecutor: + """`CommandExecutor` that runs argv directly on the host — no docker + prefix, no target at all. Mirrors `DockerExecExecutor.exec()`'s + timeout/error handling exactly so callers see identical `ExecResult` + shapes regardless of which `Environment` backs them.""" + + workdir: str | Path + + def exec( + self, + command: Sequence[str], + *, + timeout_s: float | None = None, + stdin: bytes | None = None, + ) -> ExecResult: + try: + proc = subprocess.run( + command, + capture_output=True, + # bytes `input` requires text=False; decode outputs via `_decode`. + text=stdin is None, + timeout=timeout_s, + cwd=self.workdir, + input=stdin, + ) + except subprocess.TimeoutExpired as exc: + # Same rationale as DockerExecExecutor: a wedged local command + # must not hang the caller forever; return a `timed_out` + # ExecResult so pollers can treat it as "not ready yet". + partial_err = _decode(exc.stderr) + return ExecResult( + exit_code=-1, + stdout=_decode(exc.stdout), + stderr=(f"local exec timed out after {timeout_s}s" + (f": {partial_err}" if partial_err else "")), + timed_out=True, + ) + return ExecResult( + exit_code=proc.returncode, stdout=_decode(proc.stdout), stderr=_decode(proc.stderr) + ) + + +@dataclass +class LocalEnvironment: + """`Environment` that runs directly on the host — no container, no SSH + hop. The host is already "up" by construction (it's the machine this + process is running on), so `start()` has no provisioning step of its + own; it only verifies the tools the driver depends on are present and + returns a `LocalExecutor` bound to `workdir`. + + `stop()` is a no-op: there is nothing this environment created that it + would need to tear down (unlike `DockerEnvironment`, which must `docker + rm -f` the container it `docker run`'d). Any tmux session left running + on the host outliving the `LocalEnvironment` object is intentional — + the host itself is not owned by this class the way a container is. + """ + + workdir: str | Path + require_tools: Sequence[str] = ("tmux",) + + def start(self) -> CommandExecutor: + missing = [tool for tool in self.require_tools if shutil.which(tool) is None] + if missing: + raise RuntimeError( + "LocalEnvironment.start(): required tool(s) not found on PATH: " + f"{', '.join(missing)}. Install them before starting a local " + "interactive-tmux workspace." + ) + return LocalExecutor(self.workdir) + + def stop(self) -> None: + # No-op: see class docstring — a local environment doesn't own the + # host, so there is nothing to tear down. + return None + + +@dataclass +class SSHExecutor: + """`CommandExecutor` that runs argv on a remote host over `ssh`. + + `base_argv` is the full `ssh` invocation up to (but not including) the + remote command itself — e.g. `["ssh", "-o", "BatchMode=yes", "-o", + "ConnectTimeout=10", "-i", "/key", "-p", "22", "user@host"]`. `exec()` + joins `command` into a single shell string (via `shlex.join`, optionally + prefixed with `cd &&`) and appends it as the final argv + element, mirroring `DockerExecExecutor`/`LocalExecutor`'s timeout and + `ExecResult` handling exactly so callers see identical result shapes + regardless of backend. + + Exit-code semantics are intentionally *not* special-cased beyond what + the other two executors do: `ssh` returns 255 for a connection-level + failure (vs. the remote command's own exit code otherwise), but that + raw exit code is simply surfaced via `ExecResult.exit_code` like any + other — callers that care about the distinction can inspect `stderr`. + """ + + base_argv: list[str] + workdir: str | None = None + + def exec( + self, + command: Sequence[str], + *, + timeout_s: float | None = None, + stdin: bytes | None = None, + ) -> ExecResult: + remote_cmd = shlex.join(command) + if self.workdir: + remote_cmd = f"cd {shlex.quote(self.workdir)} && {remote_cmd}" + cmd = [*self.base_argv, remote_cmd] + # `ssh` forwards our local stdin to the remote command with no extra + # flag needed, so `input=stdin` reaches the remote process directly. + try: + proc = subprocess.run( + cmd, + capture_output=True, + # bytes `input` requires text=False; decode outputs via `_decode`. + text=stdin is None, + timeout=timeout_s, + input=stdin, + ) + except subprocess.TimeoutExpired as exc: + # Same rationale as DockerExecExecutor/LocalExecutor: a wedged + # remote command must not hang the caller forever. + partial_err = _decode(exc.stderr) + return ExecResult( + exit_code=-1, + stdout=_decode(exc.stdout), + stderr=(f"ssh exec timed out after {timeout_s}s" + (f": {partial_err}" if partial_err else "")), + timed_out=True, + ) + return ExecResult( + exit_code=proc.returncode, stdout=_decode(proc.stdout), stderr=_decode(proc.stderr) + ) + + +@dataclass +class SSHEnvironment: + """`Environment` that provisions a workspace target on a remote host + reachable over `ssh`. + + Unlike `DockerEnvironment` there is no `docker run` step: the "target" + is simply the remote host itself, assumed already up. `start()` + fails fast with a reachability check (`ssh ... true`) rather than + letting the first tmux command discover a dead host mid-session, then + returns an `SSHExecutor` bound to the `ssh` argv it just proved works. + + Known limitation / follow-up: this simple version opens a fresh `ssh` + process per `exec()` call (no persistent connection is held open + between execs). A future version could hold one persistent `ssh + ControlMaster` connection for lower per-call latency. + """ + + host: str + user: str + key_path: str | Path | None = None + port: int = 22 + workdir: str | None = None + connect_timeout_s: float = 10.0 + + def _base_argv(self) -> list[str]: + return [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + f"ConnectTimeout={int(self.connect_timeout_s)}", + *(["-i", str(self.key_path)] if self.key_path else []), + "-p", + str(self.port), + "--", + f"{self.user}@{self.host}", + ] + + def start(self) -> CommandExecutor: + base_argv = self._base_argv() + try: + proc = subprocess.run( + [*base_argv, "true"], + capture_output=True, + text=True, + timeout=self.connect_timeout_s, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + f"SSHEnvironment.start(): reachability check to {self.user}@{self.host}:{self.port} " + f"timed out after {self.connect_timeout_s}s" + ) from exc + if proc.returncode != 0: + raise RuntimeError( + f"SSHEnvironment.start(): reachability check to {self.user}@{self.host}:{self.port} " + f"failed (exit {proc.returncode}): {proc.stderr.strip()}" + ) + return SSHExecutor(base_argv, workdir=self.workdir) + + def stop(self) -> None: + # No-op in this simple version: no persistent connection is held + # open between execs, so there is nothing to tear down. See class + # docstring for the ControlMaster follow-up. + return None + + # --------------------------------------------------------------------------- # tmux send-keys helpers (the only place that talks to docker exec tmux) @@ -161,36 +573,146 @@ def _redact_cmd(cmd: list[str]) -> str: return " ".join(parts) -def _run(cmd: list[str], check: bool = True, capture: bool = True) -> subprocess.CompletedProcess: +def _run( + cmd: list[str], + check: bool = True, + capture: bool = True, + timeout_s: float | None = DEFAULT_RUN_TIMEOUT_S, +) -> subprocess.CompletedProcess: """Run a subprocess; return CompletedProcess. Raises on non-zero unless - `check=False`.""" + `check=False`. + + Phase 3: bounded by `timeout_s` (default `DEFAULT_RUN_TIMEOUT_S`) so a + wedged `docker run`/`docker rm` can't block the caller forever; + `subprocess.TimeoutExpired` propagates (bounded, not silent) same as + any other subprocess failure. + """ logger.debug("exec: %s", _redact_cmd(cmd)) return subprocess.run( cmd, check=check, capture_output=capture, text=True, + timeout=timeout_s, ) -def _docker_exec(container: str, *args: str, check: bool = True) -> subprocess.CompletedProcess: - return _run(["docker", "exec", container, *args], check=check) - - -def _tmux_send_keys(container: str, window: str, *keys: str) -> None: - target = f"{TMUX_SESSION}:{window}" - _docker_exec(container, "tmux", "send-keys", "-t", target, *keys) - - -def _tmux_send_literal(container: str, window: str, text: str) -> None: - """Send `text` byte-for-byte (no special-key interpretation).""" - target = f"{TMUX_SESSION}:{window}" - # `--` ends option parsing so a prompt beginning with `-` (e.g. "-R", - # "--help") is treated as literal text, not a tmux send-keys flag. - _docker_exec(container, "tmux", "send-keys", "-t", target, "-l", "--", text) - +def _docker_exec( + target: str, + *args: str, + check: bool = True, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> subprocess.CompletedProcess: + """Run `docker exec `, via `executor` if given. + + Defaults to constructing a fresh `DockerExecExecutor(target)` so + every existing call site (which doesn't know about the executor seam) + behaves exactly as before. Returns a `subprocess.CompletedProcess` for + backward compatibility with callers that inspect `.stdout` / + `.returncode`. + + Phase 3: `timeout_s` (default `DEFAULT_EXEC_TIMEOUT_S`) is forwarded to + the executor so no `docker exec` call in this module can block forever. + + `target` is a label only when `executor` is supplied (it feeds the + logged/returned `docker exec ...` command shape, but the + actual command runs through `executor.exec()`, which may not be + Docker-backed at all) — see the `Environment` seam above. + """ + cmd = ["docker", "exec", target, *args] + logger.debug("exec: %s", _redact_cmd(cmd)) + exec_ = executor or DockerExecExecutor(target) + result = exec_.exec(list(args), timeout_s=timeout_s) + if check and result.exit_code != 0: + raise subprocess.CalledProcessError( + result.exit_code, cmd, result.stdout, result.stderr + ) + return subprocess.CompletedProcess(cmd, result.exit_code, result.stdout, result.stderr) + + +def _tmux_send_keys( + target: str, + window: str, + *keys: str, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> None: + pane = f"{TMUX_SESSION}:{window}" + _docker_exec(target, "tmux", "send-keys", "-t", pane, *keys, executor=executor, timeout_s=timeout_s) + + +def _tmux_send_literal( + target: str, + window: str, + text: str, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> None: + """Send `text` byte-for-byte (no special-key interpretation). + + Phase 3: tmux's `send-keys -l` caps payloads around 16KB — a long + model prompt (or a pasted file) silently truncates past that ceiling. + Payloads at or under `TMUX_SEND_KEYS_MAX_BYTES` keep using the small, + fast `send-keys -l` path unchanged; larger payloads are staged into a + tmux paste buffer instead: the bytes are written into a container-side + temp file (base64-chunked over the executor, same mechanism the + credential transfer uses), loaded into a named tmux buffer with + `load-buffer`, and dispatched into the target pane with `paste-buffer`. + """ + pane = f"{TMUX_SESSION}:{window}" + payload = text.encode("utf-8") + if len(payload) <= TMUX_SEND_KEYS_MAX_BYTES: + # `--` ends option parsing so a prompt beginning with `-` (e.g. + # "-R", "--help") is treated as literal text, not a send-keys flag. + # + # NOTE on multiline payloads: `send-keys -l` emits the literal bytes, + # so an embedded newline is delivered as a bare Enter. The per-agent + # adapters (`_ClaudeAdapter.submit` et al.) deliberately send the body + # here and the terminating Enter as a SEPARATE key, so a multiline + # prompt whose newlines act as line breaks (not submits) is the + # agent TUI's own paste/soft-wrap behavior - consistent with the + # bracketed-paste path below for oversized payloads. + _docker_exec( + target, "tmux", "send-keys", "-t", pane, "-l", "--", text, + executor=executor, timeout_s=timeout_s, + ) + return -def _tmux_capture(container: str, window: str) -> str: + exec_ = executor or DockerExecExecutor(target) + token = uuid.uuid4().hex + buf_path = f"/tmp/.itmux-sendkeys-{token}.buf" + buf_name = f"itmux-{token[:12]}" + try: + _write_bytes_to_container(exec_, buf_path, payload, timeout_s=timeout_s) + _run_exec_checked( + exec_, ["tmux", "load-buffer", "-b", buf_name, buf_path], timeout_s=timeout_s + ) + # `-p` sends the buffer as a BRACKETED paste: tmux wraps it in the + # terminal's paste markers (ESC[200~ ... ESC[201~) so the agent TUI + # (claude/codex/gemini all enable bracketed-paste mode) treats the + # whole payload as one atomic paste. Without it, a multiline prompt's + # embedded newlines dispatch as individual Enter presses and submit + # the prompt early, fragmenting it across turns. + _run_exec_checked( + exec_, + ["tmux", "paste-buffer", "-p", "-b", buf_name, "-d", "-t", pane], + timeout_s=timeout_s, + ) + finally: + # Best-effort cleanup of the staged temp file; a failure here must + # not mask the paste having already succeeded (or failed) above. + exec_.exec(["rm", "-f", buf_path], timeout_s=timeout_s) + + +def _tmux_capture( + target: str, + window: str, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> str: """Capture the full pane buffer including scrollback. `-S -` = start at the top of the history; `-E -` = end at the bottom @@ -203,9 +725,10 @@ def _tmux_capture(container: str, window: str) -> str: the Python driver shipped without it (D-block-3 from the Syntropic137 stress run, experiments/stress/STRESS-REPORT.md). """ - target = f"{TMUX_SESSION}:{window}" + pane = f"{TMUX_SESSION}:{window}" return _docker_exec( - container, "tmux", "capture-pane", "-p", "-t", target, "-S", "-", "-E", "-" + target, "tmux", "capture-pane", "-p", "-t", pane, "-S", "-", "-E", "-", + executor=executor, timeout_s=timeout_s, ).stdout @@ -239,6 +762,136 @@ def _pane_tail(pane_text: str, n_lines: int = DEFAULT_TMUX_SIZE[1]) -> str: return tail +# --------------------------------------------------------------------------- +# TmuxSession (Phase 4: agent-agnostic session/window handle) +# +# Everything above this point (`_tmux_send_keys`, `_tmux_send_literal`, +# `_tmux_capture`) is already agent-agnostic — it operates on a +# `(container, window)` pair with no claude/codex/gemini knowledge. Before +# Phase 4, that genericity was implicit: callers reached for the bare +# functions directly. `TmuxSession` makes it an explicit, reusable handle so +# a 4th agent adapter can be built on top of it without touching this class, +# and so `InteractiveTmuxWorkspace` has one object per enabled agent to hold +# session state (rather than re-threading `target`/`window`/`executor` +# through every call). Must not reference any agent name — only generic +# pane/window operations, all routed through the injected `CommandExecutor`. +# `target` is backend-neutral: whatever `Environment.start()` produced an +# executor for (a Docker container name today; an opaque host label for +# other backends). + + +@dataclass +class TmuxSession: + """One tmux window inside the shared `TMUX_SESSION`, agent-agnostic. + + Thin, stateless-except-for-identity wrapper around the module-level + `_tmux_send_keys` / `_tmux_send_literal` / `_tmux_capture` / `_docker_exec` + helpers. Deliberately delegates to those free functions (rather than + reimplementing their logic) so: + 1. existing tests that monkeypatch the module-level functions keep + working unchanged whether a caller goes through `TmuxSession` or + calls the free functions directly; + 2. Phase 2/3 behavior (executor injection, timeouts, payload batching) + is inherited for free instead of duplicated. + """ + + target: str + window: str + executor: CommandExecutor + + def start( + self, + cols: int, + rows: int, + *, + as_new_window: bool = False, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> None: + """Create this window: a new tmux session (first agent) or a new + window in the existing session (subsequent agents).""" + if as_new_window: + _docker_exec( + self.target, + "tmux", + "new-window", + "-t", + TMUX_SESSION, + "-n", + self.window, + executor=self.executor, + timeout_s=timeout_s, + ) + else: + _docker_exec( + self.target, + "tmux", + "new-session", + "-d", + "-s", + TMUX_SESSION, + "-n", + self.window, + "-x", + str(cols), + "-y", + str(rows), + executor=self.executor, + timeout_s=timeout_s, + ) + + def stop(self, *, timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S) -> None: + """Best-effort kill of this window. Non-fatal: tearing down the + container (the workspace-level `stop()`) removes the window anyway; + this exists for callers that want to retire one agent's pane + without stopping the whole workspace.""" + pane = f"{TMUX_SESSION}:{self.window}" + _docker_exec( + self.target, + "tmux", + "kill-window", + "-t", + pane, + executor=self.executor, + check=False, + timeout_s=timeout_s, + ) + + def send_keys(self, *keys: str, timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S) -> None: + _tmux_send_keys(self.target, self.window, *keys, executor=self.executor, timeout_s=timeout_s) + + def send_literal(self, text: str, *, timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S) -> None: + _tmux_send_literal(self.target, self.window, text, executor=self.executor, timeout_s=timeout_s) + + def capture_pane(self, *, timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S) -> str: + return _tmux_capture(self.target, self.window, executor=self.executor, timeout_s=timeout_s) + + def get_incremental_output( + self, + previous: str | None, + *, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> tuple[str, str]: + """Return `(new_text, full_pane)` for this window. + + `new_text` is the substring diff against `previous` (an earlier + `capture_pane()`/`get_incremental_output()` result) when the current + capture starts with it — the common case, since tmux scrollback only + grows. When it doesn't (pane was cleared, history rolled off the + scrollback limit, or `previous` is falsy), `new_text` falls back to + the full current capture rather than guessing at a diff. + """ + current = self.capture_pane(timeout_s=timeout_s) + if previous and current.startswith(previous): + return current[len(previous) :], current + return current, current + + def is_alive(self, *, timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S) -> bool: + """Whether the shared tmux session (not just this window) is still + reachable inside the container.""" + result = self.executor.exec(["tmux", "has-session", "-t", TMUX_SESSION], timeout_s=timeout_s) + return result.exit_code == 0 + + # --------------------------------------------------------------------------- # Per-agent adapters # @@ -377,6 +1030,9 @@ def launch_in_window( container: str, _workdir: str, plugin_dirs: Sequence[Path] | None = None, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, ) -> None: """Start `claude` in its tmux window, with optional `--plugin-dir` flags. @@ -391,10 +1047,29 @@ def launch_in_window( if plugin_dirs: flags = " ".join(f"--plugin-dir {shlex.quote(str(p))}" for p in plugin_dirs) cmd = f"claude {flags}" - _tmux_send_literal(container, _ClaudeAdapter.window, cmd) - _tmux_send_keys(container, _ClaudeAdapter.window, "Enter") + _tmux_send_literal( + container, + _ClaudeAdapter.window, + cmd, + executor=executor, + timeout_s=timeout_s, + ) + _tmux_send_keys( + container, + _ClaudeAdapter.window, + "Enter", + executor=executor, + timeout_s=timeout_s, + ) else: - _tmux_send_keys(container, _ClaudeAdapter.window, "claude", "Enter") + _tmux_send_keys( + container, + _ClaudeAdapter.window, + "claude", + "Enter", + executor=executor, + timeout_s=timeout_s, + ) @staticmethod def build_launch_command(plugin_dirs: Sequence[Path] | None = None) -> str: @@ -410,12 +1085,18 @@ def build_launch_command(plugin_dirs: Sequence[Path] | None = None) -> str: return f"claude {flags}" @staticmethod - def submit(container: str, text: str) -> None: + def submit( + container: str, + text: str, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> None: # EXP-01: two-step is the documented default. -l makes the bytes # land literally (no special-key interpretation in the text body), # then a separate Enter dispatches. - _tmux_send_literal(container, _ClaudeAdapter.window, text) - _tmux_send_keys(container, _ClaudeAdapter.window, "Enter") + _tmux_send_literal(container, _ClaudeAdapter.window, text, executor=executor, timeout_s=timeout_s) + _tmux_send_keys(container, _ClaudeAdapter.window, "Enter", executor=executor, timeout_s=timeout_s) @staticmethod def is_ready(pane_text: str) -> bool: @@ -484,17 +1165,27 @@ def prepare_host_auth( raise FileNotFoundError(f"codex auth dir not found: {host_src}") dst_dir = ctx.host_throwaway_dir / "codex.dir" dst_dir.mkdir(parents=True, exist_ok=True) - # Copy the .codex/ tree but skip the live tmp/ subdir — codex races + # Copy the .codex/ tree but skip the live tmp/ subdir - codex races # there (creates and deletes argv files during normal operation), so # copytree against it sees vanished files. The auth lives in - # auth.json / config.toml / sessions/ at the top level. - skip = {"tmp", "log", "logs"} + # auth.json / config.toml / sessions/ at the top level. `plugins/` is + # a large plugin/dependency cache (node_modules), never auth, and + # staging it turns start_workspace into a multi-minute, thousands-of- + # exec crawl - skip it (node_modules/.git anywhere are skipped by the + # copytree ignore filter as defense in depth). + skip = {"tmp", "log", "logs", "plugins"} for item in host_src.iterdir(): if item.name in skip: continue target = dst_dir / item.name if item.is_dir(): - shutil.copytree(item, target, dirs_exist_ok=True, ignore_dangling_symlinks=True) + shutil.copytree( + item, + target, + dirs_exist_ok=True, + ignore_dangling_symlinks=True, + ignore=_ignore_uncopyable, + ) else: shutil.copy2(item, target) _chown_recursive(dst_dir, 1000, 1000) @@ -505,29 +1196,58 @@ def launch_in_window( container: str, _workdir: str, plugin_dirs: Sequence[Path] | None = None, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, ) -> None: # `plugin_dirs` is accepted for signature parity with the claude # adapter; codex has no equivalent `--plugin-dir` flag, so any # value is silently ignored. del plugin_dirs # --no-alt-screen so capture-pane sees the same buffer the TUI uses. - _tmux_send_keys(container, _CodexAdapter.window, "codex --no-alt-screen", "Enter") + _tmux_send_keys( + container, + _CodexAdapter.window, + "codex --no-alt-screen", + "Enter", + executor=executor, + timeout_s=timeout_s, + ) # Trust banner: select option 1 ("Yes, trust"), confirm with Enter. time.sleep(2) - _tmux_send_keys(container, _CodexAdapter.window, "1", "Enter") + _tmux_send_keys( + container, + _CodexAdapter.window, + "1", + "Enter", + executor=executor, + timeout_s=timeout_s, + ) # Hooks-review modal: close with Escape. time.sleep(1) - _tmux_send_keys(container, _CodexAdapter.window, "Escape") + _tmux_send_keys( + container, + _CodexAdapter.window, + "Escape", + executor=executor, + timeout_s=timeout_s, + ) time.sleep(1) @staticmethod - def submit(container: str, text: str) -> None: + def submit( + container: str, + text: str, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> None: # EXP-02: literal text first (so the body's bytes don't get # tmux-special-key-interpreted), then C-j C-m to dispatch. # C-j C-m is the gotcha — bare C-m alone often does not submit # the first message. - _tmux_send_literal(container, _CodexAdapter.window, text) - _tmux_send_keys(container, _CodexAdapter.window, "C-j", "C-m") + _tmux_send_literal(container, _CodexAdapter.window, text, executor=executor, timeout_s=timeout_s) + _tmux_send_keys(container, _CodexAdapter.window, "C-j", "C-m", executor=executor, timeout_s=timeout_s) @staticmethod def is_ready(pane_text: str) -> bool: @@ -579,7 +1299,13 @@ def prepare_host_auth( for item in host_src.iterdir(): target = dst_dir / item.name if item.is_dir(): - shutil.copytree(item, target, dirs_exist_ok=True) + shutil.copytree( + item, + target, + dirs_exist_ok=True, + ignore_dangling_symlinks=True, + ignore=_ignore_uncopyable, + ) else: shutil.copy2(item, target) # Patch settings.json with folderTrust.enabled=false (EXP-03 fix). @@ -602,19 +1328,35 @@ def launch_in_window( container: str, _workdir: str, plugin_dirs: Sequence[Path] | None = None, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, ) -> None: # `plugin_dirs` is accepted for signature parity with the claude # adapter; gemini has no equivalent `--plugin-dir` flag, so any # value is silently ignored. del plugin_dirs - _tmux_send_keys(container, _GeminiAdapter.window, "gemini", "Enter") + _tmux_send_keys( + container, + _GeminiAdapter.window, + "gemini", + "Enter", + executor=executor, + timeout_s=timeout_s, + ) time.sleep(1) @staticmethod - def submit(container: str, text: str) -> None: + def submit( + container: str, + text: str, + *, + executor: CommandExecutor | None = None, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> None: # EXP-03: text first, then Enter — never C-m. - _tmux_send_literal(container, _GeminiAdapter.window, text) - _tmux_send_keys(container, _GeminiAdapter.window, "Enter") + _tmux_send_literal(container, _GeminiAdapter.window, text, executor=executor, timeout_s=timeout_s) + _tmux_send_keys(container, _GeminiAdapter.window, "Enter", executor=executor, timeout_s=timeout_s) @staticmethod def is_ready(pane_text: str) -> bool: @@ -647,6 +1389,14 @@ def response_marker() -> str: "codex": _CodexAdapter, "gemini": _GeminiAdapter, } +# Phase 4: `_ADAPTERS` is the registry a 4th agent joins by registering an +# adapter object here — no edits to `TmuxSession` (which knows nothing about +# any agent name) or to the workspace's dispatch logic are needed. Adapters +# sit ON TOP of `TmuxSession`: they encode submit/readiness heuristics and +# receive a plain `container` + `executor` (matching their existing static +# method signatures, preserved for backward compatibility with callers/tests +# that invoke them directly), while `InteractiveTmuxWorkspace` holds one +# `TmuxSession` per enabled agent for generic pane operations. # --------------------------------------------------------------------------- @@ -670,6 +1420,169 @@ def _chown_recursive(path: Path, uid: int, gid: int) -> None: _chown_path(sub, uid, gid) +# Directory names that are never credential material but are large and slow +# to stage (thousands of files) and can contain uncopyable special files. +# `.git` is where a live `fsmonitor--daemon.ipc` socket lives; `node_modules` +# and plugin caches balloon a stage into thousands of per-file `docker exec` +# calls (observed: a 4-minute codex stage that overloaded the docker daemon). +_NON_AUTH_DIR_NAMES = frozenset({".git", "node_modules"}) + + +def _ignore_uncopyable(src: str, names: list[str]) -> set[str]: + """`shutil.copytree` ignore filter for credential staging. + + Skips two classes of entry that must never be copied into an auth stage: + + * Non-copyable special files (sockets/FIFOs/devices) - most commonly a git + `fsmonitor--daemon.ipc` Unix socket left by a running fsmonitor daemon in + a `.git/` under the tree. `copy2` raises "Operation not supported on + socket" on those and aborts the whole stage. + * Heavy, never-auth directories (`.git`, `node_modules`) at any depth. + These are pure bloat for credential staging and turn it into thousands + of per-file `docker exec` calls that can overload the docker daemon. + """ + skip: set[str] = set() + for name in names: + if name in _NON_AUTH_DIR_NAMES: + skip.add(name) + continue + try: + mode = os.lstat(os.path.join(src, name)).st_mode + except OSError: + skip.add(name) + continue + if not (stat.S_ISREG(mode) or stat.S_ISDIR(mode) or stat.S_ISLNK(mode)): + skip.add(name) + return skip + + +def _run_exec_checked( + executor: CommandExecutor, + command: list[str], + *, + timeout_s: float | None = None, + stdin: bytes | None = None, + label: str | None = None, +) -> ExecResult: + """`executor.exec(command)`, raising `RuntimeError` on non-zero exit. + + Credential transfer is not optional best-effort work — a silently + failed `mkdir -p` or base64 write leaves the container half-seeded + with auth material, which is worse than failing loudly at + `start_workspace` time. + + `stdin`, when given, is piped to the command so a credential payload + can be delivered without ever appearing in argv (see + `_write_bytes_to_container`). + + `label` is a redacted description used in the raised error message in + place of the raw `command`. Credential-seeding callers pass a label + (e.g. "write credentials to ") so that neither the payload nor + even the raw command shape leaks into exceptions or logs. + """ + result = executor.exec(command, timeout_s=timeout_s, stdin=stdin) + if result.exit_code != 0: + described = label if label is not None else repr(command) + raise RuntimeError( + f"container command failed (exit {result.exit_code}): {described}\n" + f"stdout: {result.stdout}\nstderr: {result.stderr}" + ) + return result + + +def _write_bytes_to_container( + executor: CommandExecutor, + container_path: str, + data: bytes, + *, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> None: + """Write `data` into `container_path` inside the container over `executor`. + + Replaces host bind-mounting of credential material (the docker-out-of- + docker fix): instead of `-v host:container` at `docker run` time, the + file's bytes are pushed in over the executor. + + SECURITY (credential leak fix): the payload is base64-encoded and fed to + an in-container `base64 -d` over the process's STDIN - it is NEVER placed + in argv. Anything in argv is world-readable via `ps` / `/proc// + cmdline`, so the previous `printf '%s' ` shape exposed the + credential bytes to any host user for the lifetime of the exec. stdin has + no argv length limit either, so the whole file goes in ONE exec (the old + chunked loop is gone). base64 keeps arbitrary/binary bytes intact through + the `sh -c` round-trip; `base64 -d` reconstructs them container-side. + + `timeout_s` (Phase 3) bounds the exec call so a wedged container can't + hang the transfer forever. + """ + parent = posixpath.dirname(container_path) + if parent: + _run_exec_checked(executor, ["mkdir", "-p", parent], timeout_s=timeout_s) + quoted_path = shlex.quote(container_path) + # Redacted label: credential-seeding must not leak the payload OR the raw + # command into the RuntimeError _run_exec_checked raises on failure. + label = f"write bytes to {container_path}" + # Truncate/create the destination first so a re-run (or a shorter payload + # than a stale prior write) doesn't leave trailing garbage. + _run_exec_checked( + executor, ["sh", "-c", f"> {quoted_path}"], timeout_s=timeout_s, label=label + ) + encoded = base64.b64encode(data) + _run_exec_checked( + executor, + ["sh", "-c", f"base64 -d >> {quoted_path}"], + timeout_s=timeout_s, + stdin=encoded, + label=label, + ) + + +def _transfer_path_to_container( + executor: CommandExecutor, + host_path: Path, + container_path: str, + *, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> None: + """Copy a host file or directory tree into the container over `executor`. + + Mirrors what a `-v host_path:container_path` bind mount used to provide, + but works when the driver itself runs inside a container (the host + path the caller staged files into is invisible to a sibling `docker + run -v`, but `docker exec` into the *target* container always works). + """ + if host_path.is_dir(): + for root, _dirs, files in os.walk(host_path): + rel_root = Path(root).relative_to(host_path) + for fname in files: + src = Path(root) / fname + rel = fname if rel_root == Path(".") else f"{rel_root.as_posix()}/{fname}" + dst = f"{container_path.rstrip('/')}/{rel}" + _write_bytes_to_container(executor, dst, src.read_bytes(), timeout_s=timeout_s) + else: + _write_bytes_to_container(executor, container_path, host_path.read_bytes(), timeout_s=timeout_s) + + +def _secure_container_path( + executor: CommandExecutor, + container_path: str, + *, + is_dir: bool, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, +) -> None: + """chown the transferred path to the in-container agent user (uid/gid + 1000) and lock file permissions to 0600 — mirrors what the host-side + `_chown_recursive` / `os.chmod(..., 0o600)` calls used to guarantee + before the bind-mount path was removed. + """ + quoted = shlex.quote(container_path) + if is_dir: + cmd = f"chown -R 1000:1000 {quoted} && find {quoted} -type f -exec chmod 600 {{}} +" + else: + cmd = f"chown 1000:1000 {quoted} && chmod 600 {quoted}" + _run_exec_checked(executor, ["sh", "-c", cmd], timeout_s=timeout_s) + + def _build_seeded_claude_dotjson(host_dotjson: Path, workspace_path: str) -> dict: """Build the synthetic ~/.claude.json the container should see. @@ -708,6 +1621,51 @@ def _build_seeded_claude_dotjson(host_dotjson: Path, workspace_path: str) -> dic return seeded +# Substrings docker/tmux emit when the workspace target itself is GONE +# (OOM-killed, `docker rm`'d, stopped, tmux session vanished) rather than a +# transient capture hiccup while the container is still alive. Matched +# case-insensitively against a failed exec's stderr so the readiness pollers +# can break out immediately instead of spinning the full startup/await +# deadline and then reporting a misleading generic timeout (a "degraded" +# workspace that actually masks a hard container death). +# +# Deliberately EXCLUDED: "cannot connect to the Docker daemon" / generic +# "error connecting to" - those signal a transient daemon or socket outage, +# NOT a dead container (the container is very likely still running once the +# daemon recovers). Treating them as death would abort a live workspace on a +# blip; they stay on the retry path and are bounded by the overall deadline. +_CONTAINER_DEAD_STDERR_MARKERS = ( + "no such container", + "is not running", + "no server running", # tmux daemon gone + "no such session", # tmux session vanished + "can't find session", +) + + +def _container_death_reason( + exc: subprocess.CalledProcessError | subprocess.TimeoutExpired, +) -> str | None: + """Return a human-readable reason if `exc` indicates the workspace target + is GONE, else None. + + A `TimeoutExpired` is always treated as transient (the container may just + be slow this once); only a `CalledProcessError` whose stderr names a dead + container / tmux server / tmux session counts as death. This lets the + readiness pollers distinguish "one failed capture while still alive" + (keep retrying) from "the container died" (stop now, surface the real + error) instead of blindly polling until the deadline. + """ + if isinstance(exc, subprocess.TimeoutExpired): + return None + stderr = exc.stderr if isinstance(exc.stderr, str) else "" + low = stderr.lower() + for marker in _CONTAINER_DEAD_STDERR_MARKERS: + if marker in low: + return stderr.strip() or marker + return None + + # --------------------------------------------------------------------------- # The workspace @@ -743,6 +1701,40 @@ class InteractiveTmuxWorkspace: # `launch_in_window`. Lists of pathlib.Path values. _launch_extras: dict[str, list[Path]] = field(default_factory=dict) + # Phase 2 executor seam: the `CommandExecutor` used for this workspace's + # container. Defaults to a `DockerExecExecutor(self.container)` in + # `__post_init__` when not supplied — every existing caller (including + # `_load_workspace`, which never knew about executors) gets identical + # behavior. Injectable so tests (and future non-docker transports) don't + # need to monkeypatch subprocess/docker. + executor: CommandExecutor | None = None + + # Environment seam: the `Environment` used to provision this workspace's + # backing target (Docker today) and, correspondingly, tear it down in + # `stop()`. `None` for workspaces reconstructed via `_load_workspace` + # (the registry only persists identifiers, not live provisioning + # objects) — `stop()` falls back to the historical `docker rm -f` in + # that case, preserving CLI behavior. + environment: Environment | None = None + + # Phase 4: one agent-agnostic `TmuxSession` per enabled agent, built in + # `__post_init__`. `send_message`/`await_completion`/`capture_response` + # and the startup-wait loop delegate their pane operations here instead + # of re-deriving `(container, window, executor)` inline; adapter submit + # patterns (still keyed by agent) stay separate since they're agent- + # specific, not generic tmux operations. Excluded from `repr`/equality — + # it's a derived cache, not part of the workspace's identity. + _sessions: dict[str, TmuxSession] = field(default_factory=dict, repr=False, compare=False) + + def __post_init__(self) -> None: + if self.executor is None: + self.executor = DockerExecExecutor(self.container) + self._sessions = { + agent: TmuxSession(target=self.container, window=_ADAPTERS[agent].window, executor=self.executor) + for agent in self.enabled_agents + if agent in _ADAPTERS + } + # ----------------------------------------------------------------------- # Lifecycle @@ -758,6 +1750,7 @@ def start_workspace( strict_startup: bool = True, host_claude_dotjson: Path | None = None, claude_plugin_dirs: Sequence[Path] | None = None, + environment: Environment | None = None, ) -> InteractiveTmuxWorkspace: """Start a new interactive-tmux workspace. @@ -799,6 +1792,14 @@ def start_workspace( bridge experiment, `docs/plans/workflow-skills.md` §9). Equivalent to setting `ITMUX_CLAUDE_PLUGIN_DIRS` (colon- separated, like `$PATH`). + environment: the `Environment` used to provision the workspace's + backing target and obtain a `CommandExecutor` for it. When + `None` (default), a `DockerEnvironment` is constructed from + `image`/`workdir` (and the generated container name) — + identical to this method's historical behavior. Pass an + explicit `Environment` to provision on a different backend + (local process, SSH/VPS, ...) without changing any + tmux/adapter logic. Returns: InteractiveTmuxWorkspace. In both strict and lax modes, @@ -828,7 +1829,10 @@ def start_workspace( # `docker run` (or a bad host auth dir) leaks staged auth material # under /tmp. try: - for agent in AGENTS: + # Phase 4: iterate the adapter registry (not the closed `AGENTS` + # tuple) so a 4th agent becomes enable-able purely by registering + # its adapter in `_ADAPTERS`, without editing this loop. + for agent in _ADAPTERS: adapter = _ADAPTERS[agent] src = host_auth.get(agent) if src is None: @@ -843,28 +1847,36 @@ def start_workspace( if not enabled: raise ValueError("start_workspace called with no enabled agents (host_auth empty)") - # Run the container with bind mounts (each mount is a -v arg). - run_cmd = [ - "docker", - "run", - "-d", - "--name", - container, - "--workdir", - workdir, - ] + # Provision the workspace's backing target WITHOUT credential + # bind mounts (Phase 2: docker-out-of-docker fix). Bind-mounting + # `-v host:container` requires the *outer* docker daemon to + # resolve `host` on its own filesystem — that breaks when this + # driver itself runs inside a container, where the throwaway + # staging dir is only visible to the driver's own mount + # namespace, not the sibling daemon's. Instead, credentials are + # pushed into the running target below via the executor + # `Environment.start()` returns (see `_transfer_path_to_ + # container`), which always targets the right backend + # regardless of where the driver process lives. + if environment is None: + environment = DockerEnvironment(name=container, image=image, workdir=workdir) + executor: CommandExecutor = environment.start() + + # Target is up; transfer each prepared credential file/dir + # into it over the executor seam instead of bind-mounting. for host_path, container_path in all_mounts: - run_cmd.extend(["-v", f"{host_path}:{container_path}"]) - run_cmd.extend([image, "sleep", "infinity"]) - _run(run_cmd) + _transfer_path_to_container(executor, host_path, container_path) + _secure_container_path(executor, container_path, is_dir=host_path.is_dir()) except Exception: - # Best-effort: `docker run -d` can fail after the container is - # created (e.g. start failure), so remove it too. - subprocess.run( - ["docker", "rm", "-f", container], - check=False, - capture_output=True, - ) + # Best-effort: provisioning can fail after the target is + # created (e.g. start failure), so tear it down too. + try: + if environment is not None: + environment.stop() + except Exception: + # Best-effort cleanup only; don't let a failed/wedged + # teardown mask the original failure being re-raised below. + logger.warning("environment.stop() failed during start_workspace cleanup", exc_info=True) shutil.rmtree(host_throwaway_dir, ignore_errors=True) raise @@ -877,6 +1889,8 @@ def start_workspace( tmux_size=tmux_size, host_throwaway_dir=host_throwaway_dir, enabled_agents=tuple(enabled), + executor=executor, + environment=environment, ) # Per-agent launch options. Today only claude has plugin-dir # support; codex/gemini ignore the kwarg with `del plugin_dirs`. @@ -898,31 +1912,10 @@ def _bootstrap_tmux_and_launch( cols, rows = self.tmux_size first = self.enabled_agents[0] # Create the session with the first agent's window name. - _docker_exec( - self.container, - "tmux", - "new-session", - "-d", - "-s", - TMUX_SESSION, - "-n", - first, - "-x", - str(cols), - "-y", - str(rows), - ) + self._sessions[first].start(cols, rows) # Create additional windows for the rest. for agent in self.enabled_agents[1:]: - _docker_exec( - self.container, - "tmux", - "new-window", - "-t", - TMUX_SESSION, - "-n", - agent, - ) + self._sessions[agent].start(cols, rows, as_new_window=True) # Launch each agent's CLI in its window, then wait until each pane # reports `is_started()` (M1 cross-review fix). Each adapter @@ -936,7 +1929,12 @@ def _bootstrap_tmux_and_launch( for agent in self.enabled_agents: adapter = _ADAPTERS[agent] extras = self._launch_extras.get(agent) or None - adapter.launch_in_window(self.container, self.workdir, plugin_dirs=extras) + adapter.launch_in_window( + self.container, + self.workdir, + plugin_dirs=extras, + executor=self.executor, + ) self._started[agent] = True self.startup_status[agent] = self._wait_for_started(agent, startup_timeout_s) @@ -954,11 +1952,39 @@ def _bootstrap_tmux_and_launch( def _wait_for_started(self, agent: str, timeout_s: float) -> AwaitResult: """Block until `agent`'s pane reports `is_started()`, or timeout.""" adapter = _ADAPTERS[agent] + session = self._sessions[agent] start = time.monotonic() deadline = start + timeout_s pane = "" while time.monotonic() < deadline: - pane = _tmux_capture(self.container, adapter.window) + try: + pane = session.capture_pane() + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + # Finding 4: distinguish a transient capture hiccup from a + # container that is GONE. A dead container would otherwise + # spin this loop until `timeout_s` and (in lax mode) return a + # falsely-"degraded" workspace masking the real docker error. + death = _container_death_reason(exc) + if death is not None: + duration_ms = (time.monotonic() - start) * 1000 + logger.error( + "wait_for_started(%s): container is gone mid-startup: %s", agent, death + ) + return AwaitResult( + ready=False, + timed_out=False, + reason="container_dead", + duration_ms=duration_ms, + stable_polls_observed=0, + pane=pane, + error=death, + ) + # Phase 3: a single wedged/failed capture while the container + # is still alive must not abort the whole wait - keep polling + # until the overall `timeout_s` deadline. + logger.warning("wait_for_started(%s): capture failed mid-poll: %s", agent, exc) + time.sleep(0.5) + continue # D-block-2 fix: predicate evaluated on the bottom-of-pane # tail so historical text in scrollback (now captured by # default) can't fool absence checks like @@ -992,10 +2018,23 @@ def _wait_for_started(self, agent: str, timeout_s: float) -> AwaitResult: # ----------------------------------------------------------------------- # The five public primitives - def send_message(self, agent: AgentName, text: str) -> None: - """Submit `text` to `agent`'s tmux pane using the per-agent submit pattern.""" + def send_message( + self, + agent: AgentName, + text: str, + *, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> None: + """Submit `text` to `agent`'s tmux pane using the per-agent submit pattern. + + Phase 3: `timeout_s` bounds each underlying tmux/docker-exec call + (default `DEFAULT_EXEC_TIMEOUT_S`) so a wedged container can't hang + this call forever. Payloads over `TMUX_SEND_KEYS_MAX_BYTES` are + automatically staged via tmux's paste-buffer instead of raw + `send-keys` (see `_tmux_send_literal`). + """ self._check_agent(agent) - _ADAPTERS[agent].submit(self.container, text) + _ADAPTERS[agent].submit(self.container, text, executor=self.executor, timeout_s=timeout_s) def await_completion( self, @@ -1004,6 +2043,7 @@ def await_completion( stable_polls: int = 4, poll_interval: float = 0.5, warmup: float = 2.0, + poll_timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, ) -> AwaitResult: """Block until `agent` returns to a stable ready state, or `timeout` passes. @@ -1031,6 +2071,13 @@ def await_completion( `send_message` and the agent rendering its generation marker) so the first ready observation isn't a stale idle frame. + `poll_timeout_s` (Phase 3) bounds each *individual* pane-capture + call, distinct from the overall `timeout` deadline above: a single + wedged `docker exec` no longer blocks past `poll_timeout_s` (default + `DEFAULT_EXEC_TIMEOUT_S`), and a poll that fails/times out is + treated as "not ready this round" rather than raising — the loop + keeps polling until the overall `timeout` is reached. + Returns an `AwaitResult`: - `.ready=True, .reason="ready"` when a stable ready state was reached; @@ -1041,6 +2088,7 @@ def await_completion( """ self._check_agent(agent) adapter = _ADAPTERS[agent] + session = self._sessions[agent] start = time.monotonic() deadline = start + timeout time.sleep(warmup) @@ -1057,7 +2105,37 @@ def await_completion( pane = "" tail = "" while time.monotonic() < deadline: - pane = _tmux_capture(self.container, adapter.window) + try: + pane = session.capture_pane(timeout_s=poll_timeout_s) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + # Finding 4: a container that is GONE (OOM, tmux session gone, + # daemon error) must break the poll immediately rather than + # spin until `timeout` and report a misleading generic + # timeout. A genuinely transient failure while the container + # is still alive stays on the retry path below. + death = _container_death_reason(exc) + if death is not None: + duration_ms = (time.monotonic() - start) * 1000 + logger.error( + "await_completion(%s): container is gone mid-poll: %s", agent, death + ) + return AwaitResult( + ready=False, + timed_out=False, + reason="container_dead", + duration_ms=duration_ms, + stable_polls_observed=consecutive_stable_ready, + pane=pane, + error=death, + ) + # Phase 3: a single failed/wedged poll (container still alive) + # must not abort the whole await - treat it as "not ready this + # round" and keep polling until the overall deadline above. + logger.warning("await_completion(%s): capture failed/timed out mid-poll: %s", agent, exc) + consecutive_stable_ready = 0 + last_tail = None + time.sleep(poll_interval) + continue tail = _pane_tail(pane) if adapter.is_ready(tail): ever_ready = True @@ -1097,18 +2175,44 @@ def await_completion( pane=pane, ) - def capture_response(self, agent: AgentName) -> str: - """Return the current contents of `agent`'s tmux pane.""" + def capture_response( + self, + agent: AgentName, + *, + timeout_s: float | None = DEFAULT_EXEC_TIMEOUT_S, + ) -> str: + """Return the current contents of `agent`'s tmux pane. + + Phase 3: bounded by `timeout_s` (default `DEFAULT_EXEC_TIMEOUT_S`) + so a wedged container can't hang this call forever. + """ self._check_agent(agent) - return _tmux_capture(self.container, _ADAPTERS[agent].window) + return self._sessions[agent].capture_pane(timeout_s=timeout_s) def stop(self) -> None: - """Tear down the container and remove throwaway credential copies.""" - subprocess.run( - ["docker", "rm", "-f", self.container], - check=False, - capture_output=True, - ) + """Tear down the backing target and remove throwaway credential + copies. + + Delegates to `self.environment.stop()` when set (the normal path + for anything returned by `start_workspace`). Falls back to the + historical `docker rm -f` when `environment` is `None` — e.g. a + workspace reconstructed by `_load_workspace`, which only persists + identifiers, not live `Environment` objects. + """ + if self.environment is not None: + self.environment.stop() + else: + try: + subprocess.run( + ["docker", "rm", "-f", self.container], + check=False, + capture_output=True, + timeout=DEFAULT_RUN_TIMEOUT_S, + ) + except subprocess.TimeoutExpired: + # Phase 3: bounded — a wedged `docker rm -f` must not hang + # `stop()` forever. Still clean up the host-side throwaway dir. + logger.warning("docker rm -f %s timed out during stop()", self.container) shutil.rmtree(self.host_throwaway_dir, ignore_errors=True) # -----------------------------------------------------------------------