fix(acp): isolate runtimes by client session and sync history

- Key ACPProcessRuntime pool by (project_dir, model, client_session_id)
- Replace history_injected with HistoryState (message_count + prefix hash)
- On contiguous new messages, send serialize_tail; on divergence, kill and resync
- Add ACPTranscriptSerializer.serialize_tail for mid-session context
- Extend unit tests for isolation, incremental prompts, and divergence

Made-with: Cursor

Author: Mateusz

src/connectors/acp_core/base_connector.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import contextlib
+import hashlib
 import json
 import logging
 import os
@@ -20,6 +21,7 @@
     ACPSessionUpdate,
     AcpStreamPiece,
     ACPUpdateContent,
+    HistoryState,
 )
 from src.connectors.base import LLMBackend, add_vendor_prefix, strip_vendor_prefix
 from src.connectors.contracts import ConnectorChatCompletionsRequest
@@ -107,7 +109,7 @@ def __init__(
         self._process_timeout = DEFAULT_PROCESS_TIMEOUT
         self._idle_timeout = DEFAULT_IDLE_TIMEOUT
         self._runtime_pool_lock = asyncio.Lock()
-        self._runtimes: dict[tuple[str, str], ACPProcessRuntime] = {}
+        self._runtimes: dict[tuple[str, str, str], ACPProcessRuntime] = {}
 
     @property
     def has_static_credentials(self) -> bool:
@@ -153,19 +155,48 @@ async def _handle_server_request(
     def _is_usable_directory(path: Path) -> bool:
         return path.exists() and path.is_dir() and os.access(path, os.R_OK)
 
-    def _build_runtime_key(self, project_dir: Path, model: str) -> tuple[str, str]:
-        return (str(project_dir), model)
+    def _build_runtime_key(
+        self, project_dir: Path, model: str, client_session_id: str
+    ) -> tuple[str, str, str]:
+        return (str(project_dir), model, client_session_id)
 
-    def _create_runtime(self, project_dir: Path, model: str) -> ACPProcessRuntime:
+    def _create_runtime(
+        self, project_dir: Path, model: str, client_session_id: str = "default"
+    ) -> ACPProcessRuntime:
         return ACPProcessRuntime(
             project_dir=project_dir,
             model=model,
+            client_session_id=client_session_id,
             process_lock=asyncio.Lock(),
             request_lock=asyncio.Lock(),
             cancellation_lock=asyncio.Lock(),
             cancellation_event=asyncio.Event(),
         )
 
+    @staticmethod
+    def _resolve_client_session_id(request: ConnectorChatCompletionsRequest) -> str:
+        sid: str | None = None
+        if request.context is not None and request.context.session_id:
+            sid = request.context.session_id
+        if not sid:
+            raw = getattr(request.request, "session_id", None)
+            if isinstance(raw, str) and raw.strip():
+                sid = raw
+        if isinstance(sid, str) and sid.strip():
+            return sid.strip()
+        return "default"
+
+    @staticmethod
+    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()
+
     async def _acquire_runtime(
         self, request: ConnectorChatCompletionsRequest
     ) -> ACPProcessRuntime:
@@ -174,12 +205,17 @@ async def _acquire_runtime(
             request.effective_model or self._model,
             self.VENDOR_PREFIX,
         )
-        runtime_key = self._build_runtime_key(project_dir, requested_model)
+        client_session_id = self._resolve_client_session_id(request)
+        runtime_key = self._build_runtime_key(
+            project_dir, requested_model, client_session_id
+        )
 
         async with self._runtime_pool_lock:
             runtime = self._runtimes.get(runtime_key)
             if runtime is None:
-                runtime = self._create_runtime(project_dir, requested_model)
+                runtime = self._create_runtime(
+                    project_dir, requested_model, client_session_id
+                )
                 self._runtimes[runtime_key] = runtime
 
         await self._reap_idle_runtime(runtime)
@@ -276,14 +312,14 @@ async def _spawn_process(self, runtime: ACPProcessRuntime) -> None:
                 runtime.initialized = False
                 runtime.session_id = None
                 runtime.message_id = 0
-                runtime.history_injected = False
+                runtime.history_state = None
             except Exception as exc:
                 if new_process is not None:
                     self._cleanup_process(new_process)
                 runtime.process = None
                 runtime.initialized = False
                 runtime.session_id = None
-                runtime.history_injected = False
+                runtime.history_state = None
                 raise APIConnectionError(
                     message=f"Failed to start ACP process: {exc}",
                     details={
@@ -323,7 +359,7 @@ def _cleanup_runtime_state(
         runtime.session_id = None
         runtime.message_id = 0
         runtime.last_activity = 0.0
-        runtime.history_injected = False
+        runtime.history_state = None
 
     def _cleanup_process(self, process: subprocess.Popen[bytes] | None = None) -> None:
         if process is None:
@@ -767,14 +803,57 @@ async def _prepare_prompt_request_locked(
         await self._spawn_process(runtime)
         await self._initialize_runtime(runtime)
 
-        if not runtime.history_injected:
-            user_message = ACPTranscriptSerializer.serialize(request.processed_messages)
-            runtime.history_injected = True
+        messages = list(request.processed_messages)
+        if not messages:
+            raise BackendError(message="No messages found in request")
+
+        state = runtime.history_state
+        new_history_state: HistoryState
+        user_message: str
+
+        if state is None:
+            user_message = ACPTranscriptSerializer.serialize(messages)
+            new_history_state = HistoryState(
+                message_count=len(messages),
+                prefix_hash=self._hash_messages_prefix(messages, len(messages)),
+            )
         else:
-            user_message = self._extract_user_message_as_string(
-                request.processed_messages
+            n = state.message_count
+            prefix_hash = state.prefix_hash
+            diverged = (
+                len(messages) < n
+                or self._hash_messages_prefix(messages, n) != prefix_hash
             )
 
+            if diverged:
+                if logger.isEnabledFor(logging.INFO):
+                    logger.info(
+                        "ACP history diverged or shrank; resetting agent process "
+                        "(project=%s model=%s client_session=%s)",
+                        runtime.project_dir,
+                        runtime.model,
+                        runtime.client_session_id,
+                    )
+                await self._kill_runtime(runtime)
+                await self._spawn_process(runtime)
+                await self._initialize_runtime(runtime)
+                user_message = ACPTranscriptSerializer.serialize(messages)
+                new_history_state = HistoryState(
+                    message_count=len(messages),
+                    prefix_hash=self._hash_messages_prefix(messages, len(messages)),
+                )
+            elif len(messages) == n:
+                user_message = self._extract_user_message_as_string(messages)
+                new_history_state = state
+            else:
+                user_message = ACPTranscriptSerializer.serialize_tail(messages, n)
+                if not user_message.strip():
+                    user_message = self._extract_user_message_as_string(messages)
+                new_history_state = HistoryState(
+                    message_count=len(messages),
+                    prefix_hash=self._hash_messages_prefix(messages, len(messages)),
+                )
+
         if not user_message:
             raise BackendError(message="No user message found in request")
 
@@ -792,6 +871,7 @@ async def _prepare_prompt_request_locked(
             "session/prompt",
             prompt_params,
         )
+        runtime.history_state = new_history_state
         return prompt_request_id, requested_model
 
     async def _stream_response_with_lock(

src/connectors/acp_core/transcript.py

@@ -81,6 +81,73 @@ def serialize(messages: Sequence[ChatMessage | dict[str, Any] | str | Any]) -> s
 
         return "\n".join(lines)
 
+    @staticmethod
+    def serialize_tail(
+        messages: Sequence[ChatMessage | dict[str, Any] | str | Any],
+        start_index: int,
+    ) -> str:
+        """Serialize messages from ``start_index`` through the final user turn.
+
+        Used when the ACP agent already saw messages ``[:start_index]`` and new
+        messages were appended (e.g. non-ACP turns in between).
+        """
+        if start_index <= 0:
+            return ACPTranscriptSerializer.serialize(messages)
+        if not messages or start_index >= len(messages):
+            return ""
+
+        last_user_idx = -1
+        for i in range(len(messages) - 1, -1, -1):
+            if ACPTranscriptSerializer._get_role(messages[i]) == "user":
+                last_user_idx = i
+                break
+
+        if last_user_idx == -1:
+            return ""
+
+        last_user_msg = ACPTranscriptSerializer._get_content(messages[last_user_idx])
+        if last_user_idx < start_index:
+            return last_user_msg
+
+        history_msgs = list(messages[start_index:last_user_idx])
+        if not history_msgs:
+            return last_user_msg
+
+        lines = [
+            "[System Note: Additional conversation occurred since your last response. "
+            "Here is the new context:]",
+            "",
+            "--- New Messages ---",
+        ]
+
+        for msg in history_msgs:
+            role = ACPTranscriptSerializer._get_role(msg)
+            content = ACPTranscriptSerializer._get_content(msg)
+
+            if role == "system":
+                lines.append(f"**System:** {content}")
+            elif role == "user":
+                lines.append(f"**User:** {content}")
+            elif role == "assistant":
+                lines.append(f"**Assistant:** {content}")
+
+                tool_calls = ACPTranscriptSerializer._get_tool_calls(msg)
+                for tc in tool_calls:
+                    func_name = tc.get("function", {}).get("name", "unknown")
+                    func_args = tc.get("function", {}).get("arguments", "{}")
+                    lines.append(f"*Tool Call (`{func_name}`)*: `{func_args}`")
+            elif role == "tool":
+                lines.append(f"*Tool Result*: `{content}`")
+            else:
+                lines.append(f"**{role.capitalize()}:** {content}")
+
+        lines.append("------------------------")
+        lines.append("")
+        lines.append("[Current Request]")
+        lines.append(last_user_msg)
+
+        return "\n".join(lines)
+
     @staticmethod
     def _get_role(msg: Any) -> str:
         if isinstance(msg, ChatMessage):

src/connectors/acp_core/types.py

@@ -82,18 +82,27 @@ class AcpStreamPiece:
     reasoning_content: str | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class HistoryState:
+    """Tracks how much of ``processed_messages`` has been applied to the ACP agent."""
+
+    message_count: int
+    prefix_hash: str
+
+
 @dataclass(slots=True)
 class ACPProcessRuntime:
-    """Live ACP runtime bound to a project directory and model."""
+    """Live ACP runtime bound to a project directory, model, and client session."""
 
     project_dir: Path
     model: str
+    client_session_id: str = "default"
     process: Any | None = None
     session_id: str | None = None
     initialized: bool = False
     message_id: int = 0
     last_activity: float = 0.0
-    history_injected: bool = False
+    history_state: HistoryState | None = None
     process_lock: Any = field(default=None)
     request_lock: Any = field(default=None)
     cancellation_lock: Any = field(default=None)