Giving Your Open-Source Agent a Resident Coding CLI Brain — Integrating CodeBuddy into Hermes Agent over ACP

A field-tested integration guide — architecture, full source code, and eleven pitfalls we hit so you don’t have to.

• Components: Hermes Agent (open-source agent runtime) · CodeBuddy Code CLI 2.180.0 (Tencent’s Claude-Code-style coding CLI) · systemd user services
• Scope: Replace an agent’s direct model-API calls with a local, resident CodeBuddy CLI reached over ACP (Agent Client Protocol) — without touching the agent’s core loop.
• Status: Running in production on our side; every command and code block below is verbatim from the working deployment.

---

1. Why this integration

Hermes Agent already ships a complete agent loop: conversation history, tool execution, permissions, memory, retries. CodeBuddy CLI ships a full coding-agent runtime: its own model access, streaming, session management, and an ACP server.

The naive options are both bad:

• Call the model API directly — you pay for (or configure) yet another upstream provider and duplicate what CodeBuddy already provides.
• Treat CodeBuddy as a second agent — two autonomous agents executing tools on the same task means duplicated side effects, double spend, and diverging history.

What we wanted: CodeBuddy becomes a “model brain” only. Hermes keeps 100% ownership of tools, permissions, and history. CodeBuddy is reduced to a resident inference endpoint that speaks JSON-RPC over HTTP.

Hermes AIAgent / conversation loop
        │  client.chat.completions.create(...)     ← OpenAI shape; core loop untouched
        ▼
CodeBuddyDaemonClient                       ← user-level provider plugin (in-process)
        │  ACP over HTTP (JSON-RPC + SSE, POST /api/v1/acp)
        ▼
CodeBuddy Serve (resident, systemd user unit, --tools "")
        │
        ▼
CodeBuddy's configured model backend

Result: a full multi-turn, tool-using Hermes session where every inference round-trips through a local CodeBuddy. Streaming time-to-first-token ~1–2 s on warm sessions; one-command rollback to the previous provider.

---

2. Five design decisions (and why)

1. codebuddy --serve, not codebuddy daemon start.
   In 2.180.0, daemon start accepts only --host/--port — it has no --tools flag. --serve runs the same HTTP surface (REST + ACP over HTTP SSE) on a fixed port with password auth, and honors --tools "" (disable all built-in tools). One harmless leak: the model may issue a ToolSearch probe, find nothing, and move on — semantically tools stay disabled.

2. User-level provider plugin, zero repo patches.
   Hermes loads plugins from $HERMES_HOME/plugins/model-providers/<name>/ and later registrations override bundled profiles. Everything lives outside the agent’s source tree — upgrades never force a rebase.

3. A fresh ACP connection + session per completion.
   Hermes is the single source of truth for conversation history. A CodeBuddy session represents exactly one inference call; the resident server is the only long-lived thing. This eliminates history duplication, retry forks, and cross-session bleed.

4. --tools "" — CodeBuddy executes nothing.
   Hermes’ tool schemas travel into the prompt as text (Hermes’ ACP bridge already does this), and the model’s <tool_call>{...}</tool_call> output is parsed back into OpenAI tool_calls. Two agents executing tools is the single biggest semantic hazard; we removed it structurally, plus the client auto-denies any interruption_request (tool-permission prompt) with a feedback message.

5. One turn at a time per server.
   Measured: CodeBuddy serve serializes agent work per account and silently drops the loser’s final result under concurrency. A module-level lock per base_url fixes it; sequential agent turns pay nothing.

---

3. Prerequisites

• A Linux box with ≥ 4 GB RAM strongly recommended (our 2 GB VM OOM-killed things until we trimmed services — see §9).
• codebuddy CLI installed and signed in (verify: codebuddy -p "Reply with exactly: OK" prints OK).
• Node ≥ 18; git; systemd user services with Linger=yes for the running user.
• Hermes Agent installed at $HERMES_HOME/hermes-agent, running under a non-root user (call it svc below).
• Tag the Hermes repo before you start (we change nothing in it, but leave a trail):
  git -C $HERMES_HOME/hermes-agent tag poc-baseline-$(date +%Y%m%d)

---

4. Step 1 — Make the CLI usable for the service user

