Eureka/tools/mind.py

#!/usr/bin/env python3
"""The one way a Eureka script reaches the mind: the eureka-state stdio MCP binding.

call(tool, arguments, deadline) spawns the binding, does initialize,
notifications/initialized and tools/call, and returns the parsed JSON body of the
answer. It raises MindError for every failure and kills the child on every exit
path. It never retries; retry_once(fn) is the caller's choice. Standard library only.

The binding: EUREKA_STATE_EXE overrides the command; ~/.claude.json
mcpServers.eureka-state gives command, args and env; the caller's environment wins
over the binding's env. A MindError names a kind and a detail and never echoes an
input value (an endpoint, a path, a line of config).
"""
import json
import os
import queue
import subprocess
import threading
import time

MAX_ANSWER_BYTES = 8 << 20

ERROR_CODES = ("Unavailable", "Rejected", "Misconfigured", "TooLarge",
               "Unencodable", "InvalidInput", "Internal")


class MindError(Exception):
    """kind: binding (no binding or no command), transport (spawn, pipe, deadline),
    tool (the daemon refused; detail is an ERROR_CODES entry or "error"), answer
    (the body was not JSON). detail is the code, always "Unavailable" for transport."""

    def __init__(self, kind, detail):
        super().__init__(kind + ": " + detail)
        self.kind, self.detail = kind, detail

    @property
    def unavailable(self):
        return self.detail == "Unavailable"


def binding():
    exe = os.environ.get("EUREKA_STATE_EXE")
    args = []
    env = dict(os.environ)
    try:
        with open(os.path.join(os.path.expanduser("~"), ".claude.json"), encoding="utf-8") as f:
            b = json.load(f)["mcpServers"]["eureka-state"]
        exe = exe or b.get("command")
        args = list(b.get("args", []))
        for k, v in (b.get("env") or {}).items():
            env.setdefault(k, v)
    except (OSError, ValueError, KeyError, TypeError):
        if not exe:
            raise MindError("binding", "eureka-state is not configured")
    if not exe:
        raise MindError("binding", "eureka-state has no command")
    return [exe, *args], env


def call(tool, arguments, deadline, max_bytes=MAX_ANSWER_BYTES):
    """One tool call within `deadline` seconds in all, reading at most `max_bytes` of
    the binding's output (a larger answer is MindError answer). Returns the parsed
    JSON body."""
    end = time.monotonic() + deadline
    argv, env = binding()
    try:
        proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                stderr=subprocess.DEVNULL, env=env)
    except OSError:
        raise MindError("transport", "Unavailable")
    lines = queue.Queue()

    def pump():
        room = max_bytes
        while True:
            ln = proc.stdout.readline(room + 1)
            if not ln:
                break
            room -= len(ln)
            if room < 0:
                lines.put(MindError("answer", "answer too large"))
                return
            lines.put(ln)
        lines.put(None)
    threading.Thread(target=pump, daemon=True).start()

    def send(obj):
        proc.stdin.write((json.dumps(obj) + "\n").encode("utf-8"))
        proc.stdin.flush()

    def reply(want):
        while True:
            try:
                ln = lines.get(timeout=max(end - time.monotonic(), 0))
            except queue.Empty:
                raise MindError("transport", "Unavailable")
            if ln is None:
                raise MindError("transport", "Unavailable")
            if isinstance(ln, MindError):
                raise ln
            try:
                msg = json.loads(ln)
            except ValueError:
                continue
            if isinstance(msg, dict) and msg.get("id") == want:
                return msg

    try:
        send({"jsonrpc": "2.0", "id": 1, "method": "initialize",
              "params": {"protocolVersion": "2024-11-05", "capabilities": {},
                         "clientInfo": {"name": "eureka-tools", "version": "1"}}})
        reply(1)
        send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        send({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
              "params": {"name": tool, "arguments": arguments}})
        res = reply(2).get("result") or {}
    except OSError:
        raise MindError("transport", "Unavailable")
    finally:
        try:
            proc.stdin.close()
        except OSError:
            pass
        try:
            proc.kill()
            proc.wait(timeout=2)
        except (OSError, subprocess.TimeoutExpired):
            pass
    body = "".join(c.get("text", "") for c in res.get("content", []) if isinstance(c, dict))
    if res.get("isError"):
        raise MindError("tool", next((c for c in ERROR_CODES if c in body), "error"))
    try:
        return json.loads(body)
    except ValueError:
        raise MindError("answer", "unparsable answer")


def retry_once(fn):
    """Run fn; on an Unavailable MindError run it once more. Any other error stands."""
    try:
        return fn()
    except MindError as first:
        if not first.unavailable:
            raise
    return fn()