fix(acp): prune idle pool slots and stabilize history prefix hash

Replace idle runtimes in the pool under lock so acquirers always get the canonical instance. Hash conversation prefixes via ChatMessage.to_dict() with a versioned envelope so metadata-only drift does not reset the agent. Document session default pooling, HistoryState.prefix_hash, and prompt/history branches. Add tests for metadata-stable hashing, parallel acquire after idle reap, and pool recycle after kill_all.

Made-with: Cursor

Author: Mateusz

src/connectors/acp_core/base_connector.py

@@ -58,6 +58,36 @@
 ACP_AGENT_THOUGHT_CHUNK = "agent_thought_chunk"
 ACP_CANCEL_METHODS = ("session/cancel", "session/stop", "session/end")
 ACP_GRACEFUL_CANCEL_TIMEOUT_SECONDS = 12.0
+# Increment when the canonicalization used for ACP history prefix hashes changes.
+HISTORY_PREFIX_HASH_VERSION = 2
+
+
+def _canonical_chat_message_for_history_hash(message: ChatMessage) -> dict[str, Any]:
+    """Return stable identity fields for divergence detection.
+
+    Uses :meth:`ChatMessage.to_dict` so fields such as ``metadata`` that are not
+    part of the visible transcript do not spuriously invalidate the prefix hash.
+    """
+
+    return message.to_dict()
+
+
+def _hash_chat_messages_prefix_stable(
+    messages: Sequence[ChatMessage],
+    end_exclusive: int,
+) -> str:
+    """SHA-256 hex digest of the first ``end_exclusive`` messages (conversation prefix)."""
+
+    if end_exclusive <= 0:
+        return hashlib.sha256(b"").hexdigest()
+    slice_msgs = messages[:end_exclusive]
+    payload = [_canonical_chat_message_for_history_hash(m) for m in slice_msgs]
+    canonical = json.dumps(
+        {"m": payload, "v": HISTORY_PREFIX_HASH_VERSION},
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
 
 
 class _RuntimeCancellable:
@@ -175,6 +205,15 @@ def _create_runtime(
 
     @staticmethod
     def _resolve_client_session_id(request: ConnectorChatCompletionsRequest) -> str:
+        """Resolve the logical client session used to key ACP subprocess pools.
+
+        When neither ``ConnectorRequestContext.session_id`` nor
+        ``request.request.session_id`` is set, callers share the pool key
+        ``"default"`` (one ACP runtime per ``(project_dir, model)`` for all
+        such traffic). Upstream layers should set a stable session id when
+        isolation between clients or tabs is required.
+        """
+
         sid: str | None = None
         if request.context is not None and request.context.session_id:
             sid = request.context.session_id
@@ -190,12 +229,7 @@ def _resolve_client_session_id(request: ConnectorChatCompletionsRequest) -> str:
     def _hash_messages_prefix(
         messages: Sequence[ChatMessage], end_exclusive: int
     ) -> str:
-        if end_exclusive <= 0:
-            return hashlib.sha256(b"").hexdigest()
-        slice_msgs = messages[:end_exclusive]
-        payload = [m.model_dump(mode="json", exclude_none=True) for m in slice_msgs]
-        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
-        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+        return _hash_chat_messages_prefix_stable(messages, end_exclusive)
 
     async def _acquire_runtime(
         self, request: ConnectorChatCompletionsRequest
@@ -218,8 +252,7 @@ async def _acquire_runtime(
                 )
                 self._runtimes[runtime_key] = runtime
 
-        await self._reap_idle_runtime(runtime)
-        return runtime
+        return await self._reap_idle_runtime(runtime_key, runtime)
 
     def _resolve_project_dir_override(
         self, request: ConnectorChatCompletionsRequest
@@ -259,19 +292,46 @@ def _resolve_project_dir_for_request(
 
         return self._default_project_dir
 
-    async def _reap_idle_runtime(self, runtime: ACPProcessRuntime) -> None:
+    async def _reap_idle_runtime(
+        self,
+        runtime_key: tuple[str, str, str],
+        runtime: ACPProcessRuntime,
+    ) -> ACPProcessRuntime:
+        """Drop idle subprocesses and swap in a fresh :class:`ACPProcessRuntime` slot.
+
+        Replacing the pool entry (instead of only clearing ``runtime.process``)
+        avoids unbounded growth of dead :class:`ACPProcessRuntime` objects while
+        ensuring concurrent acquirers always resolve to the canonical instance
+        currently registered for ``runtime_key``.
+        """
+
         if self._idle_timeout <= 0:
-            return
+            return runtime
         if runtime.request_lock is None or runtime.request_lock.locked():
-            return
+            return runtime
         if runtime.process is None:
-            return
+            return runtime
         if runtime.last_activity <= 0:
-            return
+            return runtime
         if (time.monotonic() - runtime.last_activity) < self._idle_timeout:
-            return
+            return runtime
+
         await self._kill_runtime(runtime)
 
+        async with self._runtime_pool_lock:
+            current = self._runtimes.get(runtime_key)
+            if current is runtime:
+                replacement = self._create_runtime(
+                    runtime.project_dir,
+                    runtime.model,
+                    runtime.client_session_id,
+                )
+                self._runtimes[runtime_key] = replacement
+                return replacement
+            if current is not None:
+                return current
+        return runtime
+
     async def _spawn_process(self, runtime: ACPProcessRuntime) -> None:
         assert runtime.process_lock is not None
         async with runtime.process_lock:
@@ -800,6 +860,13 @@ async def _prepare_prompt_request_locked(
         runtime: ACPProcessRuntime,
         request: ConnectorChatCompletionsRequest,
     ) -> tuple[int, str]:
+        """Build ``session/prompt`` text and JSON-RPC id under ``runtime.request_lock``.
+
+        History is tracked with :class:`HistoryState` so we can send a compact
+        tail transcript on append-only turns, resend the full transcript after
+        detected divergence, or send only the last user line on idempotent retries.
+        """
+
         await self._spawn_process(runtime)
         await self._initialize_runtime(runtime)
 
@@ -811,6 +878,7 @@ async def _prepare_prompt_request_locked(
         new_history_state: HistoryState
         user_message: str
 
+        # First prompt for this subprocess: full Markdown transcript + state seed.
         if state is None:
             user_message = ACPTranscriptSerializer.serialize(messages)
             new_history_state = HistoryState(
@@ -825,6 +893,7 @@ async def _prepare_prompt_request_locked(
                 or self._hash_messages_prefix(messages, n) != prefix_hash
             )
 
+            # Prefix edit, branch switch, or truncated history vs. what ACP saw.
             if diverged:
                 if logger.isEnabledFor(logging.INFO):
                     logger.info(
@@ -842,9 +911,11 @@ async def _prepare_prompt_request_locked(
                     message_count=len(messages),
                     prefix_hash=self._hash_messages_prefix(messages, len(messages)),
                 )
+            # Same message list as last successful prompt (e.g. client retry).
             elif len(messages) == n:
                 user_message = self._extract_user_message_as_string(messages)
                 new_history_state = state
+            # Append-only: agent already saw messages[:n]; ship incremental context.
             else:
                 user_message = ACPTranscriptSerializer.serialize_tail(messages, n)
                 if not user_message.strip():
@@ -1035,6 +1106,8 @@ async def chat_completions(  # type: ignore[override]
         if runtime.request_lock is None:
             raise BackendError(message="ACP runtime is missing request lock")
 
+        # Streaming holds the per-runtime lock for the full SSE response so idle reap
+        # cannot swap the pool entry until the stream completes (see ``_reap_idle_runtime``).
         if bool(getattr(request.request, "stream", False)):
             await runtime.request_lock.acquire()
             try:

src/connectors/acp_core/types.py

@@ -84,7 +84,11 @@ class AcpStreamPiece:
 
 @dataclass(frozen=True, slots=True)
 class HistoryState:
-    """Tracks how much of ``processed_messages`` has been applied to the ACP agent."""
+    """Tracks how much of ``processed_messages`` has been applied to the ACP agent.
+
+    ``prefix_hash`` is an opaque SHA-256 hex digest of the acknowledged prefix; it is
+    computed only inside the ACP base connector and must not be interpreted elsewhere.
+    """
 
     message_count: int
     prefix_hash: str

tests/unit/connectors/acp_core/test_base_connector.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from typing import Any
@@ -323,6 +324,61 @@ def test_hash_messages_prefix_detects_edit(connector: DummyAcpConnector) -> None
     assert h1 != h2
 
 
+def test_hash_messages_prefix_stable_ignores_metadata_only_changes(
+    connector: DummyAcpConnector,
+) -> None:
+    a = [ChatMessage(role="user", content="hi", metadata={"a": 1})]
+    b = [ChatMessage(role="user", content="hi", metadata={"b": 2})]
+    assert connector._hash_messages_prefix(a, 1) == connector._hash_messages_prefix(
+        b, 1
+    )
+
+
+@pytest.mark.asyncio
+async def test_parallel_acquire_after_idle_reap_uses_same_pool_runtime(
+    connector: DummyAcpConnector,
+) -> None:
+    connector._default_project_dir = Path("/tmp/dummy")
+    connector._idle_timeout = 5.0
+    rid = "sess-par"
+    key = connector._build_runtime_key(Path("/tmp/dummy"), "model", rid)
+    stale = connector._create_runtime(Path("/tmp/dummy"), "model", rid)
+    stale.process = MagicMock()
+    stale.process.poll.return_value = None
+    stale.last_activity = 1.0
+    async with connector._runtime_pool_lock:
+        connector._runtimes[key] = stale
+
+    with (
+        patch(
+            "src.connectors.acp_core.base_connector.time.monotonic",
+            return_value=100.0,
+        ),
+        patch.object(connector, "_terminate_process", AsyncMock()),
+    ):
+        r1, r2 = await asyncio.gather(
+            connector._acquire_runtime(_make_request(session_id=rid)),
+            connector._acquire_runtime(_make_request(session_id=rid)),
+        )
+
+    assert r1 is r2
+    assert r1 is not stale
+    assert connector._runtimes[key] is r1
+
+
+@pytest.mark.asyncio
+async def test_kill_all_runtimes_next_acquire_creates_new_runtime_object(
+    connector: DummyAcpConnector,
+) -> None:
+    connector._default_project_dir = Path("/tmp/dummy")
+    r_before = await connector._acquire_runtime(_make_request(session_id="recycle"))
+    await connector._kill_all_runtimes()
+    assert len(connector._runtimes) == 0
+    r_after = await connector._acquire_runtime(_make_request(session_id="recycle"))
+    assert r_after is not r_before
+    assert r_after.history_state is None
+
+
 @pytest.mark.asyncio
 async def test_non_streaming_chat_completions_preserve_reasoning_content(
     connector: DummyAcpConnector,