The trap that cost us the most time: CodeBuddy’s sign-in state is not in ~/.codebuddy. Find it with strace:

strace -f -e trace=openat -o /tmp/st.log timeout 90 \
  env CODEBUDDY_CONFIG_DIR=/tmp/probe codebuddy -p "Reply with exactly: OK"
grep openat /tmp/st.log | grep -v ENOENT | grep -oE "/root/[^\" ]+" \
  | grep -vE 'probe|bun|node_modules|/proc|/sys|\.bash' | sort -u
# → ~/.local/share/CodeBuddyExtension/Data/Public/auth/Tencent-Cloud.coding-copilot.info

Copy the three pieces to the service user (here: svc, home /home/svc):

# 1) the actual credential (XDG data dir!)
mkdir -p /home/svc/.local/share/CodeBuddyExtension/Data/Public/auth
cp -a ~/.local/share/CodeBuddyExtension/Data/Public/auth/Tencent-Cloud.coding-copilot.info \
      /home/svc/.local/share/CodeBuddyExtension/Data/Public/auth/
chown -R svc:svc /home/svc/.local/share/CodeBuddyExtension

# 2) config-dir state (recommended)
cp -a ~/.codebuddy/local_storage /home/svc/.codebuddy/local_storage
cp -a ~/.codebuddy/user-state.json /home/svc/.codebuddy/
chown -R svc:svc /home/svc/.codebuddy

# 3) prove it
cd /home/svc && sudo -u svc env HOME=/home/svc codebuddy -p "Reply with exactly: OK"
# → OK

⚠️ Copying only settings.json/user-state.json/local_storage is not enough — the credential lives in the XDG data dir. This single fact cost us a full debugging loop.

---

5. Step 2 — The resident service (systemd user unit)

/home/svc/.config/systemd/user/codebuddy-acp.service:

[Unit]
Description=CodeBuddy CLI resident ACP server (tools disabled; Hermes owns tools)
After=network.target

[Service]
Type=simple
WorkingDirectory=%h
Environment=HOME=%h
ExecStart=/usr/bin/codebuddy --serve --host 127.0.0.1 --port 18090 --tools ""
Restart=on-failure
RestartSec=5

[Install]
WantedBy=default.target

loginctl show-user svc | grep Linger      # want: Linger=yes (else: loginctl enable-linger svc)
systemctl --user daemon-reload
systemctl --user enable --now codebuddy-acp
systemctl --user status codebuddy-acp     # Main PID: "CodeBuddy Serve", listening 127.0.0.1:18090

First start auto-generates the Web password into ~/.codebuddy/settings.json under gateway.password (the log prints where auth comes from). Feed it to the plugin via Hermes’ env file:

grep -q CODEBUDDY_DAEMON_PASSWORD $HERMES_HOME/.env || \
  echo "CODEBUDDY_DAEMON_PASSWORD=$(python3 -c "import json;print(json.load(open('/home/svc/.codebuddy/settings.json'))['gateway']['password'])")" \
  >> $HERMES_HOME/.env

ℹ️ Pure --acp mode (without --serve) is password-free but binds a random port — --port is ignored there — which makes the fixed systemd URL impossible. That’s why we chose --serve.

---

6. Step 3 — The provider plugin (full source)

Directory layout (all under $HERMES_HOME/plugins/model-providers/codebuddy-daemon/):

plugin.yaml
__init__.py
daemon_client.py

6.1 plugin.yaml

name: codebuddy-daemon
kind: model-provider
version: 0.1.0
description: Resident CodeBuddy CLI over local ACP/HTTP (tools disabled, Hermes owns the tool loop)
author: your-name

6.2 __init__.py

"""CodeBuddy daemon provider profile: the resident CodeBuddy CLI reached over local HTTP ACP."""

from __future__ import annotations

from typing import Any

from providers import register_provider
from providers.base import ProviderProfile


class CodeBuddyDaemonProfile(ProviderProfile):
    def create_client(self, **client_kwargs: Any) -> Any:
        from .daemon_client import CodeBuddyDaemonClient

        allowed = {"api_key", "base_url", "timeout"}
        return CodeBuddyDaemonClient(**{k: v for k, v in client_kwargs.items() if k in allowed})

    def get_max_tokens(self, model: str | None) -> int | None:
        return None  # output caps are the backend's; a Hermes default would truncate long answers


