Implement OpenCode server daemon integration with automatic fallback to CLI

opencode_llm.py

@@ -1,7 +1,8 @@
 """A Pipecat processor that puts OpenCode (with ollama-cloud/gemma4:31b) in the LLM slot.
 
-Drives the installed OpenCode CLI (`opencode run -m ollama-cloud/gemma4:31b`) to stream
+Supports both:
-cloud responses to text-to-speech downstream with zero local GPU overhead.
+1. OpenCode Server Daemon (`opencode serve --port 4096`) for zero-latency, persistent in-memory sessions.
+2. OpenCode CLI (`opencode run --continue`) for direct process invocation.
 """
 
 import asyncio
@@ -10,6 +11,7 @@ import os
 import re
 import shutil
 from pathlib import Path
+import aiohttp
 from loguru import logger
 
 from pipecat.frames.frames import (
@@ -37,6 +39,8 @@ _NOISE_TRANSCRIPTS = {
     "[silence]",
 }
 
+_server_proc: asyncio.subprocess.Process | None = None
+
 
 def _clean_spoken_text(text: str) -> str:
     """Clean text for speech output and truncate fake turn generations."""
@@ -65,33 +69,70 @@ def find_opencode_cli() -> str | None:
     )
 
 
+async def ensure_opencode_server(port: int = 4096) -> tuple[bool, str]:
+    """Ensure opencode serve daemon is running on port."""
+    global _server_proc
+    url = f"http://localhost:{port}/session"
+
+    try:
+        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=2.0)) as session:
+            async with session.get(url) as resp:
+                if resp.status == 200:
+                    return True, f"OpenCode server active on http://localhost:{port}"
+    except Exception:
+        pass
+
+    cli = find_opencode_cli()
+    if not cli:
+        return False, "OpenCode CLI binary not found"
+
+    logger.info(f"Starting OpenCode server daemon on port {port}...")
+    try:
+        _server_proc = await asyncio.create_subprocess_exec(
+            cli,
+            "serve",
+            "--port", str(port),
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        await asyncio.sleep(1.2)
+        return True, f"Started OpenCode server daemon on http://localhost:{port}"
+    except Exception as e:
+        return False, f"Failed to start OpenCode server daemon: {e}"
+
+
 def probe_opencode(model: str = "ollama-cloud/gemma4:31b") -> tuple[bool, str]:
     cli = find_opencode_cli()
     if not cli:
         return False, "OpenCode CLI binary not found"
-    return True, f"OpenCode CLI available ({cli}) with model {model}"
+    return True, f"OpenCode available ({cli}) with model {model}"
 
 
 class OpenCodeLLM(FrameProcessor):
-    """Runs user turns through the OpenCode CLI driving OpenCode Cloud models."""
+    """Runs user turns through OpenCode Server Daemon or CLI."""
 
     def __init__(
         self,
         *,
         model: str = "ollama-cloud/gemma4:31b",
         cwd: str | Path | None = None,
+        port: int = 4096,
         system_prompt: str | None = None,
         observer=None,
+        use_server: bool = True,
         **kwargs,
     ):
         super().__init__(**kwargs)
         self._model = model
         self._cwd = Path(cwd or Path.home() / "Workspace").expanduser().resolve()
+        self._port = port
         self._system_prompt = system_prompt or "You are a helpful spoken voice assistant. Keep answers brief and conversational."
         self._on_reply = observer
         self._turn_task: asyncio.Task | None = None
         self._history: list[dict[str, str]] = []
         self._cli_path = find_opencode_cli() or "opencode"
+        self._use_server = use_server
+        self._server_session_id: str | None = None
         self._has_session = False
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -99,7 +140,11 @@ class OpenCodeLLM(FrameProcessor):
 
         if isinstance(frame, StartFrame):
             await self.push_frame(frame, direction)
-            logger.info(f"OpenCode LLM engine ready: CLI={self._cli_path}, model={self._model}")
+            if self._use_server:
+                ok, reason = await ensure_opencode_server(self._port)
+                logger.info(f"OpenCode Server engine: {reason}")
+            else:
+                logger.info(f"OpenCode CLI engine ready: CLI={self._cli_path}, model={self._model}")
         elif isinstance(frame, (EndFrame, CancelFrame)):
             await self._cancel_turn()
             await self.push_frame(frame, direction)
@@ -144,7 +189,6 @@ class OpenCodeLLM(FrameProcessor):
     async def _run_turn(self, utterance: str):
         self._history.append({"role": "user", "content": utterance})
 
-        # Format prompt with system instructions and recent conversation history
         recent_history = self._history[-6:]
         conv_text = "\n".join(
             f"{'User' if m['role']=='user' else 'Assistant'}: {m['content']}"
@@ -155,7 +199,77 @@ class OpenCodeLLM(FrameProcessor):
         await self.push_frame(LLMFullResponseStartFrame())
         chunks: list[str] = []
 
-        # Use --continue for subsequent turns to persist tool session state
+        if self._use_server:
+            await self._run_turn_server(prompt_str, chunks)
+        else:
+            await self._run_turn_cli(prompt_str, chunks)
+
+        full_reply = _clean_spoken_text(" ".join(chunks))
+        if full_reply:
+            self._history.append({"role": "assistant", "content": full_reply})
+            logger.info(f"OpenCode LLM ({self._model}): {full_reply}")
+            if self._on_reply:
+                self._on_reply(full_reply)
+
+    async def _run_turn_server(self, prompt_str: str, chunks: list[str]):
+        """Run turn via OpenCode Server Daemon HTTP API."""
+        try:
+            ok, _ = await ensure_opencode_server(self._port)
+            if not ok:
+                raise RuntimeError("OpenCode server daemon unavailable")
+
+            async with aiohttp.ClientSession() as session:
+                if not self._server_session_id:
+                    create_url = f"http://localhost:{self._port}/session"
+                    async with session.post(create_url, json={"directory": str(self._cwd)}) as res:
+                        if res.status == 200:
+                            data = await res.json()
+                            self._server_session_id = data.get("id")
+                            logger.info(f"OpenCode server session created: {self._server_session_id}")
+
+                if not self._server_session_id:
+                    raise RuntimeError("Failed to create OpenCode server session")
+
+                msg_url = f"http://localhost:{self._port}/session/{self._server_session_id}/message"
+                model_id = self._model.split("/")[-1] if "/" in self._model else self._model
+                provider_id = self._model.split("/")[0] if "/" in self._model else "ollama-cloud"
+
+                payload = {
+                    "model": {"providerID": provider_id, "modelID": model_id},
+                    "parts": [{"type": "text", "text": prompt_str}],
+                }
+
+                async with session.post(msg_url, json=payload) as resp:
+                    async for line in resp.content:
+                        line_str = line.decode("utf-8").strip()
+                        if not line_str:
+                            continue
+                        try:
+                            data = json.loads(line_str)
+                            if "delta" in data:
+                                text_chunk = _clean_spoken_text(data["delta"])
+                                if text_chunk:
+                                    chunks.append(text_chunk)
+                                    await self.push_frame(LLMTextFrame(text_chunk))
+                            elif "text" in data:
+                                text_chunk = _clean_spoken_text(data["text"])
+                                if text_chunk:
+                                    chunks.append(text_chunk)
+                                    await self.push_frame(LLMTextFrame(text_chunk))
+                        except json.JSONDecodeError:
+                            pass
+
+        except asyncio.CancelledError:
+            logger.info("OpenCode server turn cancelled mid-response.")
+            raise
+        except Exception as e:
+            logger.warning(f"OpenCode server error ({e}), falling back to CLI...")
+            await self._run_turn_cli(prompt_str, chunks)
+        finally:
+            await self.push_frame(LLMFullResponseEndFrame())
+
+    async def _run_turn_cli(self, prompt_str: str, chunks: list[str]):
+        """Fallback turn via OpenCode CLI."""
         cmd = [
             self._cli_path,
             "run",
@@ -167,6 +281,7 @@ class OpenCodeLLM(FrameProcessor):
             cmd.append("--continue")
         cmd.append(prompt_str)
 
+        proc = None
         try:
             proc = await asyncio.create_subprocess_exec(
                 *cmd,
@@ -210,16 +325,7 @@ class OpenCodeLLM(FrameProcessor):
                     pass
             raise
         except Exception as e:
-            logger.error(f"OpenCode LLM error: {e}")
+            logger.error(f"OpenCode LLM CLI error: {e}")
             err_msg = "Sorry, I ran into an error generating a response."
             chunks.append(err_msg)
             await self.push_frame(LLMTextFrame(err_msg))
-        finally:
-            await self.push_frame(LLMFullResponseEndFrame())
-
-        full_reply = _clean_spoken_text(" ".join(chunks))
-        if full_reply:
-            self._history.append({"role": "assistant", "content": full_reply})
-            logger.info(f"OpenCode LLM ({self._model}): {full_reply}")
-            if self._on_reply:
-                self._on_reply(full_reply)