fix(chat): complete ACP initialization before session creation

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}

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:

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,

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,

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()

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",

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")