register_provider(CodeBuddyDaemonProfile(
    name="codebuddy-daemon",
    aliases=("codebuddy", "cb-daemon"),
    display_name="CodeBuddy Daemon (local ACP)",
    description="Resident CodeBuddy CLI over local ACP/HTTP — tools disabled, Hermes owns the tool loop",
    auth_type="api_key",
    env_vars=("CODEBUDDY_DAEMON_PASSWORD",),
    base_url="http://127.0.0.1:18090",
    supports_health_check=False,
    supports_model_listing=False,
    supports_vision=False,
    supports_streaming=True,
))

6.3 daemon_client.py

The whole file (445 lines) verbatim. Nine non-negotiable rules are encoded here — see §6.4 for the reasoning:

• reuse Hermes’ own ACP text bridge and tool-call parser (agent.acp_client, agent.acp_openai_bridge) — never hand-roll prompt rendering;
• lifecycle per completion: login → connect → initialize → session/new → (set_model / set thinking) → prompt → DELETE connection; never stop the server;
• every request carries X-CodeBuddy-Request: 1 (fixed value) and Accept: application/json, text/event-stream;
• the POST’s response body is itself an SSE stream — parse data: lines; the JSON-RPC result and all session/update notifications arrive inline there;
• map agent_message_chunk → delta.content, agent_thought_chunk → delta.reasoning_content, auto-deny interruption_request;
• tool-free requests stream live; tool-bearing requests buffer the full text first (never stream half a tool-call JSON);
• usage is always the zero-value shape (see §9, pitfall 10);
• one transparent auth retry only before the first delta; after that, errors propagate;
• a per-base_url lock wraps the whole turn.

"""CodeBuddy resident-CLI ACP client: an OpenAI ``chat.completions`` facade over CodeBuddy's
local HTTP ACP surface (``/api/v1/acp``).

Lifecycle: the CodeBuddy server process is owned by systemd, NOT by this client.
Every completion opens its own ACP connection + session, consumes the POST's inline SSE stream,
then disconnects. Hermes remains the only conversation-history owner; each session/prompt receives
the FULL history rendered as one text prompt by the shared ACP text bridge.

Tools: the server is launched with ``--tools ""`` (all CodeBuddy built-in tools disabled), so the
model can only answer through the Hermes text bridge; any tool-call JSON the model emits is parsed
by ``extract_tool_calls_from_text`` exactly like the bundled ACPClient path.

Streaming: tool-free requests stream ``delta`` chunks live from the SSE wire; tool-bearing
requests buffer the full text first — tool-call JSON must never stream half-formed.
"""

from __future__ import annotations

import json
import os
import threading
import time
import urllib.error
import urllib.request
from typing import Any, Iterator
from types import SimpleNamespace

# Reuse the bundled ACP path's proven helpers: compose module-level helpers, do not
# inherit ACPClient's subprocess lifecycle.
from agent.acp_client import _effective_timeout, _format_messages_as_prompt
from agent.acp_openai_bridge import (
    StreamChunks,
    completion_to_stream_chunks,
    extract_tool_calls_from_text,
)

_DEFAULT_BASE_URL = "http://127.0.0.1:18090"
_LOGIN_TIMEOUT = 30.0
_CONNECT_TIMEOUT = 30.0

# One turn at a time per resident server: CodeBuddy serve serializes agent work per account and
# drops the loser's final result under concurrency (measured), and serializing also keeps the
# shared account away from 429s. Hermes turns are sequential, so this costs nothing in practice.
_SERVE_LOCKS: dict[str, threading.Lock] = {}
_SERVE_LOCKS_GUARD = threading.Lock()


def _serve_lock(base_url: str) -> threading.Lock:
    with _SERVE_LOCKS_GUARD:
        return _SERVE_LOCKS.setdefault(base_url, threading.Lock())


class CodeBuddyDaemonError(RuntimeError):
    """Raised for auth/connect/protocol failures; message is safe to show to the user."""


