refactor(craft): simplify ACP session management — single resume_or_create_session entry point

Remove redundant two-layer session management that was left over from the
long-lived client pattern. With ephemeral clients, the in-memory session
cache and public wrapper methods were dead code.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rohoswagger

backend/onyx/server/features/build/sandbox/kubernetes/internal/acp_exec_client.py

@@ -4,17 +4,18 @@
 using stdin/stdout for JSON-RPC communication. This bypasses the HTTP server
 and uses the native ACP subprocess protocol.
 
-When multiple API server replicas share the same sandbox pod, this client
-uses ACP session resumption (session/list + session/resume) to maintain
-conversation context across replicas.
+Each message creates an ephemeral client (start → resume_or_create_session →
+send_message → stop) to prevent concurrent processes from corrupting
+opencode's flat file session storage.
 
 Usage:
     client = ACPExecClient(
         pod_name="sandbox-abc123",
         namespace="onyx-sandboxes",
     )
     client.start(cwd="/workspace")
-    for event in client.send_message("What files are here?"):
+    session_id = client.resume_or_create_session(cwd="/workspace/sessions/abc")
+    for event in client.send_message("What files are here?", session_id=session_id):
         print(event)
     client.stop()
 """
@@ -160,7 +161,7 @@ def start(self, cwd: str = "/workspace", timeout: float = 30.0) -> None:
         """Start the agent process via exec and initialize the ACP connection.
 
         Only performs the ACP `initialize` handshake. Sessions are created
-        separately via `create_session()` or `resume_session()`.
+        separately via `resume_or_create_session()`.
 
         Args:
             cwd: Working directory for the `opencode acp` process
@@ -502,42 +503,11 @@ def _try_resume_existing_session(self, cwd: str, timeout: float) -> str | None:
             )
             return None
 
-    def create_session(self, cwd: str, timeout: float = 30.0) -> str:
-        """Create a new ACP session on this connection.
+    def resume_or_create_session(self, cwd: str, timeout: float = 30.0) -> str:
+        """Resume a session from opencode's on-disk storage, or create a new one.
 
-        Args:
-            cwd: Working directory for the session
-            timeout: Timeout for the request
-
-        Returns:
-            The ACP session ID
-        """
-        if not self._state.initialized:
-            raise RuntimeError("Client not initialized. Call start() first.")
-        return self._create_session(cwd=cwd, timeout=timeout)
-
-    def resume_session(self, session_id: str, cwd: str, timeout: float = 30.0) -> str:
-        """Resume an existing ACP session on this connection.
-
-        Args:
-            session_id: The ACP session ID to resume
-            cwd: Working directory for the session
-            timeout: Timeout for the request
-
-        Returns:
-            The ACP session ID
-        """
-        if not self._state.initialized:
-            raise RuntimeError("Client not initialized. Call start() first.")
-        return self._resume_session(session_id=session_id, cwd=cwd, timeout=timeout)
-
-    def get_or_create_session(self, cwd: str, timeout: float = 30.0) -> str:
-        """Get an existing session for this cwd, or create/resume one.
-
-        Tries in order:
-        1. Return an already-tracked session for this cwd
-        2. Resume an existing session from opencode's storage (multi-replica)
-        3. Create a new session
+        With ephemeral clients (one process per message), this always hits disk.
+        Tries resume first to preserve conversation context, falls back to new.
 
         Args:
             cwd: Working directory for the session
@@ -549,14 +519,6 @@ def get_or_create_session(self, cwd: str, timeout: float = 30.0) -> str:
         if not self._state.initialized:
             raise RuntimeError("Client not initialized. Call start() first.")
 
-        # Check if we already have a session for this cwd
-        for sid, session in self._state.sessions.items():
-            if session.cwd == cwd:
-                logger.info(
-                    f"[ACP] Reusing existing session: " f"acp_session={sid} cwd={cwd}"
-                )
-                return sid
-
         # Try to resume from opencode's persisted storage
         resumed_id = self._try_resume_existing_session(cwd, timeout)
         if resumed_id:
@@ -844,11 +806,6 @@ def is_running(self) -> bool:
         """Check if the exec session is running."""
         return self._ws_client is not None and self._ws_client.is_open()
 
