diff --git a/examples/interactive_demo.py b/examples/interactive_demo.py new file mode 100644 index 0000000..9ce2ed4 --- /dev/null +++ b/examples/interactive_demo.py @@ -0,0 +1,129 @@ +"""Launch real harness UIs; this console only controls sessions and reports observed events. + +Run: uv run python examples/interactive_demo.py --agent all +Attach using the printed tmux command from another terminal. No prompt is sent by default. +""" + +import argparse +import asyncio +from contextlib import AsyncExitStack +import os +from pathlib import Path +import shlex +import signal +import sys + +from agent_shell import TmuxExecutionHost, TmuxPlacement, discover_terminal_launcher +from agent_shell.models.agent import AgentType +from agent_shell.shell import AgentShell + + +def arguments(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--agent", choices=["all", *[agent.value for agent in AgentType]], + default="all") + parser.add_argument("--cwd", default=os.getcwd()) + parser.add_argument("--model", help="Harness-native model selector; requires a single --agent") + parser.add_argument("--prompt", help="Optional initial prompt (makes a live model request)") + parser.add_argument("--new-terminal", action="store_true", + help="Open a terminal emulator attached to the real tmux session") + parser.add_argument("--split-pane", action="store_true", + help="Split beside the current tmux pane; keep focus on the controller") + args = parser.parse_args() + if args.model and args.agent == "all": + parser.error("--model requires a single --agent") + if args.split_pane and args.new_terminal: + parser.error("--split-pane cannot be combined with --new-terminal") + if args.split_pane and args.agent == "all": + parser.error("--split-pane requires a single --agent") + return args + + +async def report_events(name, session): + async for event in session.events(): + details = f"[{name}] {event.type}: {event.content}" + if event.session_id: + details += f" (session {event.session_id})" + if event.type == "result" and "output_tokens" in session.capabilities: + details += f"; output tokens={event.output_tokens}, cost=${event.cost:.4f}" + if event.error: + details += f"; {event.error}" + print(details, flush=True) + + +async def main(args): + names = [agent.value for agent in AgentType] if args.agent == "all" else [args.agent] + sessions = {} + observers = [] + loop = asyncio.get_running_loop() + commands = asyncio.Queue() + + def read_command(): + line = sys.stdin.readline() + if not line: + loop.remove_reader(sys.stdin) + commands.put_nowait("/quit") + else: + commands.put_nowait(line.rstrip("\n")) + + async with AsyncExitStack() as stack: + for name in names: + placement = ( + TmuxPlacement.split_pane() if args.split_pane + else TmuxPlacement.new_session() if not sessions + else TmuxPlacement.new_window(next(iter(sessions.values())).terminal.session_name) + ) + shell = AgentShell(AgentType(name), execution_host=TmuxExecutionHost(placement)) + session = await shell.open_interactive( + str(Path(args.cwd).resolve()), model=args.model, prompt=args.prompt, + ) + await stack.enter_async_context(session) + sessions[name] = session + print(f"{name}: real UI in pane {session.terminal.pane_id}; " + f"events: {', '.join(sorted(session.capabilities))}", flush=True) + + target = next(iter(sessions.values())).terminal.session_name + attach = ["tmux", "attach-session", "-t", target] + if args.split_pane: + print("Use prefix+arrow to move between the controller and agent panes.", flush=True) + else: + print("From another terminal: " + shlex.join(attach), flush=True) + print("Inside tmux: " + shlex.join(["tmux", "switch-client", "-t", target]), flush=True) + print("Use prefix+n to move between the real harness windows.", flush=True) + if args.new_terminal: + launcher = discover_terminal_launcher() + await launcher.launch(attach, cwd=args.cwd, env=os.environ.copy()) + + for name, session in sessions.items(): + observers.append(asyncio.create_task(report_events(name, session))) + loop.add_reader(sys.stdin, read_command) + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, commands.put_nowait, "/quit") + try: + print("Controller ready. Resolve trust/login prompts in the real UI first.", flush=True) + print("Send: | /key C-c | /quit", flush=True) + while (line := await commands.get()) != "/quit": + try: + if line.startswith("/key "): + _, name, key = line.split(maxsplit=2) + await sessions[name].terminal.send_key(key) + else: + name, text = line.split(maxsplit=1) + await sessions[name].terminal.send_text(text, submit=True) + except (KeyError, ValueError, RuntimeError) as error: + print(f"Could not send input: {error}", flush=True) + finally: + loop.remove_reader(sys.stdin) + for sig in (signal.SIGINT, signal.SIGTERM): + loop.remove_signal_handler(sig) + for observer in observers: + observer.cancel() + await asyncio.gather(*observers, return_exceptions=True) + print("Closed all demo sessions", flush=True) + + +if __name__ == "__main__": + try: + asyncio.run(main(arguments())) + except KeyboardInterrupt: + pass diff --git a/src/agent_shell/adapters/agent_adapter_protocol.py b/src/agent_shell/adapters/agent_adapter_protocol.py index b9ba810..edcd855 100644 --- a/src/agent_shell/adapters/agent_adapter_protocol.py +++ b/src/agent_shell/adapters/agent_adapter_protocol.py @@ -45,6 +45,7 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: ... diff --git a/src/agent_shell/adapters/claude_code_adapter.py b/src/agent_shell/adapters/claude_code_adapter.py index abdad54..56ef06e 100644 --- a/src/agent_shell/adapters/claude_code_adapter.py +++ b/src/agent_shell/adapters/claude_code_adapter.py @@ -3,6 +3,7 @@ import json import logging import os +import shlex import warnings from pathlib import Path from typing import AsyncIterator @@ -43,6 +44,59 @@ } class ClaudeCodeAdapter(): + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch, event_writer_command + + hook = {"type": "command", "command": shlex.join(event_writer_command(directory))} + settings = directory / "claude-settings.json" + settings.write_text(json.dumps({"hooks": { + name: [{"hooks": [hook]}] for name in ( + "SessionStart", "UserPromptSubmit", "PreToolUse", "Stop", + ) + }})) + command = ["claude", "--settings", str(settings)] + if session_id: + command += ["--resume", session_id] + if model: + command += ["--model", model] + if effort: + command += ["--effort", effort] + if prompt is not None: + command += ["--", prompt] + return InteractiveLaunch( + command, self.parse_interactive_event, + frozenset({"text", "session_id", "stop_requested", "tool_use"}), + ) + + def parse_interactive_event(self, event: dict) -> list[StreamEvent]: + """Observe hooks; Stop is provisional because other hooks may request continuation.""" + kind = event.get("hook_event_name") + session_id = event.get("session_id") + if kind == "SessionStart": + return [StreamEvent(type="system", content="", session_id=session_id)] + if kind == "UserPromptSubmit": + return [StreamEvent(type="status", content="turn_started", session_id=session_id)] + if kind == "PreToolUse": + return [StreamEvent(type="tool_use", content=event.get("tool_name", ""), + session_id=session_id)] + if kind == "Stop": + events = [] + text = event.get("last_assistant_message") + if text: + events.append(StreamEvent(type="text", content=text, session_id=session_id)) + events.append(StreamEvent(type="status", content="stop_requested", + session_id=session_id)) + return events + return [] + def __init__( self, execution_host: ExecutionHost | None = None, @@ -255,8 +309,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/codex_adapter.py b/src/agent_shell/adapters/codex_adapter.py index a5e9c96..47bc51e 100644 --- a/src/agent_shell/adapters/codex_adapter.py +++ b/src/agent_shell/adapters/codex_adapter.py @@ -4,6 +4,7 @@ import logging import os import warnings +from pathlib import Path from typing import AsyncIterator from agent_shell.adapters.health import run_health_probe @@ -32,6 +33,45 @@ class CodexAdapter: + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch, event_writer_command + + command = ["codex"] + if session_id: + command += ["resume", session_id] + # A per-invocation override; the user's config file is never changed. + command += ["-c", "notify=" + json.dumps(event_writer_command(directory))] + if model: + command += ["--model", model] + if effort: + command += ["-c", "model_reasoning_effort=" + json.dumps(effort)] + if prompt is not None: + command += ["--", prompt] + return InteractiveLaunch( + command, self.parse_interactive_event, + frozenset({"text", "session_id", "turn_complete"}), + ) + + def parse_interactive_event(self, event: dict) -> list[StreamEvent]: + """Normalize Codex's after-turn notification, without guessing usage or failures.""" + if event.get("type") != "agent-turn-complete": + return [] + session_id = event.get("thread-id") + events = [StreamEvent(type="system", content="", session_id=session_id)] + text = event.get("last-assistant-message") + if text: + events.append(StreamEvent(type="text", content=text, session_id=session_id)) + events.append(StreamEvent(type="result", content="ok", session_id=session_id)) + return events + def __init__( self, execution_host: ExecutionHost | None = None, @@ -308,8 +348,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/copilot_cli_adapter.py b/src/agent_shell/adapters/copilot_cli_adapter.py index 86900e7..04b7573 100644 --- a/src/agent_shell/adapters/copilot_cli_adapter.py +++ b/src/agent_shell/adapters/copilot_cli_adapter.py @@ -139,6 +139,60 @@ def _json_rpc_result(response: dict) -> dict: class CopilotCLIAdapter: + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch, event_writer_command + import shlex + + effort = _normalize_effort(effort) + (directory / ".github/plugin").mkdir(parents=True) + (directory / ".github/plugin/plugin.json").write_text(json.dumps({ + "name": "agentshell-observer", "version": "0.1.0", + })) + (directory / "hooks").mkdir() + hook = {"type": "command", "command": shlex.join(event_writer_command(directory))} + (directory / "hooks/hooks.json").write_text(json.dumps({ + "version": 1, "hooks": {name: [hook] for name in ( + "SessionStart", "UserPromptSubmit", "PostToolUse", "Stop", "ErrorOccurred", + )}, + })) + command = ["copilot", "--no-auto-update", "--plugin-dir", str(directory)] + if model: + command += ["--model", model] + if effort: + command += ["--effort", effort] + if session_id: + command += ["--resume=" + session_id] + if prompt is not None: + command += ["--interactive", prompt] + return InteractiveLaunch(command, self.parse_interactive_event, frozenset({ + "session_id", "tool_use", "stop_requested", + })) + + def parse_interactive_event(self, event: dict) -> list[StreamEvent]: + name = event.get("hook_event_name") + session_id = event.get("session_id") + if name == "SessionStart": + return [StreamEvent(type="system", content="", session_id=session_id)] + if name == "PostToolUse": + return [StreamEvent(type="tool_use", content=event.get("tool_name", "tool"), + session_id=session_id)] + # Recoverable errors and Stop gates are observations, not final turn verdicts. + if name == "ErrorOccurred": + message = (event.get("error") or {}).get("message", "Harness error") + return [StreamEvent(type="status", content=message, session_id=session_id)] + status = {"UserPromptSubmit": "turn_started", "Stop": "stop_requested"}.get(name) + if status: + return [StreamEvent(type="status", content=status, session_id=session_id)] + return [] + def __init__( self, execution_host: ExecutionHost | None = None, @@ -389,8 +443,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/cursor_adapter.py b/src/agent_shell/adapters/cursor_adapter.py index d47320c..4b851a2 100644 --- a/src/agent_shell/adapters/cursor_adapter.py +++ b/src/agent_shell/adapters/cursor_adapter.py @@ -3,6 +3,7 @@ import json import logging import os +import shlex import warnings from pathlib import Path from tempfile import NamedTemporaryFile @@ -34,6 +35,52 @@ class CursorAdapter: + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch, event_writer_command + + if effort: + raise ValueError("Cursor effort must be supplied in the native model selector") + # Cursor requires at least three non-root components in a local plugin path. + plugin = directory / "cursor-plugin" + plugin.mkdir() + (plugin / ".cursor-plugin").mkdir() + (plugin / ".cursor-plugin/plugin.json").write_text(json.dumps({ + "name": "agentshell-observer", "version": "0.1.0", "hooks": "./hooks/hooks.json", + })) + (plugin / "hooks").mkdir() + command_hook = {"command": shlex.join(event_writer_command(directory))} + (plugin / "hooks/hooks.json").write_text(json.dumps({ + "version": 1, "hooks": {name: [command_hook] for name in ( + "sessionStart", + )}, + })) + command = ["cursor-agent", "--plugin-dir", str(plugin)] + if model: + command += ["--model", model] + if session_id: + command += ["--resume=" + session_id] + if prompt is not None: + command += ["--", prompt] + # The installed CLI delivers plugin SessionStart only on new conversations. + # It does not deliver turn/response hooks, or SessionStart when resuming. + # Advertise only the capability verified against the real interactive harness. + capabilities = frozenset() if session_id else frozenset({"session_id"}) + return InteractiveLaunch(command, self.parse_interactive_event, capabilities) + + def parse_interactive_event(self, event: dict) -> list[StreamEvent]: + if event.get("hook_event_name") == "sessionStart": + return [StreamEvent(type="system", content="", + session_id=event.get("conversation_id"))] + return [] + def __init__( self, execution_host: ExecutionHost | None = None, @@ -311,8 +358,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/grok_adapter.py b/src/agent_shell/adapters/grok_adapter.py index cbbd3a7..0eab533 100644 --- a/src/agent_shell/adapters/grok_adapter.py +++ b/src/agent_shell/adapters/grok_adapter.py @@ -5,6 +5,7 @@ import os import tomllib import warnings +from uuid import UUID, uuid4 from pathlib import Path from typing import AsyncIterator @@ -72,6 +73,69 @@ def _result_error_reason(errors) -> str | None: class GrokAdapter: + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + from agent_shell.interactive import InteractiveLaunch + + # Pin a UUID so discovery cannot pick up another concurrent conversation. + selected_id = str(UUID(session_id)) if session_id else str(uuid4()) + root = Path(os.environ.get("GROK_HOME") or Path.home() / ".grok") / "sessions" + + def event_path() -> Path | None: + matches = list(root.glob(f"*/{selected_id}/updates.jsonl")) + if len(matches) > 1: + raise ValueError("Grok session UUID appears in multiple workspaces") + return matches[0] if matches else None + + previous = event_path() + offset = previous.stat().st_size if previous else 0 + command = ["grok", "--resume" if session_id else "--session-id", selected_id] + if allowed_tools is not None: + if not allowed_tools or any(not name or "," in name for name in allowed_tools): + raise ValueError("allowed_tools must contain nonempty native tool names") + command += ["--tools", ",".join(allowed_tools)] + if model: + command += ["--model", model] + if effort: + command += ["--reasoning-effort", effort] + if prompt is not None: + command += ["--", prompt] + + def parse(record: dict) -> list[StreamEvent]: + params = record.get("params") or {} + if params.get("sessionId") != selected_id: + return [] + update = params.get("update") or {} + kind = update.get("sessionUpdate") + if kind == "agent_message_chunk": + content = update.get("content") or {} + if content.get("type") == "text": + return [StreamEvent(type="text", content=content.get("text", ""), + session_id=selected_id)] + if kind == "user_message_chunk": + return [StreamEvent(type="status", content="turn_started", session_id=selected_id)] + if kind == "tool_call": + return [StreamEvent(type="tool_use", content=update.get("title", "tool"), + session_id=selected_id)] + if kind == "turn_completed": + reason = update.get("stop_reason") + usage = update.get("usage") or {} + return [StreamEvent( + type="result", content="ok" if reason == "end_turn" else "error", + error=None if reason == "end_turn" else str(reason or "Unknown stop reason"), + session_id=selected_id, output_tokens=int(usage.get("outputTokens") or 0), + duration=(update.get("elapsed_ms") or 0) / 1000, + )] + return [] + + return InteractiveLaunch( + command, parse, frozenset({"session_id", "text", "tool_use", "turn_complete", + "output_tokens", "duration"}), + event_path=event_path, event_offset=offset, + ) + def __init__( self, execution_host: ExecutionHost | None = None, @@ -323,8 +387,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/health.py b/src/agent_shell/adapters/health.py index 59979dd..cfe23d4 100644 --- a/src/agent_shell/adapters/health.py +++ b/src/agent_shell/adapters/health.py @@ -31,6 +31,7 @@ async def run_health_probe( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: events: list[StreamEvent] = [] @@ -39,6 +40,7 @@ async def _consume() -> None: cwd=cwd, prompt=HEALTH_PROMPT, model=model, + effort=effort, allowed_tools=[], auto_approve=True, ): diff --git a/src/agent_shell/adapters/opencode_adapter.py b/src/agent_shell/adapters/opencode_adapter.py index 5b14d48..423930e 100644 --- a/src/agent_shell/adapters/opencode_adapter.py +++ b/src/agent_shell/adapters/opencode_adapter.py @@ -44,6 +44,73 @@ class OpenCodeAdapter(): + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch + + if effort: + raise ValueError("OpenCode interactive CLI has no effort flag") + plugin = directory / "opencode-events.mjs" + plugin.write_text( + 'import {appendFileSync} from "node:fs";\n' + 'export default async () => ({event: async ({event}) => {\n' + ' if (event.type.startsWith("session.") || event.type.startsWith("message."))\n' + f' appendFileSync({json.dumps(str(directory / "events.jsonl"))}, ' + 'JSON.stringify(event) + "\\n");\n' + '}});\n' + ) + config = json.loads(os.environ.get("OPENCODE_CONFIG_CONTENT") or "{}") + config["plugin"] = [*config.get("plugin", []), plugin.as_uri()] + command = ["opencode"] + if model: + command += ["--model", model] + if session_id: + command += ["--session", session_id] + if prompt is not None: + command += ["--prompt", prompt] + roles: dict[str, str] = {} + seen: set[str] = set() + + def parse(event: dict) -> list[StreamEvent]: + kind = event.get("type") + data = event.get("properties") or {} + info = data.get("info") or {} + if kind == "session.created": + return [StreamEvent(type="system", content="", session_id=info.get("id"))] + if kind == "message.updated": + roles[info["id"]] = info.get("role", "") + if kind == "message.part.updated": + part = data.get("part") or {} + key = part.get("id") + if roles.get(part.get("messageID")) != "assistant" or key in seen: + return [] + session = part.get("sessionID") + if part.get("type") == "text" and (part.get("time") or {}).get("end"): + seen.add(key) + return [StreamEvent(type="text", content=part.get("text", ""), + session_id=session)] + if part.get("type") == "tool": + seen.add(key) + return [StreamEvent(type="tool_use", content=part.get("tool", "tool"), + session_id=session)] + if kind == "session.idle": + return [StreamEvent(type="status", content="idle", + session_id=data.get("sessionID"))] + if kind == "session.error": + return [StreamEvent(type="status", content=json.dumps(data.get("error")), + session_id=data.get("sessionID"))] + return [] + + return InteractiveLaunch(command, parse, frozenset({"session_id", "text", "tool_use"}), + env={"OPENCODE_CONFIG_CONTENT": json.dumps(config)}) + def __init__( self, execution_host: ExecutionHost | None = None, @@ -372,8 +439,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/adapters/pi_adapter.py b/src/agent_shell/adapters/pi_adapter.py index 41be203..3f5db31 100644 --- a/src/agent_shell/adapters/pi_adapter.py +++ b/src/agent_shell/adapters/pi_adapter.py @@ -5,6 +5,7 @@ import os import re import warnings +from pathlib import Path from typing import AsyncIterator from agent_shell.adapters.health import run_health_probe @@ -65,6 +66,82 @@ def _failure_reason(message: dict) -> str: class PiAdapter: + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None = None, + ): + if allowed_tools is not None: + raise NotImplementedError( + "Interactive allowed_tools is not implemented for this harness" + ) + + from agent_shell.interactive import InteractiveLaunch + + extension = directory / "pi-events.mjs" + destination = json.dumps(str(directory / "events.jsonl")) + extension.write_text( + 'import { appendFileSync } from "node:fs";\n' + 'export default function (pi) {\n' + ' for (const type of ["session_start", "agent_start", "message_update",\n' + ' "tool_execution_start", "agent_end", "agent_settled"]) {\n' + ' pi.on(type, (event, ctx) => {\n' + ' const record = {...event, session_id: ctx.sessionManager.getSessionId()};\n' + f' appendFileSync({destination}, JSON.stringify(record) + "\\n");\n' + ' });\n' + ' }\n' + '}\n' + ) + command = ["pi", "--extension", str(extension)] + if session_id: + command += ["--session-id", session_id] + if model: + command += ["--model", model] + if effort: + command += ["--thinking", effort] + if prompt is not None: + command += ["--", prompt] + + # State belongs to this session, not the adapter: concurrent sessions cannot mix usage. + pending: StreamEvent | None = None + tokens = 0 + cost = 0.0 + + def parse(event: dict) -> list[StreamEvent]: + nonlocal pending, tokens, cost + kind = event.get("type") + sid = event.get("session_id") + if kind == "session_start": + pending, tokens, cost = None, 0, 0.0 + return [StreamEvent(type="system", content="", session_id=sid)] + if kind == "agent_start": + return [StreamEvent(type="status", content="turn_started", session_id=sid)] + if kind == "agent_end": + results = self._parse_event(event, include_thinking=False) + pending = next((item for item in results if item.type == "result"), None) + if pending is not None: + tokens += pending.output_tokens + cost += pending.cost + return [] + if kind == "agent_settled": + if pending is None: + return [StreamEvent(type="error", content="Pi settled without a run result", + session_id=sid)] + pending.output_tokens = tokens + pending.cost = cost + pending.session_id = sid + results = [pending] + pending, tokens, cost = None, 0, 0.0 + return results + results = self._parse_event(event, include_thinking=False) + for item in results: + item.session_id = sid + return results + + return InteractiveLaunch( + command, parse, + frozenset({"text", "session_id", "tool_use", "turn_complete", "output_tokens", "cost"}), + ) + def __init__( self, execution_host: ExecutionHost | None = None, @@ -334,8 +411,9 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: - return await run_health_probe(self, cwd, model=model, timeout=timeout) + return await run_health_probe(self, cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/interactive.py b/src/agent_shell/interactive.py new file mode 100644 index 0000000..0b5c6d3 --- /dev/null +++ b/src/agent_shell/interactive.py @@ -0,0 +1,187 @@ +"""Experimental interactive session contracts and shared event delivery. + +Terminal transport is owned by the host. All harness-specific commands and event parsing +are supplied by the adapter. No terminal screen is parsed into structured results. +""" + +from __future__ import annotations + +import asyncio +from collections import deque +import contextlib +from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass +import json +import os +from pathlib import Path +import shutil +import sys +import tempfile +from typing import Protocol, runtime_checkable + +from agent_shell.execution import IsolationPolicy +from agent_shell.models.agent import StreamEvent + + +class InteractiveTerminal(Protocol): + closed: bool + returncode: int | None + + async def capture_screen(self) -> str: ... + async def send_text(self, text: str, *, submit: bool = False) -> None: ... + async def send_key(self, key: str) -> None: ... + async def resize(self, *, columns: int, rows: int) -> None: ... + async def wait(self) -> int: ... + async def close(self) -> None: ... + + +@runtime_checkable +class InteractiveExecutionHost(Protocol): + async def launch_interactive( + self, command: list[str], cwd: str, *, env: dict[str, str] | None = None, + isolation_policy: IsolationPolicy | None = None, + ) -> InteractiveTerminal: ... + + +@dataclass +class InteractiveLaunch: + command: list[str] + parse_event: Callable[[dict], list[StreamEvent]] + capabilities: frozenset[str] + env: dict[str, str] | None = None + event_path: Callable[[], Path | None] | None = None + event_offset: int = 0 + + +@runtime_checkable +class InteractiveAdapter(Protocol): + def prepare_interactive( + self, directory: Path, *, prompt: str | None, model: str | None, + effort: str | None, session_id: str | None, allowed_tools: list[str] | None, + ) -> InteractiveLaunch: ... + + +def event_writer_command(directory: Path) -> list[str]: + """Command accepting a JSON object on stdin or as its final argument.""" + return [sys.executable, str(Path(__file__).with_name("interactive_event_writer.py")), + str(directory / "events.jsonl")] + + +class InteractiveSession: + def __init__(self, terminal: InteractiveTerminal, directory: Path, launch: InteractiveLaunch): + self.terminal = terminal + self.capabilities = launch.capabilities + self._directory = directory + self._parse_event = launch.parse_event + self._reading = False + self._offset = launch.event_offset + self._event_path = launch.event_path or (lambda: directory / "events.jsonl") + self._pending: deque[StreamEvent] = deque() + self._ended = False + + async def events(self) -> AsyncIterator[StreamEvent]: + """Read hook/extension events until close. Only one reader is allowed at a time. + + Events describe the interactive conversation, including manually submitted turns. + ``result`` is emitted only by an adapter with a positive completion signal. + Use capabilities to distinguish unavailable metrics from actual reported zero values. + """ + if self._reading: + raise RuntimeError("An interactive session permits only one event reader") + if self._ended or self.terminal.closed: + return + self._reading = True + exit_task = asyncio.create_task(self.terminal.wait()) + exit_seen: float | None = None + try: + with contextlib.ExitStack() as files: + log = None + while not self.terminal.closed: + if self._pending: + yield self._pending.popleft() + continue + if log is None: + path = self._event_path() + if path is not None: + with contextlib.suppress(FileNotFoundError): + log = files.enter_context(path.open("rb")) + log.seek(self._offset) + start = log.tell() if log is not None else 0 + line = log.readline(8 * 1024 * 1024 + 1) if log is not None else b"" + if line.endswith(b"\n"): + self._offset = log.tell() + try: + record = json.loads(line) + if not isinstance(record, dict): + raise ValueError("Expected an event object") + events = self._parse_event(record) + except (ValueError, TypeError, KeyError) as error: + yield StreamEvent( + type="error", content=f"Invalid interactive event: {error}", + ) + else: + self._pending.extend(events) + continue + if len(line) > 8 * 1024 * 1024: + raise ValueError("Interactive event exceeds 8 MiB") + if log is not None: + log.seek(start) + if exit_task.done(): + # Notification subprocesses can outlive the harness briefly. This grace + # only drains records; it never infers a completed turn from silence. + now = asyncio.get_running_loop().time() + if exit_seen is None: + exit_seen = now + elif now - exit_seen >= 0.25: + self._ended = True + if line: + yield StreamEvent( + type="error", content="Incomplete interactive event", + ) + try: + status = exit_task.result() + except Exception as error: + yield StreamEvent(type="error", content=str(error)) + else: + yield StreamEvent( + type="process_exit", content="Harness process exited", + returncode=status, signal=-status if status < 0 else None, + ) + return + await asyncio.sleep(0.05) + finally: + self._reading = False + exit_task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await exit_task + + async def close(self) -> None: + await self.terminal.close() + shutil.rmtree(self._directory, ignore_errors=True) + + async def __aenter__(self) -> InteractiveSession: + return self + + async def __aexit__(self, *exc) -> None: + await self.close() + + +async def open_interactive_session( + adapter: InteractiveAdapter, host: InteractiveExecutionHost, policy: IsolationPolicy, + cwd: str, *, prompt: str | None, model: str | None, effort: str | None, + session_id: str | None, allowed_tools: list[str] | None = None, +) -> InteractiveSession: + directory = Path(tempfile.mkdtemp(prefix="agentshell-events-")) + try: + (directory / "events.jsonl").touch(mode=0o600) + launch = adapter.prepare_interactive( + directory, prompt=prompt, model=model, effort=effort, + session_id=session_id, allowed_tools=allowed_tools, + ) + terminal = await host.launch_interactive( + launch.command, cwd, env={**os.environ, **(launch.env or {})}, isolation_policy=policy, + ) + return InteractiveSession(terminal, directory, launch) + except BaseException: + shutil.rmtree(directory, ignore_errors=True) + raise diff --git a/src/agent_shell/interactive_event_writer.py b/src/agent_shell/interactive_event_writer.py new file mode 100644 index 0000000..fc19f7e --- /dev/null +++ b/src/agent_shell/interactive_event_writer.py @@ -0,0 +1,19 @@ +"""Standalone, harness-neutral JSON hook recorder. Never prints hook output.""" + +import fcntl +import json +import sys + + +def main() -> None: + record = json.loads(sys.argv[2]) if len(sys.argv) > 2 else json.load(sys.stdin) + if not isinstance(record, dict): + raise ValueError("Expected a JSON object") + with open(sys.argv[1], "a", encoding="utf-8") as log: + fcntl.flock(log, fcntl.LOCK_EX) + log.write(json.dumps(record, ensure_ascii=False) + "\n") + log.flush() + + +if __name__ == "__main__": + main() diff --git a/src/agent_shell/interactive_terminal.py b/src/agent_shell/interactive_terminal.py new file mode 100644 index 0000000..dbcdd3c --- /dev/null +++ b/src/agent_shell/interactive_terminal.py @@ -0,0 +1,234 @@ +"""Shared experimental control of a real harness terminal, with no screen interpretation.""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +from pathlib import Path +import shutil +import sys +import tempfile +import uuid + +from agent_shell.execution import IsolationPolicy, IsolationUnavailableError, NoIsolation +from agent_shell.tmux_ownership import IDENTITY_FORMAT, TmuxResource +from agent_shell.tmux import ( + TmuxPlacement, + TmuxUnavailableError, + _tmux_current_session, + _tmux_exact_session_target, +) + + +class TmuxTerminalSession: + """Own one interactive pane and its lifetime. Use ``close()`` or ``async with``. + + The UI runs directly on tmux's PTY. Screen capture is for inspection, never a success + signal. The terminal is retained after child exit until explicitly closed. + """ + + def __init__(self, tmux: str, directory: Path): + self._tmux = tmux + self._directory = directory + self._environment = os.environ.copy() + self._input_lock = asyncio.Lock() + self._resource: TmuxResource | None = None + self.pane_id = "" + self.window_id = "" + self.session_name = "" + self.pid = 0 + self.returncode: int | None = None + self.closed = False + self._close_task: asyncio.Task | None = None + self._lost = False + os.mkfifo(directory / "owner", mode=0o600) + self._owner_fd = os.open(directory / "owner", os.O_RDWR | os.O_NONBLOCK) + + @classmethod + async def launch( + cls, command: list[str], cwd: str, *, placement: TmuxPlacement | None = None, + env: dict[str, str] | None = None, isolation_policy: IsolationPolicy | None = None, + ) -> TmuxTerminalSession: + policy = isolation_policy if isolation_policy is not None else NoIsolation() + if not isinstance(policy, NoIsolation): + raise IsolationUnavailableError("Interactive tmux supports only NoIsolation") + if not command or not Path(cwd).is_dir(): + raise ValueError("A command and an existing working directory are required") + tmux = shutil.which("tmux") + if tmux is None: + raise TmuxUnavailableError( + "Interactive sessions require the optional `tmux` executable" + ) + placement = placement or TmuxPlacement.new_session() + session = placement.session or f"agentshell-ui-{uuid.uuid4().hex}" + if placement.kind in {"current-window", "split-pane"}: + session = await _tmux_current_session(tmux) + directory = Path(tempfile.mkdtemp(prefix="agentshell-ui-")) + terminal = cls(tmux, directory) + terminal.session_name = session + try: + (directory / "launch.json").write_text(json.dumps({ + "command": command, "cwd": str(Path(cwd).resolve()), + "env": dict(env) if env is not None else os.environ.copy(), + })) + worker = str(Path(__file__).with_name("interactive_worker.py")) + if placement.kind == "new-session": + args = ["new-session", "-d", "-s", session, "-x", "100", "-y", "30"] + elif placement.kind == "split-pane": + args = ["split-window", "-h" if placement.direction == "right" else "-v", + "-t", os.environ["TMUX_PANE"]] + if not placement.focus: + args.append("-d") + else: + args = ["new-window", "-t", _tmux_exact_session_target(session)] + if not placement.focus: + args.append("-d") + args += ["-P", "-F", IDENTITY_FORMAT, "--", + sys.executable, worker, str(directory)] + identity = await terminal._command(*args) + kind = ("session" if placement.kind == "new-session" + else "pane" if placement.kind == "split-pane" else "window") + try: + terminal._resource = TmuxResource.from_identity(kind, identity, directory) + except ValueError as error: + raise TmuxUnavailableError(str(error)) from error + terminal.pane_id = terminal._resource.pane_id + terminal.window_id = terminal._resource.window_id + (directory / "start").touch() + async with asyncio.timeout(5): + while not (status := terminal._status()): + await asyncio.sleep(0.02) + if "error" in status: + raise TmuxUnavailableError(status["error"]) + terminal.pid = status["pid"] + terminal.returncode = status["returncode"] + return terminal + except BaseException: + await terminal.close() + raise + + async def _command(self, *args: str, data: bytes | None = None) -> str: + process = await asyncio.create_subprocess_exec( + self._tmux, "-f", "/dev/null", + *(["-S", self._resource.socket_path] if self._resource else []), + *args, env=self._environment, + stdin=asyncio.subprocess.PIPE if data is not None else asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(data), 5) + except BaseException: + with contextlib.suppress(ProcessLookupError): + process.kill() + await process.wait() + raise + if process.returncode: + raise TmuxUnavailableError(stderr.decode(errors="replace").strip()) + return stdout.decode(errors="replace") + + def _check_open(self) -> None: + if self.closed: + raise RuntimeError("Interactive terminal is closed") + + def _status(self) -> dict: + try: + return json.loads((self._directory / "status.json").read_text()) + except FileNotFoundError: + return {} + + async def capture_screen(self) -> str: + """Capture the current screen and retained scrollback for human inspection.""" + self._check_open() + return await self._command("capture-pane", "-p", "-J", "-S", "-", "-t", self.pane_id) + + async def send_text(self, text: str, *, submit: bool = False) -> None: + """Paste literal text. The caller must ensure the harness is ready for input.""" + self._check_open() + if any(ord(char) < 32 and char not in "\n\t" for char in text) or "\x7f" in text: + raise ValueError("Text must not contain terminal control characters") + async with self._input_lock: + buffer = f"agentshell-{uuid.uuid4().hex}" + try: + await self._command("load-buffer", "-b", buffer, "-", data=text.encode()) + await self._command("paste-buffer", "-d", "-p", "-b", buffer, "-t", self.pane_id) + if submit: + await self._command("send-keys", "-t", self.pane_id, "Enter") + finally: + with contextlib.suppress(TmuxUnavailableError): + await self._command("delete-buffer", "-b", buffer) + + async def wait(self) -> int: + """Wait for process exit, which is distinct from a completed agent turn.""" + last_probe = 0.0 + while self.returncode is None: + self._check_open() + self.returncode = self._status().get("returncode") + if self.returncode is None: + now = asyncio.get_running_loop().time() + if now - last_probe >= 0.5: + try: + await self._command("display-message", "-p", "-t", self.pane_id, + "#{pane_id}") + except TmuxUnavailableError as error: + self._lost = True + raise RuntimeError( + "Interactive terminal disappeared before exit status was recorded" + ) from error + last_probe = now + await asyncio.sleep(0.05) + return self.returncode + + async def send_key(self, key: str) -> None: + """Send one supported control key, for menus, submission, or interruption.""" + self._check_open() + if key not in {"Enter", "Escape", "C-c", "C-d", "Tab", "Up", "Down", "Left", "Right"}: + raise ValueError(f"Unsupported terminal key: {key}") + async with self._input_lock: + await self._command("send-keys", "-t", self.pane_id, key) + + async def resize(self, *, columns: int, rows: int) -> None: + self._check_open() + if type(columns) is not int or type(rows) is not int or min(columns, rows) < 2: + raise ValueError("Terminal dimensions must be integers of at least 2") + split = self._resource is not None and self._resource.kind == "pane" + await self._command("resize-pane" if split else "resize-window", + "-t", self.pane_id if split else self.window_id, + "-x", str(columns), "-y", str(rows)) + + async def close(self) -> None: + """Close once, including when concurrent callers close or one caller is cancelled.""" + if self._close_task is None: + self._close_task = asyncio.create_task(self._close()) + await asyncio.shield(self._close_task) + + async def _close(self) -> None: + if self.closed: + return + try: + self.returncode = self._status().get("returncode", self.returncode) + if self.pid and not self._lost and self._directory.exists(): + (self._directory / "stop").touch() + with contextlib.suppress(TimeoutError): + async with asyncio.timeout(2): + while True: + status = self._status() + self.returncode = status.get("returncode", self.returncode) + if status.get("stopped"): + break + await asyncio.sleep(0.02) + if self._resource: + with contextlib.suppress(TmuxUnavailableError, TimeoutError, OSError): + await self._command(*self._resource.cleanup_args()) + finally: + # Releasing the FIFO lets the worker clean up even when tmux control has failed. + os.close(self._owner_fd) + self.closed = True + shutil.rmtree(self._directory, ignore_errors=True) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + await self.close() diff --git a/src/agent_shell/interactive_worker.py b/src/agent_shell/interactive_worker.py new file mode 100644 index 0000000..65b433a --- /dev/null +++ b/src/agent_shell/interactive_worker.py @@ -0,0 +1,111 @@ +"""Standalone tmux worker: the child inherits the pane's actual controlling terminal. + +Unlike the headless bridge, this worker never pipes or renders the harness's output. +Only startup/exit status travels through private files. Kept stdlib-only for direct execution. +""" + +import json +import os +from pathlib import Path +import subprocess +import signal +import contextlib +import shutil +import sys +import time + +from tmux_ownership import TmuxResource +from process_guardian import _STOP_GROUP, _send_guardian_command, _start_guardian + + +def write_status(directory: Path, value: dict) -> None: + temporary = directory / "status.tmp" + temporary.write_text(json.dumps(value)) + temporary.replace(directory / "status.json") + + +def main() -> None: + # A Python handler resets to the default on exec, unlike SIG_IGN. Ctrl-C reaches the + # harness while this supervisor survives to record its exit status. + signal.signal(signal.SIGINT, lambda *_: None) + directory = Path(sys.argv[1]) + owner_fd = None + resource = TmuxResource( + "pane", "", "", os.environ["TMUX_PANE"], + os.environ["TMUX"].rsplit(",", 2)[0], directory.name, + ) + try: + owner_fd = os.open(directory / "owner", os.O_RDONLY | os.O_NONBLOCK) + run(directory, owner_fd) + finally: + if owner_fd is not None: + os.close(owner_fd) + shutil.rmtree(directory, ignore_errors=True) + # Cover startup too: remain-on-exit must never retain an orphaned pane. The socket + # and launch marker prevent cleanup from reaching another server or a reused ID. + with contextlib.suppress(OSError, subprocess.TimeoutExpired): + subprocess.run(["tmux", *resource.cleanup_args()], timeout=2) + + +def run(directory: Path, owner_fd: int) -> None: + def owner_alive() -> bool: + try: + return os.read(owner_fd, 1) != b"" + except BlockingIOError: + return True + + # Keep the pane alive until the owner has its ID, even if exec will fail immediately. + deadline = time.monotonic() + 10 + while not (directory / "start").exists(): + if not owner_alive() or time.monotonic() > deadline: + return + time.sleep(0.02) + launch = json.loads((directory / "launch.json").read_text()) + (directory / "launch.json").unlink() + env = launch["env"] + # tmux supplies these for this pane; the parent may have different terminal coordinates. + for key in ("TERM", "TMUX", "TMUX_PANE"): + if key in os.environ: + env[key] = os.environ[key] + guardian = _start_guardian(grace_period=0.5) + child = None + try: + try: + child = subprocess.Popen( + launch["command"], cwd=launch["cwd"], env=env, process_group=guardian.pid, + ) + except OSError as error: + write_status(directory, {"error": str(error)}) + while owner_alive(): + time.sleep(0.02) + return + # The harness and guardian share a foreground group. Only the guardian signals its + # own group, so cleanup never sends a signal to a recycled numeric process-group ID. + signal.signal(signal.SIGTTOU, signal.SIG_IGN) + os.tcsetpgrp(0, guardian.pid) + os.write(guardian.control_fd, b"C") # Resume a fast reader stopped before the handoff. + write_status(directory, {"pid": child.pid, "returncode": None}) + while child.poll() is None and not (directory / "stop").exists() and owner_alive(): + time.sleep(0.02) + if child.returncode is not None: + write_status(directory, {"pid": child.pid, "returncode": child.returncode}) + # Retain the real screen and ownership of any helpers until the owner closes. + while owner_alive() and not (directory / "stop").exists(): + time.sleep(0.02) + finally: + # The grace period is independent of whether the CLI leader has already exited. + _send_guardian_command(guardian, _STOP_GROUP) + if child is not None: + returncode = child.wait() + with contextlib.suppress(FileNotFoundError): + write_status(directory, { + "pid": child.pid, "returncode": returncode, "stopped": True, + }) + + # Let the controller consume the final status before removing the private directory. + while owner_alive(): + time.sleep(0.02) + + +if __name__ == "__main__": + main() diff --git a/src/agent_shell/process_cleanup.py b/src/agent_shell/process_cleanup.py index 30d9e15..7139192 100644 --- a/src/agent_shell/process_cleanup.py +++ b/src/agent_shell/process_cleanup.py @@ -13,40 +13,16 @@ """ import asyncio import atexit -import contextlib import inspect -import logging -import os -import subprocess -import sys -from dataclasses import dataclass from typing import Protocol -logger = logging.getLogger("agent_shell.process_cleanup") - -_KILL_GROUP = b"K" -_RELEASE_GROUP = b"R" - -_GROUP_GUARDIAN = """ -import os -import signal - -command = os.read(0, 1) -if command == b"R": - raise SystemExit(0) -os.kill(0, signal.SIGKILL) -""" - - -@dataclass(slots=True) -class _GroupGuardian: - process: subprocess.Popen - control_fd: int - - @property - def pid(self) -> int: - return self.process.pid - +from agent_shell.process_guardian import ( + _GroupGuardian, + _KILL_GROUP, + _RELEASE_GROUP, + _send_guardian_command, + _start_guardian, +) # Run handles have stable identity even after their numeric PIDs are reaped and reused. _guardians: dict[object, _GroupGuardian] = {} @@ -67,43 +43,6 @@ def transfer_process_guardian(process: object, run_handle: object) -> None: _guardians[run_handle] = guardian -def _send_guardian_command(guardian: _GroupGuardian, command: bytes) -> None: - try: - os.write(guardian.control_fd, command) - except OSError as error: - logger.warning("Could not contact process-group guardian: %s", error) - finally: - with contextlib.suppress(OSError): - os.close(guardian.control_fd) - - # subprocess.Popen, rather than asyncio, owns this direct child. Waiting here reaps that - # exact child; the PID is never used to choose a process or group to signal. - with contextlib.suppress(OSError, ChildProcessError): - guardian.process.wait() - - -def _start_guardian() -> _GroupGuardian: - read_fd, write_fd = os.pipe() - argv = [sys.executable, "-I", "-S", "-c", _GROUP_GUARDIAN] - - try: - process = subprocess.Popen( - argv, - stdin=read_fd, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - close_fds=True, - process_group=0, - ) - except BaseException: - os.close(write_fd) - raise - finally: - os.close(read_fd) - - return _GroupGuardian(process=process, control_fd=write_fd) - - async def create_grouped_process( command: list[str], cwd: str, diff --git a/src/agent_shell/process_guardian.py b/src/agent_shell/process_guardian.py new file mode 100644 index 0000000..ed0085a --- /dev/null +++ b/src/agent_shell/process_guardian.py @@ -0,0 +1,98 @@ +"""Stdlib process-group guardian shared by native runs and standalone terminal workers.""" + +import contextlib +from dataclasses import dataclass +import logging +import os +import subprocess +import sys + +logger = logging.getLogger("agent_shell.process_cleanup") + +_KILL_GROUP = b"K" +_RELEASE_GROUP = b"R" +_STOP_GROUP = b"S" + +_GROUP_GUARDIAN = """ +import os +import signal +import sys +import time + +# Only the guardian ignores these signals. The CLI is a sibling with its own dispositions. +for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP, + signal.SIGTSTP, signal.SIGTTIN, signal.SIGTTOU): + signal.signal(signum, signal.SIG_IGN) +os.write(1, b"R") +os.dup2(2, 1) # Restore /dev/null and close the readiness pipe before the parent proceeds. +grace_period = float(sys.argv[1]) +while True: + command = os.read(0, 1) + if command == b"C": + os.kill(0, signal.SIGCONT) + continue + if command == b"R": + raise SystemExit(0) + if command != b"K" and grace_period: + os.kill(0, signal.SIGCONT) + os.kill(0, signal.SIGTERM) + time.sleep(grace_period) + os.kill(0, signal.SIGKILL) +""" + + +@dataclass(slots=True) +class _GroupGuardian: + process: subprocess.Popen + control_fd: int + + @property + def pid(self) -> int: + return self.process.pid + + +def _send_guardian_command(guardian: _GroupGuardian, command: bytes) -> None: + try: + os.write(guardian.control_fd, command) + except OSError as error: + logger.warning("Could not contact process-group guardian: %s", error) + finally: + with contextlib.suppress(OSError): + os.close(guardian.control_fd) + + # subprocess.Popen, rather than asyncio, owns this direct child. Waiting here reaps that + # exact child; the PID is never used to choose a process or group to signal. + with contextlib.suppress(OSError, ChildProcessError): + guardian.process.wait() + + +def _start_guardian(*, grace_period: float = 0.0) -> _GroupGuardian: + read_fd, write_fd = os.pipe() + argv = [sys.executable, "-I", "-S", "-c", _GROUP_GUARDIAN, str(grace_period)] + + try: + process = subprocess.Popen( + argv, + stdin=read_fd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + close_fds=True, + process_group=0, + ) + except BaseException: + os.close(write_fd) + raise + finally: + os.close(read_fd) + + guardian = _GroupGuardian(process=process, control_fd=write_fd) + try: + # Do not launch a CLI until the guardian can survive terminal signals. + if process.stdout.read() != b"R": + raise OSError("Process-group guardian failed to start") + except BaseException: + _send_guardian_command(guardian, _KILL_GROUP) + raise + finally: + process.stdout.close() + return guardian diff --git a/src/agent_shell/shell.py b/src/agent_shell/shell.py index 7ce0bc4..4a12e87 100644 --- a/src/agent_shell/shell.py +++ b/src/agent_shell/shell.py @@ -84,6 +84,32 @@ def _resolve_adapter(self, agent_type: AgentType) -> AgentAdapter: isolation_policy=self.isolation_policy, ) + async def open_interactive( + self, cwd: str, *, prompt: str | None = None, model: str | None = None, + effort: str | None = None, session_id: str | None = None, + allowed_tools: list[str] | None = None, + ): + """Open the real harness UI (experimental). The caller owns the returned session. + + Requires an interactive-capable host and adapter. Native harness permission prompts + remain enabled. Structured feature support is exposed by session.capabilities. + """ + from agent_shell.interactive import ( + InteractiveAdapter, InteractiveExecutionHost, open_interactive_session, + ) + + if not Path(cwd).is_dir(): + raise ValueError(f"Directory does not exist: {cwd}") + if not isinstance(self._adapter, InteractiveAdapter): + raise NotImplementedError("This adapter does not support interactive sessions") + if not isinstance(self.execution_host, InteractiveExecutionHost): + raise NotImplementedError("This execution host does not support interactive sessions") + return await open_interactive_session( + self._adapter, self.execution_host, self.isolation_policy, cwd, + prompt=prompt, model=model, effort=effort, + session_id=session_id, allowed_tools=allowed_tools, + ) + async def execute( self, cwd: str, @@ -154,12 +180,13 @@ async def health_check( cwd: str, model: str | None = None, timeout: float = 60.0, + *, effort: str | None = None, ) -> HealthCheckResult: if not Path(cwd).is_dir(): raise ValueError(f"Directory does not exist: {cwd}") - return await self._adapter.health_check(cwd=cwd, model=model, timeout=timeout) + return await self._adapter.health_check(cwd=cwd, model=model, timeout=timeout, effort=effort) async def list_models( self, diff --git a/src/agent_shell/tmux.py b/src/agent_shell/tmux.py index c422726..71c30d4 100644 --- a/src/agent_shell/tmux.py +++ b/src/agent_shell/tmux.py @@ -17,6 +17,7 @@ from typing import Literal from agent_shell import tmux_protocol +from agent_shell.tmux_ownership import IDENTITY_FORMAT, TmuxResource from agent_shell.execution import ( IsolationPolicy, IsolationUnavailableError, @@ -37,12 +38,14 @@ class TmuxPlacement: Placement is deliberately separate from cleanup ownership. A ``new-session`` placement creates a session that the resulting run owns, while a ``new-window`` placement borrows the - named session and owns only the window created for that run. + named session and owns only the window created for that run. A ``split-pane`` placement + borrows the caller's window and owns only its new pane. """ - _kind: Literal["new-session", "new-window", "current-window"] + _kind: Literal["new-session", "new-window", "current-window", "split-pane"] _session_name: str | None = None _focus: bool = False + _direction: Literal["right", "down"] = "right" @classmethod def new_session(cls, name: str | None = None) -> TmuxPlacement: @@ -78,10 +81,29 @@ def current_session(cls, focus: bool = False) -> TmuxPlacement: return cls(_kind="current-window", _focus=focus) @property - def kind(self) -> Literal["new-session", "new-window", "current-window"]: + def kind(self) -> Literal["new-session", "new-window", "current-window", "split-pane"]: """The resource creation operation represented by this placement.""" return self._kind + @classmethod + def split_pane( + cls, focus: bool = False, *, direction: Literal["right", "down"] = "right", + ) -> TmuxPlacement: + """Split right of or below the caller's TMUX_PANE, owning only the new pane. + + By default keyboard focus stays with the caller. Requires running inside tmux. + """ + if not isinstance(focus, bool): + raise TypeError("focus must be a bool") + if direction not in ("right", "down"): + raise ValueError("direction must be 'right' or 'down'") + return cls(_kind="split-pane", _focus=focus, _direction=direction) + + @property + def direction(self) -> Literal["right", "down"]: + """Where a split pane is placed relative to the caller's pane.""" + return self._direction + @property def session(self) -> str | None: """The explicit session name, when this placement has one.""" @@ -89,7 +111,7 @@ def session(self) -> str | None: @property def focus(self) -> bool: - """Whether a newly-created window should become the active window.""" + """Whether the newly created window or pane should receive focus.""" return self._focus @@ -104,6 +126,11 @@ def _validate_tmux_name(value: str, description: str) -> None: ) +def _tmux_exact_session_target(session_name: str) -> str: + """Format a borrowed session name so tmux matches the session exactly.""" + return f"={session_name}:" + + async def _tmux_receive_frame(reader: asyncio.StreamReader) -> tuple[int, bytes]: try: return await tmux_protocol.receive_frame(reader) @@ -132,9 +159,7 @@ def __init__( self, *, tmux_path: str, - resource_kind: Literal["session", "window"], - session_name: str, - window_id: str | None, + resource: TmuxResource, run_directory: str, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, @@ -142,9 +167,7 @@ def __init__( stdin_pipe: bool, ): self._tmux_path = tmux_path - self._resource_kind = resource_kind - self._session_name = session_name - self._window_id = window_id + self._resource = resource self._run_directory = run_directory self._reader = reader self._writer = writer @@ -274,22 +297,16 @@ def __del__(self): _TMUX_ACTIVE_RUNS.discard(self) def _cleanup_resource(self) -> None: - if self._resource_kind == "session": - _tmux_kill_session(self._tmux_path, self._session_name) - elif self._window_id: - _tmux_kill_window(self._tmux_path, self._window_id) + _cleanup_tmux_resource(self._tmux_path, self._resource) -def _cleanup_tmux_resource( - tmux_path: str, - resource_kind: Literal["session", "window"], - session_name: str, - window_id: str | None, -) -> None: - if resource_kind == "session": - _tmux_kill_session(tmux_path, session_name) - elif window_id: - _tmux_kill_window(tmux_path, window_id) +def _cleanup_tmux_resource(tmux_path: str, resource: TmuxResource) -> None: + with contextlib.suppress(OSError, subprocess.TimeoutExpired): + subprocess.run( + [tmux_path, "-f", "/dev/null", *resource.cleanup_args()], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + check=False, timeout=2.0, + ) class _TmuxStdin: @@ -325,30 +342,6 @@ async def wait_closed(self) -> None: await self.drain() -def _tmux_kill_session(tmux_path: str, session_name: str) -> None: - with contextlib.suppress(OSError, subprocess.TimeoutExpired): - subprocess.run( - [tmux_path, "-f", "/dev/null", "kill-session", "-t", session_name], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False, - timeout=2.0, - ) - - -def _tmux_kill_window(tmux_path: str, window_id: str) -> None: - with contextlib.suppress(OSError, subprocess.TimeoutExpired): - subprocess.run( - [tmux_path, "-f", "/dev/null", "kill-window", "-t", window_id], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False, - timeout=2.0, - ) - - async def _tmux_current_session(tmux_path: str) -> str: tmux_pane = os.environ.get("TMUX_PANE") if not os.environ.get("TMUX") or not tmux_pane: @@ -390,7 +383,7 @@ async def _tmux_current_session(tmux_path: str) -> str: class TmuxExecutionHost: - """Run one command in an AgentShell-owned tmux session or window. + """Run one command in an AgentShell-owned tmux session, window, or pane. .. warning:: This execution host is experimental. Its placement and lifecycle contract may change in a @@ -406,6 +399,17 @@ def __init__(self, placement: TmuxPlacement | None = None): raise TypeError("placement must be a TmuxPlacement") self.placement = placement + async def launch_interactive( + self, command: list[str], cwd: str, *, env: dict[str, str] | None = None, + isolation_policy: IsolationPolicy | None = None, + ): + """Launch directly on a real terminal (experimental, separate from pipe-based launch).""" + from agent_shell.interactive_terminal import TmuxTerminalSession + + return await TmuxTerminalSession.launch( + command, cwd, placement=self.placement, env=env, isolation_policy=isolation_policy, + ) + async def launch( self, command: list[str], @@ -434,17 +438,18 @@ async def launch( prepared = await policy.prepare(command, env) placement = self.placement or TmuxPlacement.new_session() - if placement.kind == "current-window": + if placement.kind in {"current-window", "split-pane"}: session_name = await _tmux_current_session(tmux_path) else: session_name = placement.session or f"agentshell-{uuid.uuid4().hex}" run_directory = tempfile.mkdtemp(prefix="agentshell-tmux-") socket_path = os.path.join(run_directory, "bridge.sock") - window_id: str | None = None - resource_kind: Literal["session", "window"] = ( - "session" if placement.kind == "new-session" else "window" + resource: TmuxResource | None = None + resource_kind: Literal["session", "window", "pane"] = ( + "session" if placement.kind == "new-session" + else "pane" if placement.kind == "split-pane" else "window" ) - resource_label = "session" if resource_kind == "session" else "window" + resource_label = resource_kind connection: asyncio.Future[tuple[asyncio.StreamReader, asyncio.StreamWriter]] = ( asyncio.get_running_loop().create_future() ) @@ -461,7 +466,6 @@ async def accept_connection( connection.set_result((reader, writer)) server: asyncio.Server | None = None - resource_created = False try: server = await asyncio.start_unix_server(accept_connection, path=socket_path) tmux_command = [tmux_path, "-f", "/dev/null"] @@ -474,19 +478,25 @@ async def accept_connection( session_name, "-P", "-F", - "#{pane_id}", + IDENTITY_FORMAT, ] ) + elif placement.kind == "split-pane": + tmux_command.extend([ + "split-window", "-h" if placement.direction == "right" else "-v", + *([] if placement.focus else ["-d"]), + "-t", os.environ["TMUX_PANE"], "-P", "-F", IDENTITY_FORMAT, + ]) else: tmux_command.extend( [ "new-window", *([] if placement.focus else ["-d"]), "-t", - session_name, + _tmux_exact_session_target(session_name), "-P", "-F", - "#{window_id}", + IDENTITY_FORMAT, ] ) tmux_command.extend( @@ -521,13 +531,12 @@ async def accept_connection( raise TmuxUnavailableError( f"tmux could not create the AgentShell {resource_label}{suffix}" ) - resource_created = True - if placement.kind != "new-session": - window_id = stdout.decode("utf-8", errors="replace").strip() - if not window_id: - raise TmuxUnavailableError( - "tmux created a run window but did not report its window id" - ) + try: + resource = TmuxResource.from_identity( + resource_kind, stdout.decode("utf-8", errors="replace"), run_directory, + ) + except ValueError as error: + raise TmuxUnavailableError(str(error)) from error reader, writer = await asyncio.wait_for(connection, timeout=5.0) channel, payload = await asyncio.wait_for( @@ -551,9 +560,7 @@ async def accept_connection( await _tmux_send_frame(writer, tmux_protocol.CONFIG, config) run_handle = _TmuxRunHandle( tmux_path=tmux_path, - resource_kind=resource_kind, - session_name=session_name, - window_id=window_id, + resource=resource, run_directory=run_directory, reader=reader, writer=writer, @@ -578,10 +585,8 @@ async def accept_connection( if server is not None: server.close() if not handed_off: - if resource_created: - _cleanup_tmux_resource( - tmux_path, resource_kind, session_name, window_id - ) + if resource is not None: + _cleanup_tmux_resource(tmux_path, resource) shutil.rmtree(run_directory, ignore_errors=True) diff --git a/src/agent_shell/tmux_ownership.py b/src/agent_shell/tmux_ownership.py new file mode 100644 index 0000000..1d09019 --- /dev/null +++ b/src/agent_shell/tmux_ownership.py @@ -0,0 +1,39 @@ +"""Shared tmux resource identity. Kept stdlib-only for standalone workers.""" + +from dataclasses import dataclass +from pathlib import Path +import shlex + + +IDENTITY_FORMAT = "#{session_id}\t#{window_id}\t#{pane_id}\t#{socket_path}" + + +@dataclass(frozen=True) +class TmuxResource: + kind: str + session_id: str + window_id: str + pane_id: str + socket_path: str + marker: str + + @classmethod + def from_identity(cls, kind: str, identity: str, directory: str | Path): + fields = identity.strip().split("\t") + if len(fields) != 4 or any( + not value.startswith(prefix) or not value[1:].isdigit() + for value, prefix in zip(fields[:3], ("$", "@", "%")) + ) or not fields[3]: + raise ValueError("tmux did not report valid session, window, pane and socket IDs") + return cls(kind, *fields, Path(directory).name) + + def cleanup_args(self) -> list[str]: + target = {"session": self.session_id, "window": self.window_id, "pane": self.pane_id} + # The unique run directory is already in the pane's creation command. Check it on + # the server before deleting: numeric IDs may be reused after a server restart. + # No shell is involved in either the check or the conditional tmux command. + return [ + "-S", self.socket_path, "if-shell", "-F", "-t", self.pane_id, + "#{m:*" + self.marker + "*,#{pane_start_command}}", + f"kill-{self.kind} -t {shlex.quote(target[self.kind])}", + ] diff --git a/tests/e2e/test_codex_e2e.py b/tests/e2e/test_codex_e2e.py index 50378c6..daa8660 100644 --- a/tests/e2e/test_codex_e2e.py +++ b/tests/e2e/test_codex_e2e.py @@ -7,8 +7,8 @@ pytestmark = pytest.mark.e2e -# Codex E2E uses gpt-5.4-mini explicitly to keep token costs low. -MODEL = "gpt-5.4-mini" +# Codex E2E uses gpt-5.6-luna explicitly to keep token costs low. +MODEL = "gpt-5.6-luna" class TestStreamE2E: @@ -22,6 +22,7 @@ async def test_stream_yields_text_and_result_events(self): cwd="/tmp", prompt="Reply with exactly the word PONG and nothing else.", model=MODEL, + effort="low", ): events.append(event) @@ -45,6 +46,7 @@ async def test_execute_returns_response_with_text_and_session_id(self): cwd="/tmp", prompt="Reply with exactly the word PONG and nothing else.", model=MODEL, + effort="low", ) # Assert @@ -72,17 +74,20 @@ async def test_resume_returns_the_same_session_id(self): cwd="/tmp", prompt="Reply with just 'OK'.", model=MODEL, + effort="low", ) resumed = await shell.execute( cwd="/tmp", prompt="Reply with just 'OK'.", model=MODEL, + effort="low", session_id=first.session_id, ) fresh = await shell.execute( cwd="/tmp", prompt="Reply with just 'OK'.", model=MODEL, + effort="low", ) # Assert @@ -104,6 +109,7 @@ async def test_web_search_deny_config_is_accepted_by_codex(self): cwd="/tmp", prompt="Reply with exactly the word PONG and nothing else.", model=MODEL, + effort="low", disallowed_tools=["web_search"], ): events.append(event) @@ -130,6 +136,7 @@ async def test_execute_reports_output_tokens(self): cwd="/tmp", prompt="Write a short paragraph about the sea.", model=MODEL, + effort="low", ) # Assert diff --git a/tests/e2e/test_execution_host_e2e.py b/tests/e2e/test_execution_host_e2e.py index b85e404..7dfba7d 100644 --- a/tests/e2e/test_execution_host_e2e.py +++ b/tests/e2e/test_execution_host_e2e.py @@ -1,6 +1,6 @@ """Real-CLI smoke coverage for the opt-in execution boundary. -Local-only: this uses the authenticated Codex CLI and incurs a small gpt-5.4-mini call per mode. +Local-only: this uses the authenticated Codex CLI and incurs a small gpt-5.6-luna call per mode. """ import asyncio @@ -284,7 +284,9 @@ async def test_real_codex_health_check_completes_inside_pid_isolation(tmp_path, ) # Act - result = await shell.health_check(cwd=str(tmp_path), model="gpt-5.4-mini", timeout=60.0) + result = await shell.health_check( + cwd=str(tmp_path), model="gpt-5.6-luna", effort="low", timeout=60.0, + ) # Assert assert result.healthy is True, result.exception diff --git a/tests/e2e/test_health_check_e2e.py b/tests/e2e/test_health_check_e2e.py index f0a240e..bdcdd18 100644 --- a/tests/e2e/test_health_check_e2e.py +++ b/tests/e2e/test_health_check_e2e.py @@ -17,7 +17,7 @@ VALID_MODEL = { AgentType.CLAUDE_CODE: "haiku", AgentType.OPENCODE: "opencode/big-pickle", - AgentType.CODEX: "gpt-5.4-mini", + AgentType.CODEX: "gpt-5.6-luna", AgentType.COPILOT_CLI: "auto", AgentType.PI: "openai-codex/gpt-5.4-mini", AgentType.CURSOR: "auto", @@ -34,7 +34,10 @@ async def test_valid_model_is_healthy(self, agent_type): shell = AgentShell(agent_type=agent_type) # Act - result = await shell.health_check(cwd="/tmp", model=VALID_MODEL[agent_type]) + result = await shell.health_check( + cwd="/tmp", model=VALID_MODEL[agent_type], + effort="low" if agent_type == AgentType.CODEX else None, + ) # Assert assert isinstance(result, HealthCheckResult) @@ -49,7 +52,10 @@ async def test_bogus_model_is_unhealthy(self, agent_type): shell = AgentShell(agent_type=agent_type) # Act - result = await shell.health_check(cwd="/tmp", model=BOGUS_MODEL) + result = await shell.health_check( + cwd="/tmp", model=BOGUS_MODEL, + effort="low" if agent_type == AgentType.CODEX else None, + ) # Assert — every CLI's bad-model path must resolve to unhealthy, despite # opencode exiting 0 and copilot/pi reporting only on stderr. diff --git a/tests/e2e/test_interactive_harness_e2e.py b/tests/e2e/test_interactive_harness_e2e.py new file mode 100644 index 0000000..b07d3bb --- /dev/null +++ b/tests/e2e/test_interactive_harness_e2e.py @@ -0,0 +1,136 @@ +"""Local-only real harness checks. Prompt tests use the configured account and incur usage. + +The working directory must already be trusted in each harness. Set +AGENTSHELL_E2E_TRUST_WORKSPACE=1 to accept Copilot's folder trust for this session only. +Tool permission dialogs are never auto-approved. +""" + +import asyncio +import os +from pathlib import Path +import shutil + +import pytest + +from agent_shell import TmuxExecutionHost, TmuxPlacement +from agent_shell.models.agent import AgentType +from agent_shell.shell import AgentShell +from tests.integration.test_interactive_terminal import isolated_tmux # noqa: F401 +from tests.integration.test_interactive_terminal import current_tmux_terminal # noqa: F401 + +pytestmark = pytest.mark.e2e + +HARNESSES = [ + (AgentType.CODEX, "codex"), (AgentType.CLAUDE_CODE, "claude"), (AgentType.PI, "pi"), + (AgentType.CURSOR, "cursor-agent"), (AgentType.COPILOT_CLI, "copilot"), + (AgentType.OPENCODE, "opencode"), (AgentType.GROK, "grok"), +] + + +@pytest.mark.parametrize("agent,binary", HARNESSES, ids=[a.value for a, _ in HARNESSES]) +@pytest.mark.parametrize("split", [False, True], ids=["session", "split"]) +async def test_real_interactive_prompt( + agent, binary, isolated_tmux, current_tmux_terminal, split, +): + # Arrange: native auth/config and a trusted workspace; unique answer absent from prompt. + if not shutil.which(binary): + pytest.skip(f"{binary} is not installed") + cwd = os.environ.get("AGENTSHELL_E2E_CWD", str(Path(__file__).resolve().parents[2])) + placement = TmuxPlacement.split_pane() if split else TmuxPlacement.new_session() + shell = AgentShell(agent, execution_host=TmuxExecutionHost(placement)) + prompt = ( + "Reply only with the concatenation of AGENTSHELL_ and E2E_OK. " + "Do not use tools or modify files." + ) + events = [] + + async def observe(session): + async for event in session.events(): + events.append(event) + + # Act + options = {"model": "gpt-5.6-luna", "effort": "low"} if agent == AgentType.CODEX else {} + async with await shell.open_interactive(cwd, prompt=prompt, **options) as session: + observer = asyncio.create_task(observe(session)) + try: + async with asyncio.timeout(120): + while True: + screen = await session.terminal.capture_screen() + if (agent == AgentType.COPILOT_CLI + and os.environ.get("AGENTSHELL_E2E_TRUST_WORKSPACE") == "1" + and "Confirm folder trust" in screen and cwd in screen + and "1. Yes" in screen): + await session.terminal.send_key("Enter") + await asyncio.sleep(0.5) + continue + text = "".join(e.content for e in events if e.type == "text") + answer_seen = "AGENTSHELL_E2E_OK" in ( + text if "text" in session.capabilities else screen + ) + lifecycle_seen = any(e.session_id for e in events) + if "turn_complete" in session.capabilities: + lifecycle_seen = any(e.type == "result" for e in events) + if answer_seen and lifecycle_seen: + break + if session.terminal.returncode is not None: + pytest.fail(f"{agent.value} exited: {screen[-2000:]}") + await asyncio.sleep(0.2) + except TimeoutError: + pytest.fail(f"{agent.value}: no observed reply; native UI:\n{screen[-2500:]}") + finally: + observer.cancel() + await asyncio.gather(observer, return_exceptions=True) + + # Assert: a reply is observed while the actual interactive harness remains alive. + if split: + assert session.terminal.window_id == current_tmux_terminal.window_id + assert session.terminal.pane_id != current_tmux_terminal.pane_id + assert answer_seen + assert any(e.session_id for e in events) + assert not any(e.type == "error" for e in events) + assert session.terminal.returncode is None + if "turn_complete" in session.capabilities: + assert any(e.type == "result" and e.content == "ok" for e in events) + assert session.terminal.closed + assert await isolated_tmux("list-panes", "-t", current_tmux_terminal.window_id, + "-F", "#{pane_id}") == current_tmux_terminal.pane_id + + +async def test_real_cursor_resumes_requested_conversation(isolated_tmux): + # Arrange: a newer decoy conversation detects accidentally resuming the latest session. + from uuid import uuid4 + if not shutil.which("cursor-agent"): + pytest.skip("cursor-agent is not installed") + cwd = os.environ.get("AGENTSHELL_E2E_CWD", str(Path(__file__).resolve().parents[2])) + shell = AgentShell(AgentType.CURSOR, execution_host=TmuxExecutionHost()) + tokens = ["amber-" + uuid4().hex[:12], "violet-" + uuid4().hex[:12]] + target_id = None + for token in tokens: + prompt = ( + f"Remember this secret phrase: {token}. " + "Reply with the concatenation of CURSOR_ and READY. Do not use tools." + ) + async with await shell.open_interactive(cwd, prompt=prompt) as session: + async with asyncio.timeout(60): + async for event in session.events(): + if event.session_id: + if target_id is None: + target_id = event.session_id + break + while "CURSOR_READY" not in await session.terminal.capture_screen(): + await asyncio.sleep(0.2) + + # Act: native Cursor emits no SessionStart on resume; observe the actual UI response. + prompt = "Reply only with the secret phrase I asked you to remember. Do not use tools." + async with await shell.open_interactive(cwd, prompt=prompt, session_id=target_id) as session: + try: + async with asyncio.timeout(60): + while tokens[0] not in (screen := await session.terminal.capture_screen()): + await asyncio.sleep(0.2) + except TimeoutError: + pytest.fail(f"Cursor resume did not recall the target conversation:\n{screen}") + + # Assert: correct remembered text, not the newer decoy conversation's phrase. + assert tokens[0] in screen + assert tokens[1] not in screen + assert session.terminal.returncode is None diff --git a/tests/integration/test_health_check_integration.py b/tests/integration/test_health_check_integration.py index 651e756..c9290a7 100644 --- a/tests/integration/test_health_check_integration.py +++ b/tests/integration/test_health_check_integration.py @@ -225,3 +225,21 @@ async def test_rejects_missing_cwd(self): # Act / Assert with pytest.raises(ValueError, match="Directory does not exist"): await shell.health_check(cwd="/does/not/exist", model="m") + + +async def test_codex_health_check_overrides_inherited_reasoning_effort(): + # Arrange: the external CLI must receive both parts of the requested model configuration. + shell = AgentShell(agent_type=AgentType.CODEX) + process = _make_mock_process(HAPPY[AgentType.CODEX]) + + # Act + with patch("asyncio.create_subprocess_exec", return_value=process) as launch: + result = await shell.health_check( + cwd="/tmp", model="gpt-5.6-luna", effort="low", + ) + + # Assert + assert result.healthy, result.exception + command = launch.call_args.args + assert command[command.index("--model") + 1] == "gpt-5.6-luna" + assert 'model_reasoning_effort="low"' in command diff --git a/tests/integration/test_interactive_agents.py b/tests/integration/test_interactive_agents.py new file mode 100644 index 0000000..1977f01 --- /dev/null +++ b/tests/integration/test_interactive_agents.py @@ -0,0 +1,549 @@ +"""AgentShell interactive flow with real tmux and controlled external CLI fixtures.""" + +import asyncio +import os +from pathlib import Path +import shutil +import sys +from contextlib import aclosing + +import pytest + +from agent_shell import TmuxExecutionHost +from agent_shell.models.agent import AgentType +from agent_shell.shell import AgentShell +from tests.integration.test_interactive_terminal import isolated_tmux, screen_contains # noqa: F401 +from tests.integration.test_interactive_terminal import current_tmux_terminal # noqa: F401 + + +def install_cli(tmp_path, monkeypatch, name, source): + binary = tmp_path / name + binary.write_text("#!/usr/bin/env python3\n" + source) + binary.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") + + +async def test_codex_real_ui_and_completion_share_one_session( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: only the external harness is substituted; terminal and notification helper are real. + install_cli(tmp_path, monkeypatch, "codex", ''' +import json, os, subprocess, sys, tomllib +assert all(os.isatty(fd) for fd in (0, 1, 2)) +assert "exec" not in sys.argv and "--json" not in sys.argv +notify = next(tomllib.loads(arg)["notify"] for arg in sys.argv if arg.startswith("notify=")) +print("CODEX UI READY", flush=True) +prompt = input() +subprocess.run(notify + [json.dumps({ + "type": "agent-turn-complete", "thread-id": "codex-session", "turn-id": "turn-1", + "last-assistant-message": "Answer: " + prompt, +})], check=True) +print("CODEX UI STILL OPEN", flush=True) +input() +''') + shell = AgentShell(AgentType.CODEX, execution_host=TmuxExecutionHost()) + + # Act + session = await shell.open_interactive(str(tmp_path)) + async with session: + await screen_contains(session.terminal, "CODEX UI READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.type == "result": + break + screen = await screen_contains(session.terminal, "CODEX UI STILL OPEN") + + # Assert + assert [(e.type, e.content) for e in events] == [ + ("system", ""), ("text", "Answer: hello"), ("result", "ok"), + ] + assert events[0].session_id == "codex-session" + assert "CODEX UI STILL OPEN" in screen + assert session.terminal.returncode is None + assert "output_tokens" not in session.capabilities + assert session.terminal.closed + + +async def test_pi_extension_waits_for_settled_and_reports_usage( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: an external Pi stand-in loads the actual extension and exercises its public hooks. + if not shutil.which("node"): + pytest.skip("Pi extension integration test requires node") + binary = tmp_path / "pi" + binary.write_text('''#!/usr/bin/env node +const {pathToFileURL} = require("node:url"); +const readline = require("node:readline"); +(async () => { + if (!process.stdin.isTTY || !process.stdout.isTTY) throw Error("not a terminal"); + if (process.argv.includes("--print")) throw Error("not interactive"); + const extension = process.argv[process.argv.indexOf("--extension") + 1]; + const handlers = new Map(); + const ctx = {sessionManager: {getSessionId: () => "pi-session"}}; + (await import(pathToFileURL(extension))).default({on: (name, fn) => handlers.set(name, fn)}); + const emit = async (type, extra = {}) => handlers.get(type)?.({type, ...extra}, ctx); + await emit("session_start"); + console.log("PI UI READY"); + const input = readline.createInterface({input: process.stdin}); + input.once("line", async () => { + await emit("agent_start"); + await emit("agent_end", {messages: [{role: "assistant", stopReason: "error", + errorMessage: "transient", usage: {output: 2, cost: {total: 0.01}}}]}); + await emit("agent_start"); + await emit("message_update", {assistantMessageEvent: {type: "text_end", content: "Pi answer"}}); + await emit("agent_end", {messages: [{role: "assistant", stopReason: "stop", + usage: {output: 8, cost: {total: 0.02}}}]}); + await emit("tool_execution_start", {toolName: "read"}); + await emit("agent_settled"); + console.log("PI UI STILL OPEN"); + }); +})(); +''') + binary.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") + shell = AgentShell(AgentType.PI, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "PI UI READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.type == "result": + break + + # Assert: the failed retry must not be mistaken for final completion or lose its usage. + assert events[0].session_id == "pi-session" + assert [e.content for e in events if e.type == "text"] == ["Pi answer"] + assert events[-1].content == "ok" + assert all(e.session_id == "pi-session" for e in events) + assert any(e.type == "tool_use" for e in events) + assert events[-1].output_tokens == 10 + assert events[-1].cost == pytest.approx(0.03) + assert {"turn_complete", "output_tokens", "cost"} <= session.capabilities + assert session.terminal.returncode is None + assert session.terminal.closed + + +async def test_claude_hooks_preserve_ui_and_do_not_claim_early_success( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + install_cli(tmp_path, monkeypatch, "claude", ''' +import json, os, subprocess, sys +assert all(os.isatty(fd) for fd in (0, 1, 2)) +assert "--print" not in sys.argv and "-p" not in sys.argv +settings = json.load(open(sys.argv[sys.argv.index("--settings") + 1])) +# Claude Code 2.1.77 rejects the entire file if any newer hook name is present. +assert set(settings["hooks"]) <= {"SessionStart", "UserPromptSubmit", "PreToolUse", "Stop"} +def emit(name, **extra): + hook = settings["hooks"][name][0]["hooks"][0]["command"] + subprocess.run(hook, shell=True, check=True, input=json.dumps({ + "hook_event_name": name, "session_id": "claude-session", **extra, + }).encode()) +emit("SessionStart") +print("CLAUDE UI READY", flush=True) +input() +emit("Stop", last_assistant_message="Provisional answer", stop_hook_active=False) +print("CLAUDE UI STILL OPEN", flush=True) +input() +''') + shell = AgentShell(AgentType.CLAUDE_CODE, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "CLAUDE UI READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.type == "status": + break + + # Assert + assert [(e.type, e.content) for e in events] == [ + ("system", ""), ("text", "Provisional answer"), + ("status", "stop_requested"), + ] + assert events[0].session_id == "claude-session" + assert "turn_complete" not in session.capabilities + assert "output_tokens" not in session.capabilities + assert session.terminal.returncode is None + + +async def test_reopening_event_reader_does_not_drop_pending_events( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + install_cli(tmp_path, monkeypatch, "codex", ''' +import json, subprocess, sys, time, tomllib +notify = next(tomllib.loads(arg)["notify"] for arg in sys.argv if arg.startswith("notify=")) +subprocess.run(notify + [json.dumps({"type": "agent-turn-complete", + "thread-id": "session", "last-assistant-message": "retained answer"})], check=True) +time.sleep(60) +''') + shell = AgentShell(AgentType.CODEX, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + async with asyncio.timeout(3): + async with aclosing(session.events()) as reader: + first = await anext(reader) + async with aclosing(session.events()) as reader: + second = await anext(reader) + third = await anext(reader) + + # Assert + assert first.type == "system" + assert second.content == "retained answer" + assert third.type == "result" + + +@pytest.mark.parametrize("split", [False, True]) +async def test_demo_controller_sends_prompt_and_cleans_up( + current_tmux_terminal, isolated_tmux, tmp_path, monkeypatch, split, +): + # Arrange + install_cli(tmp_path, monkeypatch, "codex", ''' +import json, subprocess, sys, tomllib +notify = next(tomllib.loads(arg)["notify"] for arg in sys.argv if arg.startswith("notify=")) +for prompt in sys.stdin: + subprocess.run(notify + [json.dumps({"type": "agent-turn-complete", + "thread-id": "demo", "last-assistant-message": "Demo received " + prompt.strip()})]) +''') + demo = Path(__file__).parents[2] / "examples" / "interactive_demo.py" + process = await asyncio.create_subprocess_exec( + sys.executable, str(demo), "--agent", "codex", "--cwd", str(tmp_path), + *(["--split-pane"] if split else []), + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + output = [] + + # Act + try: + async with asyncio.timeout(5): + while line := await process.stdout.readline(): + output.append(line.decode()) + if b"Controller ready" in line: + process.stdin.write(b"codex hello\n") + await process.stdin.drain() + if b"Demo received hello" in line: + process.stdin.write(b"/quit\n") + await process.stdin.drain() + await process.wait() + stderr = (await process.stderr.read()).decode() + finally: + if process.returncode is None: + process.kill() + await process.wait() + + # Assert + assert process.returncode == 0, stderr + assert any("Demo received hello" in line for line in output) + assert any("Closed all demo sessions" in line for line in output) + assert await isolated_tmux("list-panes", "-t", current_tmux_terminal.window_id, + "-F", "#{pane_id}") == current_tmux_terminal.pane_id + + +async def test_cursor_plugin_reports_only_verified_session_metadata( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: exercise the real plugin files and hook writer in an external CLI fixture. + install_cli(tmp_path, monkeypatch, "cursor-agent", ''' +import json, os, pathlib, subprocess, sys +assert all(os.isatty(fd) for fd in (0, 1, 2)) +assert "--print" not in sys.argv +plugin = pathlib.Path(sys.argv[sys.argv.index("--plugin-dir") + 1]) +# Cursor refuses extension directories with fewer than three non-root path components. +assert len(plugin.resolve().parts) >= 4 +assert json.loads((plugin / ".cursor-plugin/plugin.json").read_text())["name"] +hooks = json.loads((plugin / "hooks/hooks.json").read_text())["hooks"] +def emit(name, **data): + for hook in hooks[name]: + subprocess.run(hook["command"], shell=True, check=True, input=json.dumps({ + "hook_event_name": name, "conversation_id": "cursor-session", **data, + }), text=True) +print("CURSOR READY", flush=True) +input() +emit("sessionStart") +print("CURSOR STILL OPEN", flush=True) +input() +''') + shell = AgentShell(AgentType.CURSOR, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "CURSOR READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.type == "system": + break + + # Assert: native CLI turn hooks are unavailable, so no turn/text capability is promised. + assert [(e.type, e.content) for e in events] == [ + ("system", ""), + ] + assert events[0].session_id == "cursor-session" + assert session.capabilities == frozenset({"session_id"}) + assert session.terminal.returncode is None + + +async def test_copilot_plugin_reports_lifecycle_and_tools( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + install_cli(tmp_path, monkeypatch, "copilot", ''' +import json, os, pathlib, subprocess, sys +assert all(os.isatty(fd) for fd in (0, 1, 2)) +assert "-p" not in sys.argv and "--headless" not in sys.argv +plugin = pathlib.Path(sys.argv[sys.argv.index("--plugin-dir") + 1]) +hooks = json.loads((plugin / "hooks/hooks.json").read_text())["hooks"] +def emit(name, **data): + for hook in hooks[name]: + subprocess.run(hook["command"], shell=True, check=True, input=json.dumps({ + "hook_event_name": name, "session_id": "copilot-session", **data, + }), text=True) +print("COPILOT READY", flush=True) +input() +emit("SessionStart") +emit("PostToolUse", tool_name="Read") +emit("Stop", stop_reason="end_turn") +input() +''') + shell = AgentShell(AgentType.COPILOT_CLI, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "COPILOT READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.content == "stop_requested": + break + + # Assert + assert [(e.type, e.content) for e in events] == [ + ("system", ""), ("tool_use", "Read"), ("status", "stop_requested"), + ] + assert events[0].session_id == "copilot-session" + assert "turn_complete" not in session.capabilities + + +async def test_opencode_plugin_reports_completed_assistant_text( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: Node loads the actual OpenCode plugin; only the harness is substituted. + if not shutil.which("node"): + pytest.skip("OpenCode plugin test requires node") + binary = tmp_path / "opencode" + binary.write_text('''#!/usr/bin/env node +const readline = require("node:readline"); +(async () => { + if (!process.stdin.isTTY) throw Error("not a terminal"); + const config = JSON.parse(process.env.OPENCODE_CONFIG_CONTENT); + if (config.permission.read !== "allow") throw Error("lost inherited config"); + const plugin = (await import(config.plugin.at(-1))).default; + const hooks = await plugin({}); + console.log("OPENCODE READY"); + readline.createInterface({input: process.stdin}).once("line", async () => { + const emit = (type, properties) => hooks.event({event: {type, properties}}); + await emit("session.created", {info: {id: "oc-session"}}); + await emit("message.updated", {info: {id: "user", role: "user", sessionID: "oc-session"}}); + await emit("message.part.updated", {part: {id: "u1", messageID: "user", type: "text", + sessionID: "oc-session", text: "Do not echo this", time: {end: 1}}}); + await emit("message.updated", {info: {id: "assistant", role: "assistant", + sessionID: "oc-session"}}); + const part = {id: "a1", messageID: "assistant", sessionID: "oc-session", type: "text", + text: "OpenCode answer", time: {end: 2}}; + await emit("message.part.updated", {part}); + await emit("message.part.updated", {part}); + await emit("session.idle", {sessionID: "oc-session"}); + }); +})(); +''') + binary.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") + monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", '{"permission":{"read":"allow"}}') + shell = AgentShell(AgentType.OPENCODE, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "OPENCODE READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.content == "idle": + break + + # Assert: completed text is not echoed or duplicated, idle does not imply success. + assert [e.content for e in events if e.type == "text"] == ["OpenCode answer"] + assert events[0].session_id == "oc-session" + assert not any(e.type == "result" for e in events) + + +async def test_grok_native_log_delivers_text_and_turn_result( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: delayed creation and split writes model the real append-only session stream. + monkeypatch.setenv("GROK_HOME", str(tmp_path / "grok-home")) + install_cli(tmp_path, monkeypatch, "grok", ''' +import json, os, pathlib, sys, time +assert all(os.isatty(fd) for fd in (0, 1, 2)) +assert "-p" not in sys.argv and "stdio" not in sys.argv +sid = sys.argv[sys.argv.index("--session-id") + 1] +print("GROK READY", flush=True) +input() +path = pathlib.Path(os.environ["GROK_HOME"]) / "sessions" / "workspace" / sid / "updates.jsonl" +path.parent.mkdir(parents=True) +def emit(update): + line = json.dumps({"method": "session/update", "params": {"sessionId": sid, "update": update}}) + with path.open("a") as f: + f.write(line[:15]); f.flush(); time.sleep(0.07) + f.write(line[15:] + "\\n") +emit({"sessionUpdate": "user_message_chunk", "content": {"type": "text", "text": "hello"}}) +emit({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "Grok answer"}}) +emit({"sessionUpdate": "turn_completed", "stop_reason": "end_turn", + "usage": {"outputTokens": 12, "reasoningTokens": 4}, "elapsed_ms": 1500}) +input() +''') + shell = AgentShell(AgentType.GROK, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + await screen_contains(session.terminal, "GROK READY") + await session.terminal.send_text("hello", submit=True) + events = [] + async with asyncio.timeout(5): + async for event in session.events(): + events.append(event) + if event.type == "result": + break + + # Assert + assert [e.content for e in events if e.type == "text"] == ["Grok answer"] + assert events[-1].content == "ok" + assert events[-1].output_tokens == 12 + assert events[-1].duration == 1.5 + assert events[-1].session_id + assert session.terminal.returncode is None + + +async def test_grok_resume_skips_history_and_reports_cancelled_turn( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + import json + sid = "11111111-1111-4111-8111-111111111111" + root = tmp_path / "grok-home" + log = root / "sessions" / "workspace" / sid / "updates.jsonl" + log.parent.mkdir(parents=True) + log.write_text(json.dumps({"params": {"sessionId": sid, "update": { + "sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "OLD"}, + }}}) + "\n") + monkeypatch.setenv("GROK_HOME", str(root)) + install_cli(tmp_path, monkeypatch, "grok", ''' +import json, os, pathlib, sys +sid = sys.argv[sys.argv.index("--resume") + 1] +path = pathlib.Path(os.environ["GROK_HOME"]) / "sessions" / "workspace" / sid / "updates.jsonl" +print("GROK RESUMED", flush=True) +input() +with path.open("a") as f: + f.write(json.dumps({"params": {"sessionId": sid, "update": { + "sessionUpdate": "turn_completed", "stop_reason": "cancelled", + }}}) + "\\n") +input() +''') + shell = AgentShell(AgentType.GROK, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path), session_id=sid) as session: + await screen_contains(session.terminal, "GROK RESUMED") + await session.terminal.send_text("hello", submit=True) + async with asyncio.timeout(5), aclosing(session.events()) as events: + result = await anext(events) + + # Assert: old answer isn't replayed and cancellation never becomes success. + assert result.type == "result" + assert result.content == "error" + assert result.error == "cancelled" + assert log.exists(), "AgentShell must not delete the harness's own session history" + + +async def test_grok_interactive_review_can_restrict_native_tools( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange: the native whitelist must reach Grok without enabling blanket approval. + install_cli(tmp_path, monkeypatch, "grok", ''' +import sys +assert sys.argv[sys.argv.index("--tools") + 1] == "read_file,grep,list_dir" +assert "--always-approve" not in sys.argv +print("RESTRICTED REVIEW READY", flush=True) +input() +''') + shell = AgentShell(AgentType.GROK, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive( + str(tmp_path), allowed_tools=["read_file", "grep", "list_dir"], + ) as session: + screen = await screen_contains(session.terminal, "RESTRICTED REVIEW READY") + + # Assert + assert "RESTRICTED REVIEW READY" in screen + + +@pytest.mark.parametrize("agent", [a for a in AgentType if a != AgentType.GROK]) +async def test_unimplemented_interactive_tool_whitelist_fails_explicitly(agent, tmp_path): + # Arrange + shell = AgentShell(agent, execution_host=TmuxExecutionHost()) + + # Act / Assert: a review restriction must never be silently ignored. + with pytest.raises(NotImplementedError, match="allowed_tools"): + await shell.open_interactive(str(tmp_path), allowed_tools=["read"]) + + +@pytest.mark.parametrize("agent,binary,expected", [ + (AgentType.CURSOR, "cursor-agent", ["--model", "test-model", "--resume=test-session"]), + (AgentType.COPILOT_CLI, "copilot", ["--model", "test-model", "--resume=test-session"]), + (AgentType.OPENCODE, "opencode", ["--model", "test-model", "--session", "test-session"]), +]) +async def test_interactive_selection_and_initial_prompt_reach_native_cli( + agent, binary, expected, isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + prompt = "literal $HOME `whoami` café" + source = ( + "import sys\n" + f"expected = {expected!r}\n" + "start = sys.argv.index('--model')\n" + "assert sys.argv[start:start + len(expected)] == expected\n" + f"assert sys.argv[-1] == {prompt!r}\n" + "print('SELECTION ACCEPTED', flush=True)\ninput()\n" + ) + install_cli(tmp_path, monkeypatch, binary, source) + shell = AgentShell(agent, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive( + str(tmp_path), prompt=prompt, model="test-model", session_id="test-session", + ) as session: + screen = await screen_contains(session.terminal, "SELECTION ACCEPTED") + + # Assert + assert "SELECTION ACCEPTED" in screen + if agent == AgentType.CURSOR: + assert session.capabilities == frozenset() diff --git a/tests/integration/test_interactive_terminal.py b/tests/integration/test_interactive_terminal.py new file mode 100644 index 0000000..c774110 --- /dev/null +++ b/tests/integration/test_interactive_terminal.py @@ -0,0 +1,748 @@ +"""Real terminal behavior through the execution host's public interactive boundary.""" + +import asyncio +import fcntl +import os +import shutil +import sys +import tempfile +from pathlib import Path +import subprocess + +import pytest + +from agent_shell import TmuxExecutionHost, TmuxPlacement +from agent_shell.execution import IsolationUnavailableError, LinuxPidNamespaceIsolation +from agent_shell.models.agent import AgentType +from agent_shell.shell import AgentShell + + +@pytest.fixture +def isolated_tmux(monkeypatch): + real_tmux = shutil.which("tmux") + if not real_tmux: + pytest.skip("interactive terminal integration tests require tmux") + with tempfile.TemporaryDirectory(prefix="as-tmux-test-") as directory: + # Every command, including worker cleanup, explicitly selects this test's server. + wrapper = Path(directory) / "tmux" + wrapper.write_text( + f"#!{sys.executable}\nimport os, sys\n" + f"os.execv({real_tmux!r}, [{real_tmux!r}, '-L', 'test', *sys.argv[1:]])\n" + ) + wrapper.chmod(0o755) + monkeypatch.setenv("PATH", f"{directory}{os.pathsep}{os.environ['PATH']}") + monkeypatch.setenv("TMUX_TMPDIR", directory) + monkeypatch.delenv("TMUX", raising=False) + monkeypatch.delenv("TMUX_PANE", raising=False) + + async def command(*args): + process = await asyncio.create_subprocess_exec( + str(wrapper), *args, stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + assert process.returncode == 0, stderr.decode() + return stdout.decode().strip() + + try: + yield command + finally: + subprocess.run( + [str(wrapper), "kill-server"], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=5, + ) + + +@pytest.fixture +async def current_tmux_terminal(isolated_tmux, tmp_path, monkeypatch): + async with await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", + "print('ORIGINAL', flush=True); print('ORIGINAL:' + input(), flush=True); input()"], + str(tmp_path), + ) as original: + await screen_contains(original, "ORIGINAL") + socket = await isolated_tmux("display-message", "-p", "#{socket_path}") + monkeypatch.setenv("TMUX", f"{socket},0,0") + monkeypatch.setenv("TMUX_PANE", original.pane_id) + yield original + + +@pytest.mark.parametrize("focus", [False, True]) +@pytest.mark.parametrize("direction", [None, "right", "down"]) +async def test_split_accepts_input_and_preserves_original_pane( + current_tmux_terminal, isolated_tmux, tmp_path, focus, direction, +): + # Arrange + original = current_tmux_terminal + options = {"direction": direction} if direction is not None else {} + host = TmuxExecutionHost(TmuxPlacement.split_pane(focus=focus, **options)) + + # Act + async with await host.launch_interactive( + [sys.executable, "-c", "print('READY', flush=True); print('REPLY:' + input()); input()"], + str(tmp_path), + ) as child: + await screen_contains(child, "READY") + await child.send_text("hello split", submit=True) + screen = await screen_contains(child, "REPLY:hello split") + active = await isolated_tmux("display-message", "-p", "-t", original.window_id, + "#{pane_id}") + layout = await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_id} #{pane_left} #{pane_top}") + assert child.window_id == original.window_id + assert active == (child.pane_id if focus else original.pane_id) + panes = [line.split() for line in layout.splitlines()] + assert len(panes) == 2 + assert panes[0][0] == original.pane_id + assert panes[1][0] == child.pane_id + if direction == "down": + assert panes[0][1] == panes[1][1] + assert int(panes[0][2]) < int(panes[1][2]) + else: + assert panes[0][2] == panes[1][2] + assert int(panes[0][1]) < int(panes[1][1]) + await original.send_text("still alive", submit=True) + original_screen = await screen_contains(original, "ORIGINAL:still alive") + + # Assert + assert "REPLY:hello split" in screen + assert "ORIGINAL:still alive" in original_screen + assert await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_id}") == original.pane_id + + +async def test_split_resize_preserves_window_size(current_tmux_terminal, isolated_tmux, tmp_path): + # Arrange + original = current_tmux_terminal + window_size = await isolated_tmux("display-message", "-p", "-t", original.window_id, + "#{window_width} #{window_height}") + async with await TmuxExecutionHost(TmuxPlacement.split_pane()).launch_interactive( + [sys.executable, "-c", "input()"], str(tmp_path), + ) as child: + # Act + await child.resize(columns=30, rows=30) + + # Assert + assert await isolated_tmux("display-message", "-p", "-t", original.window_id, + "#{window_width} #{window_height}") == window_size + assert await isolated_tmux("display-message", "-p", "-t", child.pane_id, + "#{pane_width}") == "30" + + +async def test_failed_split_launch_preserves_original_pane( + current_tmux_terminal, isolated_tmux, tmp_path, +): + # Arrange + from agent_shell import TmuxUnavailableError + original = current_tmux_terminal + host = TmuxExecutionHost(TmuxPlacement.split_pane()) + await isolated_tmux("set-option", "-w", "-t", original.window_id, "remain-on-exit", "on") + + # Act + with pytest.raises(TmuxUnavailableError, match="No such file"): + await host.launch_interactive([str(tmp_path / "missing-command")], str(tmp_path)) + await original.send_text("still alive", submit=True) + + # Assert + assert "ORIGINAL:still alive" in await screen_contains(original, "ORIGINAL:still alive") + assert await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_id}") == original.pane_id + + +@pytest.mark.parametrize("interactive", [True, False]) +async def test_split_requires_current_tmux_context(isolated_tmux, tmp_path, interactive): + # Arrange + host = TmuxExecutionHost(TmuxPlacement.split_pane()) + launch = host.launch_interactive if interactive else host.launch + + # Act / Assert + from agent_shell import TmuxUnavailableError + with pytest.raises(TmuxUnavailableError, match="requires running inside tmux"): + await launch([sys.executable, "-c", "pass"], str(tmp_path)) + + +@pytest.mark.parametrize("cancel", [False, True]) +@pytest.mark.parametrize("direction", ["right", "down"]) +async def test_headless_split_preserves_original_pane( + current_tmux_terminal, isolated_tmux, tmp_path, cancel, direction, +): + # Arrange + original = current_tmux_terminal + host = TmuxExecutionHost(TmuxPlacement.split_pane(direction=direction)) + code = "import time; print('HEADLESS', flush=True); time.sleep(60)" if cancel else ( + "print('HEADLESS', flush=True)" + ) + + # Act + handle = await host.launch([sys.executable, "-c", code], str(tmp_path)) + try: + output = await asyncio.wait_for(handle.stdout.readline(), 5) + panes = await isolated_tmux("list-panes", "-t", original.window_id, "-F", "#{pane_id}") + layout = await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_left} #{pane_top}") + if cancel: + await handle.cancel() + else: + await asyncio.wait_for(handle.wait(), 5) + finally: + handle.release() + await original.send_text("still alive", submit=True) + + # Assert + assert output == b"HEADLESS\n" + assert len(panes.splitlines()) == 2 + first, second = [list(map(int, line.split())) for line in layout.splitlines()] + if direction == "down": + assert first[0] == second[0] + assert first[1] < second[1] + else: + assert first[1] == second[1] + assert first[0] < second[0] + assert "ORIGINAL:still alive" in await screen_contains(original, "ORIGINAL:still alive") + assert await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_id}") == original.pane_id + + +async def screen_contains(terminal, text): + async with asyncio.timeout(5): + while text not in (screen := await terminal.capture_screen()): + await asyncio.sleep(0.02) + return screen + + +async def test_real_terminal_accepts_input_and_reports_exit(isolated_tmux, tmp_path): + # Arrange: a real process requires a foreground controlling terminal on all three streams. + code = ( + "import os, sys; " + "assert all(os.isatty(fd) for fd in (0, 1, 2)); " + "assert os.tcgetpgrp(0) == os.getpgrp(); " + "print('READY', flush=True); " + "print('REPLY:' + input(), flush=True); " + "sys.exit(7)" + ) + host = TmuxExecutionHost() + + # Act + terminal = await host.launch_interactive([sys.executable, "-c", code], str(tmp_path)) + try: + await screen_contains(terminal, "READY") + await terminal.send_text("literal $HOME `whoami` café", submit=True) + screen = await screen_contains(terminal, "REPLY:") + status = await asyncio.wait_for(terminal.wait(), 5) + finally: + await terminal.close() + + # Assert + assert "REPLY:literal $HOME `whoami` café" in screen + assert status == 7 + assert terminal.returncode == 7 + assert terminal.closed + with pytest.raises(RuntimeError, match="closed"): + await terminal.send_text("cannot write after close") + + +async def test_resize_and_interrupt_reach_the_real_process(isolated_tmux, tmp_path): + # Arrange + code = ( + "import os, signal, time; " + "signal.signal(signal.SIGWINCH, " + "lambda *_: print('SIZE:' + str(os.get_terminal_size()), flush=True)); " + "print('READY', flush=True); time.sleep(60)" + ) + terminal = await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", code], str(tmp_path), + ) + + # Act + try: + await screen_contains(terminal, "READY") + await terminal.resize(columns=90, rows=25) + screen = await screen_contains(terminal, "SIZE:") + await terminal.send_key("C-c") + status = await asyncio.wait_for(terminal.wait(), 5) + finally: + await terminal.close() + + # Assert + assert "columns=90, lines=25" in screen + assert status != 0 + + +async def test_unsupported_isolation_fails_before_launch(tmp_path): + # Arrange / Act / Assert + with pytest.raises(IsolationUnavailableError, match="only NoIsolation"): + await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", "pass"], str(tmp_path), + isolation_policy=LinuxPidNamespaceIsolation(), + ) + + +async def test_close_kills_child_but_preserves_borrowed_session(isolated_tmux, tmp_path): + # Arrange: own a window in a session that another live terminal still uses. + original = await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", "import time; print('ORIGINAL', flush=True); time.sleep(60)"], + str(tmp_path), + ) + host = TmuxExecutionHost(TmuxPlacement.new_window(original.session_name)) + child = await host.launch_interactive( + [sys.executable, "-c", "import time; print('CHILD', flush=True); time.sleep(60)"], + str(tmp_path), + ) + + # Act + try: + await screen_contains(child, "CHILD") + await child.close() + status = await asyncio.wait_for(child.wait(), 2) + screen = await original.capture_screen() + original_status = original.returncode + finally: + await child.close() + await original.close() + + # Assert + assert status < 0 + assert "ORIGINAL" in screen + assert original_status is None + + +async def test_agentshell_interactive_exactly_targets_numeric_session( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + real_tmux = shutil.which("tmux") + session_name = "6" + create = await asyncio.create_subprocess_exec( + real_tmux, "-f", "/dev/null", "new-session", "-d", "-s", session_name, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, + ) + _, create_stderr = await create.communicate() + assert create.returncode == 0, create_stderr.decode() + + argv_file = tmp_path / "tmux-argv" + wrapper = tmp_path / "tmux" + wrapper.write_text( + f"#!{sys.executable}\n" + "import pathlib, os, sys\n" + "args = sys.argv[1:]\n" + "if 'new-window' in args:\n" + f" pathlib.Path({str(argv_file)!r}).write_text('\\0'.join(args))\n" + f"os.execv({real_tmux!r}, [{real_tmux!r}, *args])\n" + ) + wrapper.chmod(0o755) + codex = tmp_path / "codex" + codex.write_text("#!/usr/bin/env python3\nimport time\ntime.sleep(60)\n") + codex.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}{os.pathsep}{os.environ['PATH']}") + shell = AgentShell( + AgentType.CODEX, + execution_host=TmuxExecutionHost( + placement=TmuxPlacement.new_window(session=session_name) + ), + ) + interactive = None + + # Act + try: + interactive = await shell.open_interactive(str(tmp_path)) + tmux_args = argv_file.read_text().split("\0") + finally: + if interactive is not None: + await interactive.close() + cleanup = await asyncio.create_subprocess_exec( + real_tmux, "-f", "/dev/null", "kill-session", "-t", session_name, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, + ) + await cleanup.wait() + + # Assert + assert tmux_args[tmux_args.index("-t") + 1] == "=6:" + + +async def test_process_exit_finishes_event_stream_without_claiming_turn_success( + isolated_tmux, tmp_path, monkeypatch, +): + # Arrange + from agent_shell.models.agent import AgentType + from agent_shell.shell import AgentShell + + binary = tmp_path / "codex" + binary.write_text("#!/usr/bin/env python3\nraise SystemExit(4)\n") + binary.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") + shell = AgentShell(AgentType.CODEX, execution_host=TmuxExecutionHost()) + + # Act + async with await shell.open_interactive(str(tmp_path)) as session: + async with asyncio.timeout(3): + events = [event async for event in session.events()] + + # Assert + assert [(e.type, e.returncode) for e in events] == [("process_exit", 4)] + + +async def test_owner_death_removes_its_terminal(isolated_tmux, tmp_path): + # Arrange: the owner is a real separate process that cannot run graceful Python cleanup. + code = ''' +import asyncio, sys +from agent_shell import TmuxExecutionHost +async def main(): + terminal = await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", "import time; time.sleep(60)"], sys.argv[1]) + print(terminal.session_name, flush=True) + await asyncio.sleep(60) +asyncio.run(main()) +''' + owner = await asyncio.create_subprocess_exec( + sys.executable, "-c", code, str(tmp_path), stdout=asyncio.subprocess.PIPE, + ) + session_name = (await asyncio.wait_for(owner.stdout.readline(), 5)).decode().strip() + assert session_name.startswith("agentshell-ui-") + + # Act + owner.kill() + await owner.wait() + try: + async with asyncio.timeout(3): + while True: + probe = await asyncio.create_subprocess_exec( + "tmux", "has-session", "-t", session_name, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, + ) + if await probe.wait() != 0: + break + await asyncio.sleep(0.05) + finally: + cleanup = await asyncio.create_subprocess_exec( + "tmux", "kill-session", "-t", session_name, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, + ) + await cleanup.wait() + + # Assert + assert probe.returncode != 0 + + +@pytest.mark.parametrize("interruption", ["death", "cancel", "bad-receipt"]) +async def test_interrupted_startup_removes_pane_with_remain_on_exit( + current_tmux_terminal, isolated_tmux, tmp_path, interruption, +): + # Arrange: delay only the external tmux launch receipt, after the pane exists. + original = current_tmux_terminal + await isolated_tmux("set-option", "-w", "-t", original.window_id, "remain-on-exit", "on") + tmux = shutil.which("tmux") + receipt = tmp_path / "receipt" + wrapper = tmp_path / "tmux" + wrapper.write_text( + f"#!{sys.executable}\nimport pathlib, subprocess, sys, time\n" + f"result = subprocess.run([{tmux!r}, *sys.argv[1:]], capture_output=True)\n" + "if 'split-window' in sys.argv and result.returncode == 0:\n" + f" pathlib.Path({str(receipt)!r}).write_bytes(result.stdout)\n" + " time.sleep(1)\n" + f" if {interruption!r} == 'bad-receipt': result.stdout = b'bad receipt'\n" + "sys.stdout.buffer.write(result.stdout)\n" + "sys.stderr.buffer.write(result.stderr)\n" + "raise SystemExit(result.returncode)\n" + ) + wrapper.chmod(0o755) + code = ''' +import asyncio, os, sys +from pathlib import Path +from agent_shell import TmuxExecutionHost, TmuxPlacement, TmuxUnavailableError +async def main(): + task = asyncio.create_task(TmuxExecutionHost(TmuxPlacement.split_pane()).launch_interactive( + [sys.executable, '-c', 'input()'], sys.argv[1])) + async with asyncio.timeout(5): + while not Path(sys.argv[2]).exists(): + await asyncio.sleep(0.01) + if sys.argv[3] == 'death': + os._exit(0) + if sys.argv[3] == 'cancel': + task.cancel() + try: + await task + except (asyncio.CancelledError, TmuxUnavailableError): + pass + else: + raise AssertionError('launch unexpectedly succeeded') +asyncio.run(main()) +''' + env = dict(os.environ, PATH=f"{tmp_path}:{os.environ['PATH']}") + + # Act + owner = await asyncio.create_subprocess_exec( + sys.executable, "-c", code, str(tmp_path), str(receipt), interruption, env=env, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + _, stderr = await asyncio.wait_for(owner.communicate(), 8) + assert owner.returncode == 0, stderr.decode() + + # Assert: inherited remain-on-exit must not leave a dead split behind. + async with asyncio.timeout(3): + while await isolated_tmux("list-panes", "-t", original.window_id, + "-F", "#{pane_id}") != original.pane_id: + await asyncio.sleep(0.02) + await original.send_text("still alive", submit=True) + await screen_contains(original, "ORIGINAL:still alive") + finally: + if owner.returncode is None: + owner.kill() + await owner.wait() + + +@pytest.mark.parametrize("shutdown", [ + "close", "owner-death", "pane-removal", "suspended-close", "suspended-owner-death", +]) +@pytest.mark.parametrize("leader_exits", [False, True]) +async def test_shutdown_removes_resistant_helper( + current_tmux_terminal, isolated_tmux, tmp_path, shutdown, leader_exits, +): + # Arrange: the helper ignores terminal hangup and polite termination, holding a real lock. + helper = tmp_path / "helper.py" + lock = tmp_path / "helper.lock" + release = tmp_path / "release-helper" + helper.write_text(''' +import fcntl, signal, sys, time +from pathlib import Path +signal.signal(signal.SIGTERM, signal.SIG_IGN) +signal.signal(signal.SIGHUP, signal.SIG_IGN) +with open(sys.argv[1], 'w') as locked: + fcntl.flock(locked, fcntl.LOCK_EX) + print('HELPER READY', flush=True) + deadline = time.monotonic() + 15 + while not Path(sys.argv[2]).exists() and time.monotonic() < deadline: + time.sleep(0.02) +''') + leader = tmp_path / "leader.py" + leader.write_text(''' +import subprocess, sys, time +subprocess.Popen([sys.executable, sys.argv[1], sys.argv[2], sys.argv[3]]) +if sys.argv[4] == 'False': + time.sleep(60) +''') + close = tmp_path / "close" + code = ''' +import asyncio, sys +from pathlib import Path +from agent_shell import TmuxExecutionHost, TmuxPlacement +async def main(): + terminal = await TmuxExecutionHost(TmuxPlacement.split_pane()).launch_interactive( + [sys.executable, *sys.argv[1:6]], str(Path(sys.argv[1]).parent)) + async with asyncio.timeout(5): + while 'HELPER READY' not in await terminal.capture_screen(): + await asyncio.sleep(0.01) + if sys.argv[5] == 'True': + await terminal.wait() + print(terminal.pane_id, flush=True) + while not Path(sys.argv[6]).exists(): + await asyncio.sleep(0.02) + await terminal.close() +asyncio.run(main()) +''' + owner = await asyncio.create_subprocess_exec( + sys.executable, "-c", code, str(leader), str(helper), str(lock), str(release), + str(leader_exits), str(close), stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + pane = (await asyncio.wait_for(owner.stdout.readline(), 5)).decode().strip() + assert pane.startswith("%") + with lock.open("a") as probe: + with pytest.raises(BlockingIOError): + fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) + + # Act: Ctrl-Z suspends the foreground group before cleanup. + if shutdown.startswith("suspended-"): + await isolated_tmux("send-keys", "-t", pane, "C-z") + shutdown = shutdown.removeprefix("suspended-") + if shutdown == "owner-death": + owner.kill() + elif shutdown == "pane-removal": + await isolated_tmux("kill-pane", "-t", pane) + else: + close.touch() + + # Assert: the helper releases its lock well before its own safety deadline. + async with asyncio.timeout(3): + while True: + try: + fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + await asyncio.sleep(0.02) + await current_tmux_terminal.send_text("still alive", submit=True) + await screen_contains(current_tmux_terminal, "ORIGINAL:still alive") + finally: + release.touch() + close.touch() + if owner.returncode is None and shutdown == "pane-removal": + owner.kill() + try: + await asyncio.wait_for(owner.communicate(), 5) + except TimeoutError: + owner.kill() + await owner.wait() + + +async def test_concurrent_close_is_idempotent(isolated_tmux, tmp_path): + # Arrange + terminal = await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", "import time; time.sleep(60)"], str(tmp_path), + ) + + # Act + await asyncio.gather(terminal.close(), terminal.close()) + + # Assert + assert terminal.closed + assert await terminal.wait() < 0 + + +@pytest.mark.parametrize("interactive", [True, False]) +@pytest.mark.parametrize("replacement_name", ["owned-work", "owned"]) +async def test_close_does_not_target_another_session_after_owned_session_disappears( + isolated_tmux, current_tmux_terminal, tmp_path, replacement_name, interactive, +): + # Arrange: retain a sentinel so the server cannot restart and reuse IDs. + host = TmuxExecutionHost(TmuxPlacement.new_session("owned")) + launch = host.launch_interactive if interactive else host.launch + terminal = await launch([sys.executable, "-c", "pass"], str(tmp_path)) + await asyncio.wait_for(terminal.wait(), 5) + pane = await isolated_tmux("list-panes", "-t", "=owned:", "-F", "#{pane_id}") + await isolated_tmux("kill-pane", "-t", pane) + async with await TmuxExecutionHost( + TmuxPlacement.new_session(replacement_name), + ).launch_interactive([sys.executable, "-c", "input()"], str(tmp_path)) as replacement: + # Act + if interactive: + await terminal.close() + else: + terminal.release() + + # Assert + assert await isolated_tmux("list-panes", "-t", replacement.pane_id, + "-F", "#{pane_id}") == replacement.pane_id + + +@pytest.mark.parametrize("interactive", [True, False]) +@pytest.mark.parametrize("kind", ["session", "window", "pane"]) +async def test_close_preserves_replacement_after_server_restart( + isolated_tmux, tmp_path, monkeypatch, interactive, kind, +): + # Arrange: restart only the fixture's dedicated server, deliberately reusing resource IDs. + async def prepare_server(): + if kind == "session": + return TmuxPlacement.new_session("owned") + identity = await isolated_tmux( + "new-session", "-d", "-s", "base", "-P", "-F", "#{pane_id} #{socket_path}", + "--", sys.executable, "-c", "import time; time.sleep(60)", + ) + pane, socket = identity.split() + monkeypatch.setenv("TMUX", f"{socket},0,0") + monkeypatch.setenv("TMUX_PANE", pane) + return TmuxPlacement.split_pane() if kind == "pane" else TmuxPlacement.current_session() + + host = TmuxExecutionHost(await prepare_server()) + launch = host.launch_interactive if interactive else host.launch + terminal = await launch([sys.executable, "-c", "pass"], str(tmp_path)) + await asyncio.wait_for(terminal.wait(), 5) + old_ids = await isolated_tmux("list-panes", "-a", "-F", "#{session_id} #{window_id} #{pane_id}") + await isolated_tmux("kill-server") + await prepare_server() + async with await host.launch_interactive( + [sys.executable, "-c", "print('READY', flush=True); input()"], str(tmp_path), + ) as replacement: + await screen_contains(replacement, "READY") + assert await isolated_tmux( + "list-panes", "-a", "-F", "#{session_id} #{window_id} #{pane_id}", + ) == old_ids + + # Act + if interactive: + await terminal.close() + else: + terminal.release() + + # Assert + assert await isolated_tmux( + "list-panes", "-a", "-F", "#{session_id} #{window_id} #{pane_id}", + ) == old_ids + + +async def test_manually_removed_pane_does_not_hang_waiter(isolated_tmux, tmp_path): + # Arrange + terminal = await TmuxExecutionHost().launch_interactive( + [sys.executable, "-c", "import time; time.sleep(60)"], str(tmp_path), + ) + + # Act: emulate a person closing the pane from tmux. + try: + process = await asyncio.create_subprocess_exec("tmux", "kill-pane", "-t", terminal.pane_id) + await process.wait() + async with asyncio.timeout(2): + with pytest.raises(RuntimeError, match="terminal disappeared"): + await terminal.wait() + finally: + await terminal.close() + + +async def test_immediate_process_exit_keeps_exact_status(isolated_tmux, tmp_path): + # Arrange: the shortest real executable exercises exit during the foreground handoff. + host = TmuxExecutionHost() + + # Act + async with await host.launch_interactive(["/bin/true"], str(tmp_path)) as terminal: + status = await asyncio.wait_for(terminal.wait(), 5) + + # Assert + assert status == 0 + + +async def test_tmux_kill_timeout_still_releases_owner(isolated_tmux, tmp_path, monkeypatch): + # Arrange: substitute only the external tmux executable's failing kill operation. + real_tmux = shutil.which("tmux") + wrapper = tmp_path / "tmux" + wrapper.write_text( + f"#!{sys.executable}\nimport os, sys, time\n" + "if any(arg.startswith(('kill-session', 'kill-window')) for arg in sys.argv):\n" + " time.sleep(60)\n" + f"os.execv({real_tmux!r}, [{real_tmux!r}, *sys.argv[1:]])\n" + ) + wrapper.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") + # A separate controller ensures failed assertions cannot leak a test-owned FIFO descriptor. + code = ''' +import asyncio, sys +from agent_shell import TmuxExecutionHost +async def main(): + terminal = await TmuxExecutionHost().launch_interactive( + [sys.executable, '-c', 'import time; time.sleep(60)'], sys.argv[1], + ) + await terminal.close() + assert terminal.closed + await terminal.close() + # Keep the controller alive: its FIFO release must remove the owned session now. + async with asyncio.timeout(3): + while True: + probe = await asyncio.create_subprocess_exec( + sys.argv[2], 'has-session', '-t', terminal.session_name, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, + ) + if await probe.wait() != 0: + break + await asyncio.sleep(0.05) + print('CLOSED TWICE', flush=True) +asyncio.run(main()) +''' + + # Act + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", code, str(tmp_path), real_tmux, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for(process.communicate(), 12) + + # Assert + assert process.returncode == 0, stderr.decode() + assert b"CLOSED TWICE" in stdout diff --git a/tests/integration/test_tmux_execution_host.py b/tests/integration/test_tmux_execution_host.py index 94b60bd..4f5984b 100644 --- a/tests/integration/test_tmux_execution_host.py +++ b/tests/integration/test_tmux_execution_host.py @@ -26,6 +26,7 @@ def fake_tmux(monkeypatch, tmp_path): """Replace the external tmux boundary while keeping the real bridge process and IPC.""" registry = tmp_path / "tmux-registry" registry.mkdir() + (tmp_path / "tmux-commands").mkdir() bin_dir = tmp_path / "tmux-bin" bin_dir.mkdir() fake = bin_dir / "tmux" @@ -46,45 +47,45 @@ def fake_tmux(monkeypatch, tmp_path): " stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,\n" " start_new_session=True)\n" " (registry / session).write_text(str(worker.pid))\n" + " (registry.parent / 'tmux-commands' / str(worker.pid)).write_text(' '.join(command))\n" + " print(f'${worker.pid}\\t@{worker.pid}\\t%{worker.pid}\\t/fake/socket')\n" " raise SystemExit(0)\n" "if 'new-window' in args:\n" - " session = args[args.index('-t') + 1]\n" + " session = args[args.index('-t') + 1].removeprefix('=').removesuffix(':')\n" " if not (registry / session).exists():\n" " raise SystemExit(1)\n" " command = args[args.index('--') + 1:]\n" - " window = session + '-window-' + str(os.getpid()) + '-' + str(len(list(registry.iterdir())))\n" + " window = session + '-window-' + str(os.getpid()) + '-' + " + "str(len(list(registry.iterdir())))\n" " worker = subprocess.Popen(command, stdin=subprocess.DEVNULL,\n" " stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,\n" " start_new_session=True)\n" " (registry / window).write_text(str(worker.pid))\n" + " (registry.parent / 'tmux-commands' / str(worker.pid)).write_text(' '.join(command))\n" " argv_file = os.environ.get('AGENTSHELL_FAKE_TMUX_ARGV_FILE')\n" " if argv_file:\n" " pathlib.Path(argv_file).write_text('\\0'.join(args))\n" " if os.environ.get('AGENTSHELL_FAKE_TMUX_EMPTY_WINDOW_ID'):\n" " print('', flush=True)\n" " else:\n" - " print(window, flush=True)\n" + " print(f'${worker.pid}\\t@{worker.pid}\\t%{worker.pid}\\t/fake/socket')\n" " raise SystemExit(0)\n" "if 'display-message' in args:\n" " print(os.environ.get('AGENTSHELL_FAKE_TMUX_CURRENT_SESSION', ''))\n" " raise SystemExit(0)\n" - "if 'kill-session' in args:\n" - " session = args[args.index('-t') + 1]\n" - " marker = registry / session\n" - " try:\n" - " os.kill(int(marker.read_text()), signal.SIGKILL)\n" - " except (FileNotFoundError, ProcessLookupError, ValueError):\n" - " pass\n" - " marker.unlink(missing_ok=True)\n" - " raise SystemExit(0)\n" - "if 'kill-window' in args:\n" - " window = args[args.index('-t') + 1]\n" - " marker = registry / window\n" - " try:\n" - " os.kill(int(marker.read_text()), signal.SIGKILL)\n" - " except (FileNotFoundError, ProcessLookupError, ValueError):\n" - " pass\n" - " marker.unlink(missing_ok=True)\n" + "if 'if-shell' in args:\n" + " pid = args[args.index('-t') + 1].removeprefix('%')\n" + " condition = args[args.index('-t') + 2]\n" + " token = condition.split('*')[1]\n" + " command = registry.parent / 'tmux-commands' / pid\n" + " if command.exists() and token in command.read_text():\n" + " for marker in registry.iterdir():\n" + " if marker.read_text() == pid:\n" + " try:\n" + " os.kill(int(pid), signal.SIGKILL)\n" + " except ProcessLookupError:\n" + " pass\n" + " marker.unlink()\n" " raise SystemExit(0)\n" "if 'list-panes' in args:\n" " for marker in sorted(registry.iterdir()):\n" @@ -216,6 +217,29 @@ async def test_new_window_placement_borrows_session_and_owns_only_new_window( borrowed_session.unlink() +async def test_new_window_placement_exactly_targets_numeric_session( + fake_tmux, monkeypatch, tmp_path +): + # Arrange + borrowed_session = fake_tmux / "6" + borrowed_session.write_text(str(os.getpid())) + argv_file = tmp_path / "tmux-argv" + monkeypatch.setenv("AGENTSHELL_FAKE_TMUX_ARGV_FILE", str(argv_file)) + host = TmuxExecutionHost(placement=TmuxPlacement.new_window(session="6")) + + # Act + try: + run = await host.launch([sys.executable, "-c", "pass"], cwd=str(tmp_path)) + await run.communicate() + tmux_args = argv_file.read_text().split("\0") + run.release() + finally: + borrowed_session.unlink(missing_ok=True) + + # Assert + assert tmux_args[tmux_args.index("-t") + 1] == "=6:" + + async def test_current_session_placement_fails_clearly_outside_tmux( monkeypatch, tmp_path ): @@ -325,7 +349,7 @@ async def test_unidentifiable_new_window_preserves_borrowed_session( # Act / Assert try: - with pytest.raises(TmuxUnavailableError, match="window id"): + with pytest.raises(TmuxUnavailableError, match="valid session, window, pane"): await host.launch([sys.executable, "-c", "pass"], cwd=str(tmp_path)) assert borrowed_session.exists() finally: diff --git a/tests/unit/test_process_cleanup.py b/tests/unit/test_process_cleanup.py index efa200c..d409674 100644 --- a/tests/unit/test_process_cleanup.py +++ b/tests/unit/test_process_cleanup.py @@ -70,7 +70,7 @@ def test_missing_guardian_never_falls_back_to_a_numeric_group_signal(self): process = _fake_process(pid=12345) # Act - with patch("agent_shell.process_cleanup.os.killpg") as killpg: + with patch("os.killpg") as killpg: kill_process_group(process) # Assert