mcp: add the daemon socket protocol

packages/simplex-chat-mcp/src/protocol.ts

@@ -0,0 +1,50 @@
+import {CallToolResult} from "@modelcontextprotocol/server"
+import {Config} from "./config"
+import {ToolName} from "./tools"
+
+export const protocolVersion = 1
+
+export interface DaemonInfo {
+  version: string
+  protocol: number
+}
+
+export interface OwnerState {
+  owners: string[]
+  link?: string
+}
+
+export interface SessionRequest {
+  id: string
+  name?: string
+  project: string
+  config: Config
+}
+
+export interface SessionReply extends DaemonInfo {
+  number: number
+  displayName: string
+  owner: OwnerState
+}
+
+export interface AttachRequest {
+  id: string
+}
+
+export interface CallRequest {
+  id: string
+  tool: ToolName
+  input: unknown
+}
+
+export type CallLine = {progress: string} | {result: CallToolResult | null}
+
+export interface StatusReply extends DaemonInfo {
+  mainGroup: string
+  owner: OwnerState
+  sessions: {number: number; name: string; attached: number}[]
+}
+
+export interface ResetReply {
+  link: string
+}

packages/simplex-chat-mcp/src/socket.ts

@@ -0,0 +1,129 @@
+import {timingSafeEqual} from "crypto"
+import {existsSync, unlinkSync} from "fs"
+import {IncomingMessage, Server, ServerResponse, createServer, request} from "http"
+import {logError} from "./log"
+
+const maxBody = 1_048_576
+
+export type Handler = (body: unknown, res: ServerResponse, req: IncomingMessage) => Promise<void>
+
+export class SocketError extends Error {
+  constructor(message: string, readonly status: number) {
+    super(message)
+  }
+}
+
+export function listen(path: string, token: string, routes: {[route: string]: Handler}): Promise<Server> {
+  const expected = Buffer.from(`Bearer ${token}`)
+  const authorized = (header: string | undefined): boolean => {
+    const given = Buffer.from(header ?? "")
+    return given.length === expected.length && timingSafeEqual(given, expected)
+  }
+  const server = createServer((req, res) => {
+    if (!authorized(req.headers.authorization)) return sendJson(res, 401, {error: "Unauthorized"})
+    const handler = routes[`${req.method} ${req.url}`]
+    if (handler === undefined) return sendJson(res, 404, {error: "Not found"})
+    readBody(req)
+      .then(body => handler(body, res, req))
+      .catch(error => {
+        if (error instanceof SocketError) return sendJson(res, error.status, {error: error.message})
+        logError(`${req.method} ${req.url}`, error)
+        if (!res.headersSent) sendJson(res, 500, {error: error instanceof Error ? error.message : String(error)})
+        else res.destroy()
+      })
+  })
+  server.requestTimeout = 0
+  server.headersTimeout = 60_000
+  server.keepAliveTimeout = 5_000
+  server.timeout = 0
+  if (process.platform !== "win32" && existsSync(path)) unlinkSync(path)
+  return new Promise((resolve, reject) => {
+    server.once("error", reject)
+    server.listen(path, () => {
+      server.off("error", reject)
+      resolve(server)
+    })
+  })
+}
+
+export function closeServer(server: Server): Promise<void> {
+  const closed = new Promise<void>(resolve => server.close(() => resolve()))
+  server.closeAllConnections()
+  return closed
+}
+
+export function sendJson(res: ServerResponse, status: number, value: unknown): void {
+  res.writeHead(status, {"content-type": "application/json"})
+  res.end(JSON.stringify(value))
+}
+
+export function startLines(res: ServerResponse): (line: unknown) => void {
+  res.writeHead(200, {"content-type": "application/x-ndjson"})
+  return line => {
+    res.write(`${JSON.stringify(line)}\n`)
+  }
+}
+
+export function isNotRunning(error: unknown): boolean {
+  const code = (error as NodeJS.ErrnoException | undefined)?.code
+  return code === "ENOENT" || code === "ECONNREFUSED"
+}
+
+export async function requestJson<T>(path: string, token: string, method: string, route: string, body?: unknown): Promise<T> {
+  const res = await send(path, token, method, route, body)
+  const text = await readText(res)
+  if (res.statusCode !== 200) throw new SocketError(errorOf(text), res.statusCode ?? 0)
+  return JSON.parse(text) as T
+}
+
+export async function streamLines(path: string, token: string, route: string, body: unknown, onLine: (line: unknown) => void, signal?: AbortSignal): Promise<void> {
+  const res = await send(path, token, "POST", route, body, signal)
+  if (res.statusCode !== 200) throw new SocketError(errorOf(await readText(res)), res.statusCode ?? 0)
+  let buffer = ""
+  for await (const chunk of res) {
+    buffer += chunk
+    for (let i = buffer.indexOf("\n"); i >= 0; i = buffer.indexOf("\n")) {
+      onLine(JSON.parse(buffer.slice(0, i)))
+      buffer = buffer.slice(i + 1)
+    }
+  }
+}
+
+function send(path: string, token: string, method: string, route: string, body: unknown, signal?: AbortSignal): Promise<IncomingMessage> {
+  return new Promise((resolve, reject) => {
+    const req = request({socketPath: path, method, path: route, headers: {authorization: `Bearer ${token}`, "content-type": "application/json"}, signal}, resolve)
+    req.on("error", reject)
+    req.end(body === undefined ? undefined : JSON.stringify(body))
+  })
+}
+
+async function readBody(req: IncomingMessage): Promise<unknown> {
+  let text = ""
+  let tooLarge = false
+  for await (const chunk of req) {
+    if (tooLarge) continue
+    text += chunk
+    tooLarge = text.length > maxBody
+  }
+  if (tooLarge) throw new SocketError("Body too large", 413)
+  if (text === "") return undefined
+  try {
+    return JSON.parse(text)
+  } catch {
+    throw new SocketError("Invalid JSON", 400)
+  }
+}
+
+async function readText(res: IncomingMessage): Promise<string> {
+  let text = ""
+  for await (const chunk of res) text += chunk
+  return text
+}
+
+function errorOf(text: string): string {
+  try {
+    return (JSON.parse(text) as {error?: string}).error ?? text
+  } catch {
+    return text
+  }
+}