class CodeBuddyDaemonClient:
    """Implements the ``client.chat.completions.create(...)`` surface Hermes calls."""

    def __init__(
        self,
        *,
        api_key: str | None = None,
        base_url: Any = None,
        timeout: Any = None,
        session_cwd: str | None = None,
        thinking: str | None = None,
        **_: Any,
    ) -> None:
        self.base_url = str(base_url or os.environ.get("CODEBUDDY_DAEMON_URL") or _DEFAULT_BASE_URL).rstrip("/")
        self._password = api_key or ""
        self._default_timeout = _effective_timeout(timeout) or 600.0
        self._session_cwd = session_cwd or os.environ.get("CODEBUDDY_SESSION_CWD") or os.environ.get("HOME") or "/tmp"
        self._thinking = thinking or os.environ.get("CODEBUDDY_THINKING") or None  # e.g. "disabled"
        self._token: str | None = None
        self._token_lock = threading.Lock()
        self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))

    # ------------------------------------------------------------------ HTTP

    def _headers(self, *, token: bool = True) -> dict[str, str]:
        headers = {
            "X-CodeBuddy-Request": "1",  # fixed value per CodeBuddy CSRF design
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
        }
        if token and self._token:
            headers["Authorization"] = f"Bearer {self._token}"
        return headers

    def _request(self, method: str, path: str, *, body: dict | None = None, timeout: float = 30.0,
                 headers: dict[str, str] | None = None) -> tuple[int, bytes]:
        url = f"{self.base_url}{path}"
        data = json.dumps(body).encode("utf-8") if body is not None else None
        req = urllib.request.Request(url, data=data, method=method, headers=headers or self._headers())
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                return resp.status, resp.read()
        except urllib.error.HTTPError as exc:
            return exc.code, exc.read() if hasattr(exc, "read") else b""
        except urllib.error.URLError as exc:
            raise CodeBuddyDaemonError(
                f"CodeBuddy daemon not reachable at {self.base_url} ({exc.reason}). "
                "Is the codebuddy-acp service running?"
            ) from exc

    # ------------------------------------------------------------------ auth

    def _login(self) -> None:
        """Password login → cached Bearer token."""
        with self._token_lock:
            if self._token:
                return
            if not self._password:
                raise CodeBuddyDaemonError(
                    "CodeBuddy daemon password missing. Set it as this provider's api_key "
                    "(or CODEBUDDY_DAEMON_PASSWORD)."
                )
            status, raw = self._request(
                "POST", "/api/v1/auth/login", body={"password": self._password}, timeout=_LOGIN_TIMEOUT)
            if status != 200:
                raise CodeBuddyDaemonError(f"CodeBuddy daemon login failed (HTTP {status}). Check the daemon password.")
            token = (json.loads(raw.decode("utf-8")) or {}).get("token")
            if not token:
                raise CodeBuddyDaemonError("CodeBuddy daemon login returned no token.")
            self._token = token

    def _reset_token(self) -> None:
        with self._token_lock:
            self._token = None

    # ----------------------------------------------------------- ACP session

    def _connect(self) -> str:
        status, raw = self._request("POST", "/api/v1/acp/connect", timeout=_CONNECT_TIMEOUT)
        if status != 200:
            raise CodeBuddyDaemonError(f"ACP connect failed (HTTP {status}).")
        connection_id = (json.loads(raw.decode("utf-8")) or {}).get("connectionId")
        if not connection_id:
            raise CodeBuddyDaemonError("ACP connect returned no connectionId.")
        return connection_id

    def _disconnect(self, connection_id: str) -> None:
        try:
            self._request("DELETE", "/api/v1/acp", timeout=10.0,
                          headers=self._headers() | {"acp-connection-id": connection_id})
        except Exception:
            pass  # request-scoped cleanup only; never disturb the resident server

    def _open_sse(self, connection_id: str, payload: dict, timeout: float):
        """POST one JSON-RPC message and return the raw streaming response (SSE lines)."""
        url = f"{self.base_url}/api/v1/acp"
        headers = self._headers() | {"acp-connection-id": connection_id}
        req = urllib.request.Request(url, data=json.dumps(payload).encode("utf-8"),
                                     method="POST", headers=headers)
        try:
            return urllib.request.urlopen(req, timeout=timeout)
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", "replace")[:300]
            if exc.code in (401, 403):
                self._reset_token()
            raise CodeBuddyDaemonError(f"ACP request failed (HTTP {exc.code}): {detail}") from exc
        except urllib.error.URLError as exc:
            raise CodeBuddyDaemonError(f"ACP request failed: {exc.reason}") from exc

    def _iter_sse(self, resp, deadline: float) -> Iterator[dict]:
        """Yield each JSON object carried on an SSE ``data:`` line."""
        for raw_line in resp:
            if time.monotonic() > deadline:
                raise CodeBuddyDaemonError("CodeBuddy completion exceeded its time budget.")
            line = raw_line.decode("utf-8", "replace").strip()
            if not line.startswith("data:"):
                continue  # ":ok" keepalives and "event:" lines
            chunk = line[5:].strip()
            if not chunk:
                continue
            try:
                yield json.loads(chunk)
            except json.JSONDecodeError:
                continue

    # ------------------------------------------------------------- interrupt

    def _deny_interrupt(self, connection_id: str, session_id: str, update: dict) -> None:
        """Auto-deny tool-permission prompts: Hermes owns tools, CodeBuddy must not act."""
        payload = {"jsonrpc": "2.0", "method": "_codebuddy.ai/resolveInterruption",
                   "params": {"sessionId": session_id, "toolCallId": update.get("toolCallId", ""),
                              "decision": "deny",
                              "answers": {"feedback": "Hermes owns tool execution; do not use built-in tools."}}}
        try:
            urllib.request.urlopen(urllib.request.Request(
                f"{self.base_url}/api/v1/acp", data=json.dumps(payload).encode("utf-8"), method="POST",
                headers=self._headers() | {"acp-connection-id": connection_id}), timeout=10).close()
        except Exception:
            pass

    # ------------------------------------------------------------- session

    def _setup_session(self, connection_id: str, model: str | None, timeout: float) -> tuple[str, dict]:
        self._rpc_call(connection_id, 1, "initialize",
                       {"protocolVersion": 1, "clientCapabilities": {}}, timeout=timeout)
        session = self._rpc_call(connection_id, 2, "session/new",
                                 {"cwd": self._session_cwd, "mcpServers": []}, timeout=timeout)
        session_id = session.get("sessionId")
        if not session_id:
            raise CodeBuddyDaemonError("session/new returned no sessionId.")
        self._maybe_select_model(connection_id, session_id, session, model, timeout)
        self._maybe_apply_thinking(connection_id, session_id, session, timeout)
        return session_id, session

    def _rpc_call(self, connection_id: str, rpc_id: int, method: str, params: dict, timeout: float) -> dict:
        """One request → its result dict (the response arrives on the POST's own SSE stream)."""
        payload = {"jsonrpc": "2.0", "id": rpc_id, "method": method, "params": params}
        resp = self._open_sse(connection_id, payload, timeout)
        try:
            for obj in self._iter_sse(resp, time.monotonic() + timeout):
                if obj.get("id") == rpc_id:
                    if "error" in obj:
                        err = obj["error"] or {}
                        raise CodeBuddyDaemonError(
                            f"ACP {method} error: {err.get('message', 'unknown')} ({err.get('data', '')})")
                    return obj.get("result") or {}
            raise CodeBuddyDaemonError(f"ACP {method}: stream ended without a response.")
        finally:
            try:
                resp.close()
            except Exception:
                pass

    def _maybe_select_model(self, connection_id: str, session_id: str, session: dict,
                            model: str | None, timeout: float) -> None:
        if not model:
            return
        models = (session.get("models") or {}).get("availableModels") or []
        if not models:
            return
        available = {str(m.get("modelId")) for m in models}
        target = model if model in available else next(
            (m for m in available if m.startswith(model)), None)
        current = (session.get("models") or {}).get("currentModelId")
        if target and target != current:
            try:
                self._rpc_call(connection_id, 99, "session/set_model",
                               {"sessionId": session_id, "modelId": target}, timeout=30.0)
            except CodeBuddyDaemonError:
                pass  # model selection is best-effort; the session keeps its default

    def _maybe_apply_thinking(self, connection_id: str, session_id: str, session: dict, timeout: float) -> None:
        """Best-effort ``session/set_config_option(thinking=...)`` to cut reasoning latency."""
        if not self._thinking:
            return
        options = session.get("configOptions") or []
        current = next((o.get("currentValue") for o in options if o.get("id") == "thinking"), None)
        if current is None or current == self._thinking:
            return
        try:
            self._rpc_call(connection_id, 98, "session/set_config_option",
                           {"sessionId": session_id, "configId": "thinking", "value": self._thinking},
                           timeout=30.0)
        except CodeBuddyDaemonError:
            pass  # cosmetic option; never fail a completion over it

    # ------------------------------------------------------------- prompt

    def _iter_prompt_events(self, connection_id: str, session_id: str, prompt_text: str,
                            timeout_s: float, prompt_rpc_id: int) -> Iterator[dict]:
        """Yield ``{"kind": "text"|"reasoning", "piece": str}`` events live, then
        ``{"kind": "final", "stop": ...}`` from the result."""
        payload = {"jsonrpc": "2.0", "id": prompt_rpc_id, "method": "session/prompt",
                   "params": {"sessionId": session_id,
                              "prompt": [{"type": "text", "text": prompt_text}]}}
        resp = self._open_sse(connection_id, payload, timeout_s)
        finished = False
        try:
            for obj in self._iter_sse(resp, time.monotonic() + timeout_s):
                if obj.get("id") == prompt_rpc_id:
                    if "error" in obj:
                        raise CodeBuddyDaemonError(f"ACP session/prompt error: {obj['error']}")
                    stop = (obj.get("result") or {}).get("stopReason")
                    if stop and stop != "end_turn":
                        raise CodeBuddyDaemonError(f"CodeBuddy turn ended with stopReason={stop}")
                    finished = True
                    yield {"kind": "final", "stop": stop or "end_turn"}
                    break
                params = obj.get("params") if isinstance(obj.get("params"), dict) else {}
                update = params.get("update") if isinstance(params.get("update"), dict) else {}
                kind = update.get("sessionUpdate")
                if kind == "agent_message_chunk":
                    piece = str((update.get("content") or {}).get("text", ""))
                    if piece:
                        yield {"kind": "text", "piece": piece}
                elif kind == "agent_thought_chunk":
                    piece = str((update.get("content") or {}).get("text", ""))
                    if piece:
                        yield {"kind": "reasoning", "piece": piece}
                elif kind == "interruption_request":
                    self._deny_interrupt(connection_id, session_id, update)
                elif kind == "session_end":
                    break  # safety net; the final result normally arrives first
            if not finished:
                raise CodeBuddyDaemonError("CodeBuddy turn produced no final result before the stream ended.")
        finally:
            try:
                resp.close()
            except Exception:
                pass

    def _new_turn(self, model: str | None, prompt_text: str, timeout_s: float) -> tuple[str, str, Iterator[dict]]:
        """Login → connect → session → prompt; returns (connection_id, session_id, event iterator)."""
        self._login()
        connection_id = self._connect()
        try:
            session_id, _session = self._setup_session(connection_id, model, _CONNECT_TIMEOUT)
            events = self._iter_prompt_events(connection_id, session_id, prompt_text,
                                              timeout_s, prompt_rpc_id=10)
            return connection_id, session_id, events
        except BaseException:
            self._disconnect(connection_id)
            raise

    # ------------------------------------------------------------ completion

    def _create_chat_completion(
        self, *, model: str | None = None, messages: list[dict] | None = None, timeout: float | None = None,
        tools: list[dict] | None = None, tool_choice: Any = None, stream: bool = False, **_: Any,
    ) -> Any:
        prompt_text = _format_messages_as_prompt(messages or [], model=model, tools=tools, tool_choice=tool_choice)
        timeout_s = _effective_timeout(timeout) or self._default_timeout

        if stream and not tools:
            return self._live_delta_stream(model, prompt_text, timeout_s)
        return self._buffered_completion(model, prompt_text, timeout_s, stream)

    # ---- live streaming (tool-free requests only) --------------------------

    def _live_delta_stream(self, model: str | None, prompt_text: str, timeout_s: float):
        """Yield OpenAI-style delta chunks as SSE events arrive; finish + usage last.

        One transparent auth retry is allowed only during setup (before the first yield);
        once deltas flow, errors propagate — no transparent replay after side effects.
        The whole turn runs under the per-server lock (concurrent turns are dropped by
        CodeBuddy serve — see module docstring).
        """
        with _serve_lock(self.base_url):
            connection_id = session_id = events = None
            for attempt in (1, 2):
                try:
                    connection_id, session_id, events = self._new_turn(model, prompt_text, timeout_s)
                    break
                except CodeBuddyDaemonError as exc:
                    if attempt == 1 and self._is_auth_error(exc):
                        self._reset_token()
                        continue
                    raise
            try:
                text_parts: list[str] = []
                got_any = False
                for ev in events:
                    got_any = True
                    if ev["kind"] == "text":
                        text_parts.append(ev["piece"])
                        yield self._delta_chunk(model, content=ev["piece"])
                    elif ev["kind"] == "reasoning":
                        yield self._delta_chunk(model, reasoning=ev["piece"])
                text = "".join(text_parts)
                if not got_any and not text:
                    raise CodeBuddyDaemonError("CodeBuddy turn produced no response.")
                # Tool-call JSON, if any, is parsed from the full text and attached to the final
                # delta as tool_calls; content is NOT repeated (it already streamed live).
                tool_calls, _cleaned = extract_tool_calls_from_text(text)
                finish_reason = "tool_calls" if tool_calls else "stop"
                if tool_calls:
                    yield self._delta_chunk(model, tool_calls=tool_calls, finish_reason=finish_reason)
                else:
                    yield self._delta_chunk(model, finish_reason=finish_reason)
                yield SimpleNamespace(choices=[], model=model or "codebuddy",
                                      usage=SimpleNamespace(prompt_tokens=0, completion_tokens=0, total_tokens=0))
            finally:
                self._disconnect(connection_id)

    @staticmethod
    def _delta_chunk(model: str | None, *, content: str | None = None, reasoning: str | None = None,
                     tool_calls: Any = None, finish_reason: str | None = None) -> SimpleNamespace:
        delta = SimpleNamespace(role="assistant" if finish_reason is None else None,
                                content=content, tool_calls=tool_calls,
                                reasoning_content=reasoning, reasoning=reasoning)
        return SimpleNamespace(
            choices=[SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason)],
            model=model or "codebuddy", usage=None,
        )

    # ---- buffered (non-stream path; and stream when tools were offered) ----

    def _buffered_completion(self, model: str | None, prompt_text: str, timeout_s: float,
                             stream: bool) -> Any:
        text = reasoning = None
        connection_id = None
        with _serve_lock(self.base_url):
            for attempt in (1, 2):
                try:
                    connection_id, session_id, events = self._new_turn(model, prompt_text, timeout_s)
                    text_parts: list[str] = []
                    reasoning_parts: list[str] = []
                    got_final = False
                    for ev in events:
                        got_final = got_final or ev["kind"] == "final"
                        if ev["kind"] == "text":
                            text_parts.append(ev["piece"])
                        elif ev["kind"] == "reasoning":
                            reasoning_parts.append(ev["piece"])
                    text, reasoning = "".join(text_parts), ("".join(reasoning_parts) or None)
                    if not text and not got_final:
                        raise CodeBuddyDaemonError("CodeBuddy turn produced no response.")
                    break
                except CodeBuddyDaemonError as exc:
                    if attempt == 1 and self._is_auth_error(exc):
                        self._reset_token()
                        continue
                    raise
                finally:
                    if connection_id:
                        self._disconnect(connection_id)

        tool_calls, cleaned_text = extract_tool_calls_from_text(text)
        message = SimpleNamespace(
            content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning,
            reasoning_content=reasoning, reasoning_details=None,
        )
        # usage: CodeBuddy's usage_update is CONTEXT occupancy, not billing tokens — report the
        # same unknown-zeros shape the bundled ACPClient path has always returned.
        completion = SimpleNamespace(
            choices=[SimpleNamespace(message=message, finish_reason="tool_calls" if tool_calls else "stop")],
            usage=SimpleNamespace(prompt_tokens=0, completion_tokens=0, total_tokens=0,
                                  prompt_tokens_details=SimpleNamespace(cached_tokens=0)),
            model=model or "codebuddy",
        )
        if stream:
            # Tool-bearing stream: full text buffered, tool-call JSON complete before deltas.
            return completion_to_stream_chunks(completion)
        return completion

    @staticmethod
    def _is_auth_error(exc: CodeBuddyDaemonError) -> bool:
        text = str(exc).lower()
        return "auth" in text and ("401" in text or "403" in text or "required" in text or "expired" in text)

