zettelkasten.llm_adapter

zettelkasten.llm_adapter

Shared LLM-adapter plumbing for the zettelkasten literature-review loop.

The "run a one-shot in-process dashboard agent and parse its streamed SSE reply" pattern was copy-pasted across the claim-producer, outline, auto-note, tables, and remine passes. This module holds the single source of truth for it:

• :func:parse_sse — parse one Server-Sent-Event chunk into (event, data).
• :func:run_async — drive an awaitable to completion, tolerating a loop that is already running (re-raising worker-thread failures rather than swallowing them).
• :func:make_write_free_agent — construct the in-process zettelkasten dashboard agent with graph writes (and deletes) disabled.
• :func:run_agent_once — run a pre-built agent once and collect its final text.
• :func:run_write_free_agent — the convenience combination the callers use: build a write-free agent, run it once, return its text.

The dashboard/LLM stack is imported lazily inside :func:make_write_free_agent, so importing this module never pulls it in — the review loop stays optional.

parse_sse

parse_sse(chunk: str) -> tuple[str, dict[str, Any]]

Parse one Server-Sent-Event string into (event, data) (best-effort).

The whole per-line parse is wrapped in a broad guard that resets (event, data) to ("", {}) on any failure. This mirrors the outer try/except Exception the original propose/classify/summarize/extract copies wrapped their parse in: a malformed or unexpected chunk (e.g. a non-str chunk whose .splitlines() raises) is skipped rather than propagated, so the enclosing async for keeps streaming instead of the whole call failing. The inner json.loads guard is kept for a malformed data: payload.

Source code in zettelkasten/llm_adapter.py

def parse_sse(chunk: str) -> tuple[str, dict[str, Any]]:
    """Parse one Server-Sent-Event string into ``(event, data)`` (best-effort).

    The whole per-line parse is wrapped in a broad guard that resets
    ``(event, data)`` to ``("", {})`` on any failure. This mirrors the outer
    ``try/except Exception`` the original propose/classify/summarize/extract
    copies wrapped their parse in: a malformed or unexpected chunk (e.g. a
    non-``str`` chunk whose ``.splitlines()`` raises) is skipped rather than
    propagated, so the enclosing ``async for`` keeps streaming instead of the
    whole call failing. The inner ``json.loads`` guard is kept for a malformed
    ``data:`` payload.
    """
    event = ""
    data: dict[str, Any] = {}
    try:
        for line in chunk.splitlines():
            if line.startswith("event:"):
                event = line[len("event:"):].strip()
            elif line.startswith("data:"):
                try:
                    data = json.loads(line[len("data:"):].strip())
                    # A ``data:`` line can be valid JSON that is NOT a dict (e.g.
                    # ``data: 123`` -> int, ``data: [1,2]`` -> list). Callers do
                    # ``data.get(...)``, so normalize any non-dict payload to {}.
                    if not isinstance(data, dict):
                        data = {}
                except Exception:
                    data = {}
    except Exception:
        event, data = "", {}
    return event, data

run_async

run_async(coro, timeout: float | None = None) -> Any

Run an awaitable to completion, tolerating an already-running loop.

The review passes are synchronous but the dashboard agent is async. When no loop is running we just :func:asyncio.run; when one IS (e.g. called from inside a dashboard endpoint) we run the coroutine on a dedicated thread with its own loop so we never re-enter the caller's loop.

The worker thread captures BOTH success and failure: a bare box["value"] = asyncio.run(coro) silently drops any exception (the thread dies, the box stays empty, and the caller reads None → an empty result with no error). We re-raise it on the calling thread so a failure surfaces as a real error instead of a phantom "it just produced nothing".

timeout is a BACKSTOP for the worker-thread path only: if the coroutine does not finish within it we abandon the (daemon) worker and raise :class:TimeoutError so the caller is never wedged forever. The real, clean-teardown timeout lives inside the coroutine (see :func:run_agent_once, which wraps its stream in asyncio.wait_for); this is only the safety net for a coroutine that ignores cancellation. Pass a value slightly LARGER than the inner wait_for so the inner one fires first and the bridge is disposed cleanly.

Source code in zettelkasten/llm_adapter.py