packages/simplex-chat-mcp/test/unit/socket.test.ts

@@ -0,0 +1,71 @@
+import {once} from "events"
+import {existsSync, mkdtempSync, writeFileSync} from "fs"
+import {Server} from "http"
+import {tmpdir} from "os"
+import {join} from "path"
+import {SocketError, closeServer, isNotRunning, listen, requestJson, sendJson, startLines, streamLines} from "../../src/socket"
+
+let dir: string
+let path: string
+let server: Server | undefined
+const token = "t0ken"
+
+beforeEach(() => {
+  dir = mkdtempSync(join(tmpdir(), "simplex-mcp-socket-"))
+  path = join(dir, "daemon.sock")
+})
+
+afterEach(async () => {
+  if (server) await closeServer(server)
+  server = undefined
+})
+
+describe("socket", () => {
+  it("checks the token, the route and the body", async () => {
+    server = await listen(path, token, {
+      "POST /echo": async (body, res) => sendJson(res, 200, body),
+      "POST /fail": async () => {
+        throw new Error("broken")
+      },
+    })
+    expect(await requestJson(path, token, "POST", "/echo", {a: 1})).toEqual({a: 1})
+    await expect(requestJson(path, "wrong", "POST", "/echo", {})).rejects.toEqual(new SocketError("Unauthorized", 401))
+    await expect(requestJson(path, token, "GET", "/echo")).rejects.toEqual(new SocketError("Not found", 404))
+    await expect(requestJson(path, token, "POST", "/fail", {})).rejects.toEqual(new SocketError("broken", 500))
+    await expect(requestJson(path, token, "POST", "/echo", "x".repeat(1_048_577))).rejects.toEqual(new SocketError("Body too large", 413))
+  })
+
+  it("streams lines and aborts the request on the signal", async () => {
+    let closedEarly: Promise<unknown> | undefined
+    server = await listen(path, token, {
+      "POST /lines": async (_body, res) => {
+        const write = startLines(res)
+        write({progress: "one"})
+        write({progress: "two"})
+        closedEarly = once(res, "close")
+      },
+    })
+    const lines: unknown[] = []
+    const controller = new AbortController()
+    const streaming = streamLines(path, token, "/lines", {}, line => {
+      lines.push(line)
+      if (lines.length === 2) controller.abort()
+    }, controller.signal)
+    await expect(streaming).rejects.toThrow()
+    expect(lines).toEqual([{progress: "one"}, {progress: "two"}])
+    await closedEarly
+  })
+
+  it("reports a daemon that is not running by the error code", async () => {
+    const error = await requestJson(path, token, "GET", "/status").catch(e => e)
+    expect(isNotRunning(error)).toBe(true)
+    expect(isNotRunning(new SocketError("x", 500))).toBe(false)
+  })
+
+  it("replaces a stale socket file", async () => {
+    writeFileSync(path, "")
+    server = await listen(path, token, {"GET /ok": async (_body, res) => sendJson(res, 200, {})})
+    expect(existsSync(path)).toBe(true)
+    expect(await requestJson(path, token, "GET", "/ok")).toEqual({})
+  })
+})