6.4 Regression scripts

Two standalone scripts validated the client before and after cutover (plain text; tool bridge emitting get_time with correct JSON arguments; live-stream TTFT; two concurrent streams; wrong-password and unreachable-server error paths). Keep them — they are your contract tests for every CLI upgrade.

---

7. Step 4 — Cutover & verification

# 1) back up config
cp $HERMES_HOME/config.yaml $HERMES_HOME/config.yaml.bak-pre-cutover

# 2) switch the default model block:
# model:
#   default: <a modelId from session/new availableModels>
#   provider: codebuddy-daemon

# 3) restart the gateway so config + plugin load
systemctl --user restart hermes-gateway

Verification ladder (cheap → production):

1. Client level (in-process, no gateway): plain text reply; a tool schema that must come back as tool_calls with valid arguments; stream=True with sub-2s TTFT.
2. Hermes one-shot: hermes --provider codebuddy-daemon -m <model> -z "Reply with exactly: PONG".
3. Production: a real message to the bot; confirm in journalctl --user -u hermes-gateway which provider served the turn.

🚨 Never run hermes one-shots while the gateway is running on a small box. The CLI detects the running gateway and hands the turn over — you end up with two full agent processes (≈1 GB combined). On a 2 GB VM this OOM-kills the gateway. We did this three times so you don’t have to. Production validation = real messages through the gateway process itself.

