Launch native agent UIs through shared tmux hosts

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: <agent> <prompt> | /key <agent> 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

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:
         ...
 

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,

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,

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,