def run_async(coro, timeout: float | None = None) -> Any:
    """Run an awaitable to completion, tolerating an already-running loop.

    The review passes are synchronous but the dashboard agent is async. When no
    loop is running we just :func:`asyncio.run`; when one IS (e.g. called from
    inside a dashboard endpoint) we run the coroutine on a dedicated thread with
    its own loop so we never re-enter the caller's loop.

    The worker thread captures BOTH success and failure: a bare
    ``box["value"] = asyncio.run(coro)`` silently drops any exception (the thread
    dies, the box stays empty, and the caller reads ``None`` → an empty result
    with no error). We re-raise it on the calling thread so a failure surfaces as
    a real error instead of a phantom "it just produced nothing".

    ``timeout`` is a BACKSTOP for the worker-thread path only: if the coroutine
    does not finish within it we abandon the (daemon) worker and raise
    :class:`TimeoutError` so the caller is never wedged forever. The real,
    clean-teardown timeout lives inside the coroutine (see
    :func:`run_agent_once`, which wraps its stream in ``asyncio.wait_for``); this
    is only the safety net for a coroutine that ignores cancellation. Pass a
    value slightly LARGER than the inner ``wait_for`` so the inner one fires
    first and the bridge is disposed cleanly.
    """
    import asyncio
    import threading

    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)

    box: dict[str, Any] = {}

    def _worker() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as exc:  # noqa: BLE001 — propagate verbatim below
            box["error"] = exc

    t = threading.Thread(target=_worker, daemon=True)
    t.start()
    t.join(timeout)
    if t.is_alive():
        # The worker ignored its inner cancellation and is still running. It is a
        # daemon thread, so abandoning it will not block interpreter exit; raise so
        # the caller (and the whole build) is not wedged behind a hung LLM call.
        raise TimeoutError(
            f"async agent run did not complete within {timeout:.0f}s"
        )
    if "error" in box:
        raise box["error"]
    return box.get("value")

make_write_free_agent

make_write_free_agent(name: str, system: str, *, disable_write: bool = True, graphs_dir: 'str | Path | None' = None, model: str = '')

Construct the in-process zettelkasten dashboard agent, writes disabled.

Lazy/optional by construction — the dashboard/LLM stack is imported INSIDE this function, so importing this module never requires it.

graphs_dir is the store root exported to the agent's zettelkasten MCP as ZETTELKASTEN_PATH. Each caller passes ITS OWN module's GRAPHS_DIR binding so a per-module monkeypatch.setattr(<caller>, "GRAPHS_DIR", tmp) is honored (the original inline agent-launch blocks read the caller module's binding, not this adapter's). It defaults to :data:zettelkasten.graph.GRAPHS_DIR only as a fallback when a caller omits it.

The agent's zettelkasten MCP is launched with ZK_DISABLE_DELETE=1 always, and additionally ZK_DISABLE_WRITE=1 when disable_write (the default). ZK_DISABLE_WRITE drops every graph-mutating tool (it implies ZK_DISABLE_DELETE); the delete flag is kept explicit for clarity and for older server builds that predate the write flag. A pass that hands the agent all its material in-prompt (it only returns JSON/prose) wants the write-free default; the DRAFT path keeps deletes disabled but leaves writes available. A non-empty model is forwarded as the requested Cursor model id. Empty keeps :class:DashboardAgent's configured default.

Source code in zettelkasten/llm_adapter.py

def make_write_free_agent(
    name: str,
    system: str,
    *,
    disable_write: bool = True,
    graphs_dir: "str | Path | None" = None,
    model: str = "",
):
    """Construct the in-process zettelkasten dashboard agent, writes disabled.

    Lazy/optional by construction — the dashboard/LLM stack is imported INSIDE
    this function, so importing this module never requires it.

    ``graphs_dir`` is the store root exported to the agent's zettelkasten MCP as
    ``ZETTELKASTEN_PATH``. Each caller passes ITS OWN module's ``GRAPHS_DIR``
    binding so a per-module ``monkeypatch.setattr(<caller>, "GRAPHS_DIR", tmp)``
    is honored (the original inline agent-launch blocks read the caller module's
    binding, not this adapter's). It defaults to
    :data:`zettelkasten.graph.GRAPHS_DIR` only as a fallback when a caller omits
    it.

    The agent's zettelkasten MCP is launched with ``ZK_DISABLE_DELETE=1`` always,
    and additionally ``ZK_DISABLE_WRITE=1`` when ``disable_write`` (the default).
    ``ZK_DISABLE_WRITE`` drops every graph-mutating tool (it implies
    ``ZK_DISABLE_DELETE``); the delete flag is kept explicit for clarity and for
    older server builds that predate the write flag. A pass that hands the agent
    all its material in-prompt (it only returns JSON/prose) wants the write-free
    default; the DRAFT path keeps deletes disabled but leaves writes available.
    A non-empty ``model`` is forwarded as the requested Cursor model id.
    Empty keeps :class:`DashboardAgent`'s configured default.
    """
    from memory.dashboard.backend.agent import DashboardAgent

    root = GRAPHS_DIR if graphs_dir is None else graphs_dir
    workspace = Path(os.environ.get("ANGELO_WORKSPACE") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()).expanduser()
    env = {"ZETTELKASTEN_PATH": str(root), "ZK_DISABLE_DELETE": "1"}
    if disable_write:
        env["ZK_DISABLE_WRITE"] = "1"
    agent_kwargs: dict[str, Any] = {}
    if model:
        agent_kwargs["model"] = model
    return DashboardAgent(
        workspace=workspace,
        name=name,
        system_context=system,
        mcp_servers={
            "zettelkasten": {
                "command": "angelo-zettelkasten",
                "env": env,
            },
        },
        **agent_kwargs,
    )