---

8. Operations runbook

# daily
systemctl --user status|start|stop|restart codebuddy-acp
journalctl --user -u codebuddy-acp -f
curl -s -H "X-CodeBuddy-Request: 1" http://127.0.0.1:18090/api/v1/auth/status

# rollback (1 minute, session data untouched)
cp $HERMES_HOME/config.yaml.bak-pre-cutover $HERMES_HOME/config.yaml
systemctl --user restart hermes-gateway
systemctl --user disable --now codebuddy-acp     # optional

Monitoring: gateway RSS (>550 MB warn), host memory available (<300 MB warn), OOM in dmesg, turn error rate, non-end_turn stopReasons, TTFT, and a credits reconciliation channel on the CodeBuddy side (client-side usage is deliberately zero — see below).

---

9. Eleven pitfalls, compressed

#	Symptom	Root cause	Fix
1	spawn node EACCES as the service user	sudo/ssh inherited cwd the user can’t enter	cd into the user’s home first
2	“Authentication required” after copying settings	credential lives in ~/.local/share/CodeBuddyExtension/...	copy the XDG auth file (§4)
3	Every API call: Missing required header	CSRF-style design	send X-CodeBuddy-Request: 1
4	POST returns an empty body	response is an SSE stream (streamable-http semantics)	parse data: lines; require the SSE Accept header
5	session/new: “Server not initialized”	ACP state machine	initialize first, per connection
6	daemon start can’t disable tools	no --tools flag on the daemon subcommand	use --serve + --tools ""
7	Concurrent turns: loser’s result vanishes	serve serializes per account	client-side lock per base_url
8	Gateway OOM-killed	2 GB box, two agent processes	right-size the VM; never oneshot-while-gateway-runs
9	CLI updates itself, protocol drifts	auto-updater on	DISABLE_AUTOUPDATER=1; contract-test before upgrades
10	Treating usage_update as billing	it’s context occupancy (used/size), not tokens	report zero-value usage; reconcile credits separately
11	Plugin “not loaded”	plugins load at gateway start	install plugin before starting/restarting the gateway

---

10. Closing notes

• What this is not: it does not make inference free or local — CodeBuddy still runs the model in its cloud, under its quota and terms. The resident server removes process startup cost, nothing else.
• Stability: CodeBuddy’s HTTP API is labeled Beta; pin the CLI version, keep the two regression scripts as contract tests, and upgrade deliberately.
• Measured results on our deployment: text round-trips 5–20 s (thinking-heavy), warm-session one-shots ~2–5 s, streaming TTFT 1.2–2.3 s, full multi-turn tool loop through Hermes with zero core-loop patches.

If this guide saved you a weekend, the credit goes to the Hermes Agent and CodeBuddy teams — we just wired the plugs together.