diff --git a/examples/loopx-chat-acp-adapter-smoke.py b/examples/loopx-chat-acp-adapter-smoke.py index 9997b90dae..a9163f08b1 100644 --- a/examples/loopx-chat-acp-adapter-smoke.py +++ b/examples/loopx-chat-acp-adapter-smoke.py @@ -29,6 +29,7 @@ session_id = "acp:fixture/session" active_prompt_id = None session_cwd = "" +client_initialized = False for line in sys.stdin: request = json.loads(line) assert request.get("jsonrpc") == "2.0", request @@ -44,7 +45,12 @@ "sessionCapabilities": {"close": {}}, }, } + elif method == "initialized": + assert request_id is None + client_initialized = True + continue elif method == "session/new": + assert client_initialized, "client must send initialized before session/new" assert request["params"]["mcpServers"] == [] session_cwd = request["params"]["cwd"] result = {"sessionId": session_id} diff --git a/loopx/chat_acp.py b/loopx/chat_acp.py index 0d99583465..0a42a97da6 100644 --- a/loopx/chat_acp.py +++ b/loopx/chat_acp.py @@ -29,6 +29,8 @@ EventSink = Callable[[str, dict[str, Any]], None] ACP_PROTOCOL_VERSION = 1 +ACP_INITIALIZE_TIMEOUT_DEFAULT_SEC = 30.0 +ACP_SESSION_STARTUP_TIMEOUT_DEFAULT_SEC = 45.0 def _reader(stream: Any, messages: "queue.Queue[dict[str, Any] | Exception]") -> None: @@ -84,7 +86,7 @@ def start( work_dir: Path, agent_work_dir: Path | None = None, resume_thread_id: str | None = None, - startup_timeout_sec: float = 30.0, + startup_timeout_sec: float | None = None, idle_timeout_sec: float = 180.0, hard_timeout_sec: float = 900.0, execution_mode: bool = False, @@ -117,6 +119,11 @@ def start( messages: "queue.Queue[dict[str, Any] | Exception]" = queue.Queue() assert process.stdout is not None threading.Thread(target=_reader, args=(process.stdout, messages), daemon=True).start() + initialize_timeout_sec = ( + ACP_INITIALIZE_TIMEOUT_DEFAULT_SEC + if startup_timeout_sec is None + else startup_timeout_sec + ) adapter = cls( process=process, messages=messages, @@ -125,7 +132,7 @@ def start( agent_work_dir=agent_work_dir or work_dir.expanduser().resolve(), agent_capabilities={}, execution_mode=execution_mode, - startup_timeout_sec=startup_timeout_sec, + startup_timeout_sec=initialize_timeout_sec, idle_timeout_sec=idle_timeout_sec, hard_timeout_sec=hard_timeout_sec, ) @@ -145,12 +152,20 @@ def start( }, }, request_id=1, - timeout_sec=startup_timeout_sec, + timeout_sec=initialize_timeout_sec, ) if int(initialized.get("protocolVersion") or 0) != ACP_PROTOCOL_VERSION: raise adapter._runtime_error("ACP Agent selected an unsupported protocol version.") capabilities = initialized.get("agentCapabilities") adapter.agent_capabilities = capabilities if isinstance(capabilities, dict) else {} + # ACP agents may defer session creation until the client confirms + # initialization before accepting ``session/new``. + adapter._notify("initialized", {}) + session_startup_timeout_sec = ( + ACP_SESSION_STARTUP_TIMEOUT_DEFAULT_SEC + if startup_timeout_sec is None + else startup_timeout_sec + ) session_params = {"cwd": str(adapter.agent_work_dir), "mcpServers": []} if resume_thread_id: session_capabilities = adapter.agent_capabilities.get("sessionCapabilities") @@ -160,14 +175,14 @@ def start( "session/resume", {**session_params, "sessionId": resume_thread_id}, request_id=2, - timeout_sec=startup_timeout_sec, + timeout_sec=session_startup_timeout_sec, ) elif bool(adapter.agent_capabilities.get("loadSession")): result = adapter._request( "session/load", {**session_params, "sessionId": resume_thread_id}, request_id=2, - timeout_sec=startup_timeout_sec, + timeout_sec=session_startup_timeout_sec, ) else: raise adapter._runtime_error("ACP Agent does not support session recovery.") @@ -180,7 +195,7 @@ def start( "session/new", session_params, request_id=2, - timeout_sec=startup_timeout_sec, + timeout_sec=session_startup_timeout_sec, ) adapter.session_id = str(result.get("sessionId") or "") if not adapter.session_id: diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 157d1e089d..966d3cb38c 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -315,7 +315,7 @@ def __init__( codex_bin: str, claude_bin: str = "claude", kiro_cli_bin: str = KIRO_CLI_BIN, - startup_timeout_sec: float = 30.0, + startup_timeout_sec: float | None = None, idle_timeout_sec: float = 180.0, hard_timeout_sec: float = 900.0, endpoint_registry: AgentEndpointRegistry | None = None, @@ -336,9 +336,9 @@ def __init__( or os.environ.get("CODEX_HOME") or "~/.codex" ).expanduser().resolve() - self.claude_bin = claude_bin - self.kiro_cli_bin = kiro_cli_bin - self.startup_timeout_sec = startup_timeout_sec + self.claude_bin, self.kiro_cli_bin = claude_bin, kiro_cli_bin + self.acp_startup_timeout_sec = startup_timeout_sec + self.startup_timeout_sec = 30.0 if startup_timeout_sec is None else startup_timeout_sec self.idle_timeout_sec = idle_timeout_sec self.hard_timeout_sec = hard_timeout_sec self.endpoint_registry = endpoint_registry or AgentEndpointRegistry(store.root) @@ -561,7 +561,7 @@ def _start_adapter( command=kiro_cli_chat_command(self.kiro_cli_bin), work_dir=work_dir, resume_thread_id=resume_thread_id, - startup_timeout_sec=self.startup_timeout_sec, + startup_timeout_sec=self.acp_startup_timeout_sec, idle_timeout_sec=self.idle_timeout_sec, hard_timeout_sec=self.hard_timeout_sec, execution_mode=execution_mode, @@ -573,7 +573,7 @@ def _start_adapter( work_dir=work_dir, agent_work_dir=endpoint.mapped_work_dir(work_dir), resume_thread_id=resume_thread_id, - startup_timeout_sec=self.startup_timeout_sec, + startup_timeout_sec=self.acp_startup_timeout_sec, idle_timeout_sec=self.idle_timeout_sec, hard_timeout_sec=self.hard_timeout_sec, execution_mode=execution_mode, diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 9bfeb40d8d..88b48ea138 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -1566,7 +1566,7 @@ def serve_chat( claude_bin: str = "claude", kiro_cli_bin: str = KIRO_CLI_BIN, lark_cli_bin: str | None = None, - startup_timeout_sec: float = 30.0, + startup_timeout_sec: float | None = None, idle_timeout_sec: float = 180.0, hard_timeout_sec: float = 900.0, assets_dir: Path | None = None, diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index b99e6a5e6f..df8ea7a8c6 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -642,7 +642,11 @@ def handle_support_control_command( claude_bin=args.claude_bin, kiro_cli_bin=getattr(args, "kiro_cli_bin", KIRO_CLI_BIN), lark_cli_bin=args.lark_cli_bin, - startup_timeout_sec=max(0.1, float(args.startup_timeout_seconds)), + startup_timeout_sec=( + max(0.1, float(args.startup_timeout_seconds)) + if args.startup_timeout_seconds is not None + else None + ), idle_timeout_sec=max(0.1, float(args.idle_timeout_seconds)), hard_timeout_sec=max(0.1, float(args.hard_timeout_seconds)), assets_dir=Path(args.assets_dir).expanduser() diff --git a/loopx/cli_commands/support_control_chat.py b/loopx/cli_commands/support_control_chat.py index 45dd4d962c..bd8f5cff15 100644 --- a/loopx/cli_commands/support_control_chat.py +++ b/loopx/cli_commands/support_control_chat.py @@ -36,8 +36,10 @@ def register_chat_and_dashboard_commands( chat_parser.add_argument( "--startup-timeout-seconds", type=float, - default=30.0, - help="Maximum seconds allowed for Codex app-server startup and handshake.", + help=( + "Maximum seconds allowed for Agent startup and session creation. " + "Defaults to 30 seconds for Codex startup and 45 seconds for ACP session creation." + ), ) chat_parser.add_argument( "--idle-timeout-seconds", diff --git a/tests/test_chat_acp_startup_timeout.py b/tests/test_chat_acp_startup_timeout.py new file mode 100644 index 0000000000..be2b2771ee --- /dev/null +++ b/tests/test_chat_acp_startup_timeout.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import os +from pathlib import Path +import sys + +import pytest + +from loopx.chat_acp import ACPStdioAdapter +from loopx.chat_endpoints import AgentEndpointRegistry +from loopx.chat_runtime import ChatRuntimeController +from loopx.chat_store import ChatSessionStore + + +SLOW_SESSION_ACP = r''' +import json +import os +from pathlib import Path +import sys +import time + +delay = float(sys.argv[1]) +Path(sys.argv[2]).write_text(str(os.getpid()), encoding="utf-8") +session_method = sys.argv[3] +for line in sys.stdin: + request = json.loads(line) + method = request.get("method") + request_id = request.get("id") + if method == "initialize": + capabilities = ( + {"loadSession": True} + if session_method == "session/load" + else {"sessionCapabilities": {"resume": {}}} + if session_method == "session/resume" + else {} + ) + result = {"protocolVersion": 1, "agentCapabilities": capabilities} + elif method == "initialized": + continue + elif method == session_method: + time.sleep(delay) + result = {"sessionId": "acp:fixture/slow-session"} + elif method == "session/close": + result = {} + else: + continue + print(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result}), flush=True) +''' + + +def _command( + tmp_path: Path, + *, + name: str, + session_method: str = "session/new", +) -> tuple[tuple[str, ...], Path]: + agent = tmp_path / "slow-session-acp.py" + agent.write_text(SLOW_SESSION_ACP, encoding="utf-8") + pid_path = tmp_path / f"{name}.pid" + return ( + sys.executable, + str(agent), + "0.15", + str(pid_path), + session_method, + ), pid_path + + +def test_acp_default_and_larger_session_startup_windows_accept_slow_peer( + tmp_path: Path, +) -> None: + default_command, _ = _command(tmp_path, name="default") + default_adapter = ACPStdioAdapter.start(command=default_command, work_dir=tmp_path) + assert default_adapter.upstream_thread_id == "acp:fixture/slow-session" + default_adapter.close_session() + + longer_command, _ = _command(tmp_path, name="longer") + longer_adapter = ACPStdioAdapter.start( + command=longer_command, + work_dir=tmp_path, + startup_timeout_sec=0.5, + ) + assert longer_adapter.upstream_thread_id == "acp:fixture/slow-session" + longer_adapter.close_session() + + +@pytest.mark.parametrize( + ("session_method", "resume_thread_id"), + [ + ("session/new", None), + ("session/load", "acp:fixture/slow-session"), + ("session/resume", "acp:fixture/slow-session"), + ], +) +def test_acp_explicit_short_startup_limit_rejects_and_reaps_peer( + tmp_path: Path, + session_method: str, + resume_thread_id: str | None, +) -> None: + command, pid_path = _command( + tmp_path, + name=session_method.replace("/", "-"), + session_method=session_method, + ) + with pytest.raises(TimeoutError, match=session_method): + ACPStdioAdapter.start( + command=command, + work_dir=tmp_path, + resume_thread_id=resume_thread_id, + startup_timeout_sec=0.05, + ) + + process_id = int(pid_path.read_text(encoding="utf-8")) + with pytest.raises(ProcessLookupError): + os.kill(process_id, 0) + + +def test_chat_runtime_uses_default_acp_session_startup_window(tmp_path: Path) -> None: + store = ChatSessionStore(tmp_path / "runtime") + registry = AgentEndpointRegistry(store.root) + command, _ = _command(tmp_path, name="controller-default") + registry.upsert( + { + "agent_id": "slow-default-acp", + "display_name": "Slow default ACP", + "command": list(command), + "location": "local", + } + ) + controller = ChatRuntimeController( + store=store, + codex_bin="missing-codex-for-fixture", + endpoint_registry=registry, + ) + + session, was_resumed = controller.open_session( + goal_id="fixture-default-timeout", + agent_id="slow-default-acp", + work_dir=tmp_path, + objective="Exercise the default ACP session startup window.", + mode="new", + ) + assert was_resumed is False + controller.close_session(str(session["session_id"])) + + +def test_chat_runtime_preserves_explicit_acp_startup_limit(tmp_path: Path) -> None: + store = ChatSessionStore(tmp_path / "runtime") + registry = AgentEndpointRegistry(store.root) + command, _ = _command(tmp_path, name="controller-short") + registry.upsert( + { + "agent_id": "slow-short-acp", + "display_name": "Slow short ACP", + "command": list(command), + "location": "local", + } + ) + controller = ChatRuntimeController( + store=store, + codex_bin="missing-codex-for-fixture", + endpoint_registry=registry, + startup_timeout_sec=0.05, + ) + + with pytest.raises(TimeoutError, match="session/new"): + controller.open_session( + goal_id="fixture-short-timeout", + agent_id="slow-short-acp", + work_dir=tmp_path, + objective="Exercise an explicit ACP session startup limit.", + mode="new", + ) + assert not store.list_sessions(agent_id="slow-short-acp")