make_toolless_agent

make_toolless_agent(name: str, system: str, *, model: str = '')

Construct the in-process dashboard agent with NO MCP servers attached.

A HARD grounding guarantee for pure synthesis passes: with an empty mcp_servers the agent has no tools whatsoever, so it CANNOT read the graph (or anything else) and can only work from the material handed to it in the prompt. Contrast :func:make_write_free_agent, which still attaches a read-only zettelkasten MCP (writes disabled, but reads available). Use this when the prompt already carries all the evidence and any extra retrieval would be a grounding leak. A non-empty model is forwarded as the requested Cursor model id; empty keeps :class:DashboardAgent's configured default. Lazy/optional by construction — the dashboard/LLM stack is imported INSIDE this function, so importing this module never requires it.

Source code in zettelkasten/llm_adapter.py

def make_toolless_agent(name: str, system: str, *, model: str = ""):
    """Construct the in-process dashboard agent with NO MCP servers attached.

    A HARD grounding guarantee for pure synthesis passes: with an empty
    ``mcp_servers`` the agent has no tools whatsoever, so it CANNOT read the graph
    (or anything else) and can only work from the material handed to it in the
    prompt. Contrast :func:`make_write_free_agent`, which still attaches a
    read-only zettelkasten MCP (writes disabled, but reads available). Use this
    when the prompt already carries all the evidence and any extra retrieval would
    be a grounding leak. A non-empty ``model`` is forwarded as the requested
    Cursor model id; empty keeps :class:`DashboardAgent`'s configured default.
    Lazy/optional by construction — the dashboard/LLM stack is imported INSIDE
    this function, so importing this module never requires it.
    """
    from memory.dashboard.backend.agent import DashboardAgent

    workspace = Path(os.environ.get("ANGELO_WORKSPACE") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()).expanduser()
    agent_kwargs: dict[str, Any] = {}
    if model:
        agent_kwargs["model"] = model
    return DashboardAgent(
        workspace=workspace,
        name=name,
        system_context=system,
        mcp_servers={},
        **agent_kwargs,
    )

run_agent_once

run_agent_once(agent, prompt: str, *, error_message: str = 'agent error', timeout: float | None = None) -> str

Run a pre-built dashboard agent once and return its final text.

Streams the agent's SSE reply, preferring the final result text and falling back to the assembled assistant deltas. An error event is turned into a :class:RuntimeError (carrying error_message when the event omits its own message) so the caller/route can surface it. The agent is always disposed. Blocks via :func:run_async.

timeout (seconds) bounds the whole stream: when set, the streaming is wrapped in :func:asyncio.wait_for, so a bridge that launches but never returns a completion (the failure mode when the cursor-sdk agent runtime does not come up — e.g. inside a stdio MCP server) raises :class:TimeoutError instead of hanging indefinitely. On timeout the agent is still disposed (the finally), tearing the bridge down so a caller degrades gracefully rather than wedging. None (the default) preserves the historical unbounded behavior for callers that block on their own loop (e.g. the dashboard).

Source code in zettelkasten/llm_adapter.py