-    @property
-    def session_ids(self) -> list[str]:
-        """Get all tracked session IDs."""
-        return list(self._state.sessions.keys())
-
     def __enter__(self) -> "ACPExecClient":
         """Context manager entry."""
         return self

backend/onyx/server/features/build/sandbox/kubernetes/kubernetes_sandbox_manager.py

@@ -353,12 +353,6 @@ def _initialize(self) -> None:
         self._service_account = SANDBOX_SERVICE_ACCOUNT_NAME
         self._file_sync_service_account = SANDBOX_FILE_SYNC_SERVICE_ACCOUNT
 
-        # Maps (sandbox_id, craft_session_id) → ACP session ID.
-        # Cached across messages so we can resume the same opencode session.
-        # The actual opencode session is persisted on disk in the sandbox pod,
-        # so this cache is just an optimization to skip session/list + resume.
-        self._acp_session_ids: dict[tuple[UUID, UUID], str] = {}
-
         # Load AGENTS.md template path
         build_dir = Path(__file__).parent.parent.parent  # /onyx/server/features/build/
         self._agent_instructions_template_path = build_dir / "AGENTS.template.md"
@@ -1174,11 +1168,6 @@ def terminate(self, sandbox_id: UUID) -> None:
         Args:
             sandbox_id: The sandbox ID to terminate
         """
-        # Remove all session mappings for this sandbox
-        keys_to_remove = [key for key in self._acp_session_ids if key[0] == sandbox_id]
-        for key in keys_to_remove:
-            del self._acp_session_ids[key]
-
         # Clean up Kubernetes resources (needs string for pod/service names)
         self._cleanup_kubernetes_resources(str(sandbox_id))
 
@@ -1422,15 +1411,6 @@ def cleanup_session_workspace(
             nextjs_port: Optional port where Next.js server is running (unused in K8s,
                         we use PID file instead)
         """
-        # Remove the ACP session mapping (shared client persists)
-        session_key = (sandbox_id, session_id)
-        acp_session_id = self._acp_session_ids.pop(session_key, None)
-        if acp_session_id:
-            logger.info(
-                f"[SANDBOX-ACP] Removed ACP session mapping: "
-                f"session={session_id} acp_session={acp_session_id}"
-            )
-
         pod_name = self._get_pod_name(str(sandbox_id))
         session_path = f"/workspace/sessions/{session_id}"
 
@@ -1865,43 +1845,6 @@ def _create_ephemeral_acp_client(self, sandbox_id: UUID) -> ACPExecClient:
         )
         return acp_client
 
-    def _get_or_create_acp_session(
-        self,
-        sandbox_id: UUID,
-        session_id: UUID,
-        acp_client: ACPExecClient,
-    ) -> str:
-        """Get the ACP session ID for a craft session, creating one if needed.
-
-        Uses the session mapping cache first, then falls back to
-        `get_or_create_session()` which handles resume from opencode's
-        persisted storage on disk.
-
-        Args:
-            sandbox_id: The sandbox ID
-            session_id: The craft session ID
-            acp_client: The ACP client for this message
-
-        Returns:
-            The ACP session ID
-        """
-        session_key = (sandbox_id, session_id)
-        acp_session_id = self._acp_session_ids.get(session_key)
-
-        if acp_session_id and acp_session_id in acp_client.session_ids:
-            return acp_session_id
-
-        # Session not tracked or new process — get or create from disk
-        session_path = f"/workspace/sessions/{session_id}"
-        acp_session_id = acp_client.get_or_create_session(cwd=session_path)
-        self._acp_session_ids[session_key] = acp_session_id
-
-        logger.info(
-            f"[SANDBOX-ACP] Session mapped: "
-            f"craft_session={session_id} acp_session={acp_session_id}"
-        )
-        return acp_session_id
-
     def send_message(
         self,
         sandbox_id: UUID,
@@ -1931,10 +1874,9 @@ def send_message(
         acp_client = self._create_ephemeral_acp_client(sandbox_id)
 
         try:
-            # Get or create the ACP session for this craft session
-            acp_session_id = self._get_or_create_acp_session(
-                sandbox_id, session_id, acp_client
-            )
+            # Resume (or create) the ACP session from opencode's on-disk storage
+            session_path = f"/workspace/sessions/{session_id}"
+            acp_session_id = acp_client.resume_or_create_session(cwd=session_path)
 
             logger.info(
                 f"[SANDBOX-ACP] Sending message: "