def run_agent_once(
    agent, prompt: str, *, error_message: str = "agent error", timeout: float | None = None
) -> str:
    """Run a pre-built dashboard ``agent`` once and return its final text.

    Streams the agent's SSE reply, preferring the final ``result`` text and
    falling back to the assembled ``assistant`` deltas. An ``error`` event is
    turned into a :class:`RuntimeError` (carrying ``error_message`` when the event
    omits its own message) so the caller/route can surface it. The agent is always
    disposed. Blocks via :func:`run_async`.

    ``timeout`` (seconds) bounds the whole stream: when set, the streaming is
    wrapped in :func:`asyncio.wait_for`, so a bridge that launches but never
    returns a completion (the failure mode when the cursor-sdk agent runtime does
    not come up — e.g. inside a stdio MCP server) raises :class:`TimeoutError`
    instead of hanging indefinitely. On timeout the agent is still disposed (the
    ``finally``), tearing the bridge down so a caller degrades gracefully rather
    than wedging. ``None`` (the default) preserves the historical unbounded
    behavior for callers that block on their own loop (e.g. the dashboard).
    """
    import asyncio

    async def _run() -> str:
        final = ""
        assembled = ""

        async def _drive() -> None:
            nonlocal final, assembled
            async for chunk in agent.stream_chat(prompt, graph=None):
                event, data = parse_sse(chunk)
                if event == "result":
                    final = data.get("text", final)
                elif event == "assistant":
                    assembled += data.get("text", "")
                elif event == "error":
                    raise RuntimeError(data.get("message", error_message))

        try:
            if timeout and timeout > 0:
                await asyncio.wait_for(_drive(), timeout=timeout)
            else:
                await _drive()
        finally:
            await agent.dispose()
        return final or assembled

    # Give the worker-thread backstop a little headroom over the inner wait_for so
    # the inner timeout (which disposes the bridge cleanly) fires first.
    join_timeout = (timeout + 15.0) if (timeout and timeout > 0) else None
    return run_async(_run(), timeout=join_timeout) or ""

run_write_free_agent

run_write_free_agent(name: str, system: str, prompt: str, *, disable_write: bool = True, error_message: str = 'agent error', graphs_dir: 'str | Path | None' = None, timeout: float | None = None, model: str = '') -> str

Build a write-free in-process agent, run it once, and return its text.

The convenience combination of :func:make_write_free_agent and :func:run_agent_once that the review passes call. disable_write, error_message, graphs_dir and timeout are forwarded so each call site keeps its exact launch discipline, error string, per-module GRAPHS_DIR monkeypatch surface, requested model, and (optional) per-call timeout.

Source code in zettelkasten/llm_adapter.py

def run_write_free_agent(
    name: str,
    system: str,
    prompt: str,
    *,
    disable_write: bool = True,
    error_message: str = "agent error",
    graphs_dir: "str | Path | None" = None,
    timeout: float | None = None,
    model: str = "",
) -> str:
    """Build a write-free in-process agent, run it once, and return its text.

    The convenience combination of :func:`make_write_free_agent` and
    :func:`run_agent_once` that the review passes call. ``disable_write``,
    ``error_message``, ``graphs_dir`` and ``timeout`` are forwarded so each call
    site keeps its exact launch discipline, error string, per-module
    ``GRAPHS_DIR`` monkeypatch surface, requested ``model``, and (optional)
    per-call timeout.
    """
    agent = make_write_free_agent(
        name, system, disable_write=disable_write, graphs_dir=graphs_dir,
        model=model,
    )
    return run_agent_once(agent, prompt, error_message=error_message, timeout=timeout)

run_toolless_agent

run_toolless_agent(name: str, system: str, prompt: str, *, error_message: str = 'agent error', timeout: float | None = None, model: str = '') -> str

Build a TOOL-FREE in-process agent, run it once, and return its text.

The hard-grounding companion to :func:run_write_free_agent: the agent has no MCP servers, so it works only from prompt. For synthesis passes that must not touch the graph at all. model is the requested Cursor model id and is forwarded to :class:DashboardAgent; timeout (seconds) bounds the call so a bridge that never returns a completion degrades to a raised :class:TimeoutError instead of hanging.

Source code in zettelkasten/llm_adapter.py

def run_toolless_agent(
    name: str,
    system: str,
    prompt: str,
    *,
    error_message: str = "agent error",
    timeout: float | None = None,
    model: str = "",
) -> str:
    """Build a TOOL-FREE in-process agent, run it once, and return its text.

    The hard-grounding companion to :func:`run_write_free_agent`: the agent has no
    MCP servers, so it works only from ``prompt``. For synthesis passes that must
    not touch the graph at all. ``model`` is the requested Cursor model id and is
    forwarded to :class:`DashboardAgent`; ``timeout`` (seconds) bounds the call
    so a bridge that never returns a completion degrades to a raised
    :class:`TimeoutError` instead of hanging.
    """
    return run_agent_once(
        make_toolless_agent(name, system, model=model), prompt,
        error_message=error_message, timeout=timeout,
    )