diff --git a/bun.lock b/bun.lock index 5de4743d..3e767fce 100644 --- a/bun.lock +++ b/bun.lock @@ -315,6 +315,24 @@ "vue": ">=3.3.0", }, }, + "packages/typescript/integrations/bridge-server": { + "name": "@slop-ai/bridge-server", + "version": "0.1.0", + "bin": { + "slop-bridge-server": "dist/cli.js", + }, + "dependencies": { + "@modelcontextprotocol/sdk": "^1.29.0", + "@slop-ai/consumer": "workspace:*", + "@slop-ai/relay-cli": "workspace:*", + "zod": "^4.3.6", + }, + "devDependencies": { + "@types/node": "^25.6.0", + "@types/ws": "^8.5.0", + "ws": "^8.18.0", + }, + }, "packages/typescript/integrations/claude/slop-mcp-proxy/servers": { "name": "slop-bridge-mcp-proxy", "version": "0.1.0", @@ -392,6 +410,22 @@ "openclaw": ">=2026.0.0", }, }, + "packages/typescript/integrations/relay-cli": { + "name": "@slop-ai/relay-cli", + "version": "0.1.0", + "bin": { + "slop-relay": "dist/cli.js", + }, + "dependencies": { + "@slop-ai/consumer": "workspace:*", + "@slop-ai/discovery": "workspace:*", + "ws": "^8.18.0", + }, + "devDependencies": { + "@types/node": "^25.6.0", + "@types/ws": "^8.5.0", + }, + }, "packages/typescript/sdk/client": { "name": "@slop-ai/client", "version": "0.1.0", @@ -1202,6 +1236,8 @@ "@slop-ai/angular": ["@slop-ai/angular@workspace:packages/typescript/adapters/angular"], + "@slop-ai/bridge-server": ["@slop-ai/bridge-server@workspace:packages/typescript/integrations/bridge-server"], + "@slop-ai/client": ["@slop-ai/client@workspace:packages/typescript/sdk/client"], "@slop-ai/consumer": ["@slop-ai/consumer@workspace:packages/typescript/sdk/consumer"], @@ -1216,6 +1252,8 @@ "@slop-ai/react": ["@slop-ai/react@workspace:packages/typescript/adapters/react"], + "@slop-ai/relay-cli": ["@slop-ai/relay-cli@workspace:packages/typescript/integrations/relay-cli"], + "@slop-ai/server": ["@slop-ai/server@workspace:packages/typescript/sdk/server"], "@slop-ai/solid": ["@slop-ai/solid@workspace:packages/typescript/adapters/solid"], diff --git a/fly.toml b/fly.toml new file mode 100644 index 00000000..7305fe01 --- /dev/null +++ b/fly.toml @@ -0,0 +1,20 @@ +app = "slop-bridge" +primary_region = "iad" + +[build] + dockerfile = "packages/typescript/integrations/bridge-server/Dockerfile" + +[env] + PORT = "8080" + +[http_service] + internal_port = 8080 + force_https = true + auto_stop_machines = "off" + auto_start_machines = true + min_machines_running = 1 + +[[vm]] + memory = "512mb" + cpu_kind = "shared" + cpus = 1 diff --git a/packages/typescript/integrations/bridge-server/Dockerfile b/packages/typescript/integrations/bridge-server/Dockerfile new file mode 100644 index 00000000..aafb5e07 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/Dockerfile @@ -0,0 +1,26 @@ +FROM oven/bun:1-alpine AS build + +WORKDIR /app + +COPY package.json bun.lock ./ +# Drop workspace globs that don't ship in this build context (apps/, examples/, website/, benchmarks/) +RUN bun -e "const fs=require('fs');const p=JSON.parse(fs.readFileSync('package.json','utf8'));p.workspaces=p.workspaces.filter(w=>w.startsWith('packages/typescript'));fs.writeFileSync('package.json',JSON.stringify(p,null,2));" +COPY packages/typescript ./packages/typescript + +RUN bun install +RUN cd packages/typescript/integrations/relay-cli && bun run build +RUN cd packages/typescript/integrations/bridge-server && bun run build + +FROM oven/bun:1-alpine + +WORKDIR /app +ENV NODE_ENV=production +ENV PORT=8080 + +COPY --from=build /app/package.json /app/bun.lock ./ +COPY --from=build /app/node_modules ./node_modules +COPY --from=build /app/packages/typescript ./packages/typescript + +EXPOSE 8080 + +CMD ["bun", "packages/typescript/integrations/bridge-server/dist/cli.js"] diff --git a/packages/typescript/integrations/bridge-server/README.md b/packages/typescript/integrations/bridge-server/README.md new file mode 100644 index 00000000..6a1f463d --- /dev/null +++ b/packages/typescript/integrations/bridge-server/README.md @@ -0,0 +1,49 @@ +# @slop-ai/bridge-server + +Hosted bridge for SLOP relay clients and remote MCP hosts. + +The bridge exposes: + +- `POST /mcp`, `GET /mcp`, and `DELETE /mcp` for Streamable HTTP MCP clients. +- `GET /relay` as an authenticated WebSocket endpoint for `slop-relay`. +- `GET /healthz` for health checks. + +## Configuration + +`SLOP_BRIDGE_USERS` is required. It is a JSON object keyed by user id: + +```json +{ + "user_123": { + "mcpToken": "mcp-token-at-least-16-chars", + "relayToken": "relay-token-at-least-16-chars", + "label": "optional display label" + } +} +``` + +MCP hosts connect with `Authorization: Bearer `. + +Local relay agents connect with `Authorization: Bearer `. + +## Local Development + +```sh +SLOP_BRIDGE_USERS='{"dev":{"mcpToken":"dev-mcp-token-0001","relayToken":"dev-relay-token-0001"}}' \ + bun run dev -- --port 8080 +``` + +Point `slop-relay` at the bridge: + +```sh +SLOP_RELAY_TOKEN=dev-relay-token-0001 \ + slop-relay --url ws://localhost:8080/relay +``` + +Use `http://localhost:8080/mcp` as the remote MCP URL with the MCP token. + +## Build + +```sh +bun run build +``` diff --git a/packages/typescript/integrations/bridge-server/__tests__/end-to-end.test.ts b/packages/typescript/integrations/bridge-server/__tests__/end-to-end.test.ts new file mode 100644 index 00000000..af5dc9be --- /dev/null +++ b/packages/typescript/integrations/bridge-server/__tests__/end-to-end.test.ts @@ -0,0 +1,171 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js"; +import type { SlopNode } from "@slop-ai/consumer"; +import type { Down, Up } from "@slop-ai/relay-cli/protocol"; +import WebSocket from "ws"; +import { type BridgeServerHandle, createBridgeServer } from "../src/http"; +import { parseTokenRegistry } from "../src/tokens"; + +const MCP_TOKEN = "test-mcp-token-0001"; +const RELAY_TOKEN = "test-relay-token-0001"; + +const tree: SlopNode = { + id: "root", + type: "app", + properties: { title: "Fake App" }, + children: [ + { + id: "button", + type: "button", + properties: { label: "Increment" }, + affordances: [ + { + action: "click", + description: "Increment the counter.", + params: { type: "object", properties: {} }, + }, + ], + }, + ], +}; + +describe("bridge server", () => { + let server: BridgeServerHandle | null = null; + let relay: WebSocket | null = null; + const relayFrames: Down[] = []; + + beforeEach(async () => { + relayFrames.length = 0; + server = createBridgeServer({ + port: 0, + hostname: "127.0.0.1", + tokens: parseTokenRegistry( + JSON.stringify({ + user_test: { mcpToken: MCP_TOKEN, relayToken: RELAY_TOKEN }, + }), + ), + logger: { info: () => {}, error: () => {} }, + idleTimeoutMs: 60_000, + }); + + relay = await connectRelay(server.port, relayFrames); + }); + + afterEach(async () => { + relay?.close(); + relay = null; + server?.stop(); + server = null; + }); + + test("lists relayed apps, opens state, invokes actions, and unsubscribes on session close", async () => { + if (!server) throw new Error("server not started"); + + const transport = new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${server.port}/mcp`), { + requestInit: { + headers: { Authorization: `Bearer ${MCP_TOKEN}` }, + }, + }); + const client = new Client({ name: "bridge-test", version: "0.0.0" }); + await client.connect(transport); + + const listResult = (await client.callTool({ name: "list_apps", arguments: {} })) as CallToolResult; + const listPayload = listResult.structuredContent as { providers: { id: string; connected: boolean }[] }; + expect(listPayload.providers).toContainEqual(expect.objectContaining({ id: "fake-app", connected: true })); + + const openResult = (await client.callTool({ + name: "open_app", + arguments: { app: "fake-app" }, + })) as CallToolResult; + const openPayload = openResult.structuredContent as { selected: { id: string; tree: SlopNode } }; + expect(openPayload.selected.id).toBe("fake-app"); + expect(openPayload.selected.tree.id).toBe("root"); + expect(relayFrames).toContainEqual(expect.objectContaining({ t: "subscribe", providerId: "fake-app" })); + + const actionResult = (await client.callTool({ + name: "app_action", + arguments: { app: "fake-app", path: "/button", action: "click", params: {} }, + })) as CallToolResult; + expect(actionResult.isError).not.toBe(true); + expect(actionResult.content?.[0]?.text).toContain("Done"); + expect(relayFrames).toContainEqual( + expect.objectContaining({ t: "invoke", providerId: "fake-app", path: "/button", action: "click" }), + ); + + await transport.terminateSession(); + await waitFor(() => relayFrames.some((frame) => frame.t === "unsubscribe")); + }); +}); + +async function connectRelay(port: number, received: Down[]): Promise { + const relay = new WebSocket(`ws://127.0.0.1:${port}/relay`, { + headers: { Authorization: `Bearer ${RELAY_TOKEN}` }, + }); + relay.on("message", (raw) => { + const frame = JSON.parse(raw.toString()) as Down; + received.push(frame); + if (frame.t === "subscribe") { + sendRelay(relay, { + t: "snapshot", + reqId: frame.reqId, + subId: frame.subId, + tree, + version: 1, + seq: 0, + }); + sendRelay(relay, { + t: "patch", + subId: frame.subId, + ops: [{ op: "replace", path: "/properties/title", value: "Fake App Updated" }], + version: 2, + seq: 1, + }); + return; + } + if (frame.t === "invoke") { + sendRelay(relay, { + t: "result", + reqId: frame.reqId, + ok: true, + data: { invoked: frame.action }, + }); + } + }); + + await new Promise((resolve, reject) => { + relay.once("open", () => resolve()); + relay.once("error", reject); + }); + + sendRelay(relay, { t: "hello", relayVersion: "test", protocolVersion: 1 }); + sendRelay(relay, { + t: "providers", + list: [ + { + id: "fake-app", + name: "Fake App", + transport: "ws", + source: "local", + status: "connected", + capabilities: ["state", "invoke"], + }, + ], + }); + return relay; +} + +function sendRelay(relay: WebSocket, frame: Up): void { + relay.send(JSON.stringify(frame)); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > timeoutMs) { + throw new Error("Timed out waiting for condition"); + } + await Bun.sleep(10); + } +} diff --git a/packages/typescript/integrations/bridge-server/package.json b/packages/typescript/integrations/bridge-server/package.json new file mode 100644 index 00000000..bc127438 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/package.json @@ -0,0 +1,41 @@ +{ + "name": "@slop-ai/bridge-server", + "version": "0.1.0", + "type": "module", + "description": "Hosted bridge server for remote MCP hosts and local SLOP relay agents.", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "bin": { + "slop-bridge-server": "dist/cli.js" + }, + "exports": { + ".": { + "import": "./dist/index.js", + "types": "./dist/index.d.ts" + } + }, + "files": ["src", "dist", "Dockerfile", "fly.toml"], + "sideEffects": false, + "license": "MIT", + "repository": { + "type": "git", + "url": "https://github.com/devteapot/slop", + "directory": "packages/typescript/integrations/bridge-server" + }, + "scripts": { + "build": "bun build src/index.ts --outdir dist --format esm --target bun --external '@slop-ai/*' --external '@modelcontextprotocol/*' --external 'zod' && bun build src/cli.ts --outdir dist --format esm --target bun --external '@slop-ai/*' --external '@modelcontextprotocol/*' --external 'zod' && bunx tsc --emitDeclarationOnly --outDir dist", + "dev": "bun src/cli.ts", + "clean": "rm -rf dist" + }, + "dependencies": { + "@modelcontextprotocol/sdk": "^1.29.0", + "@slop-ai/consumer": "workspace:*", + "@slop-ai/relay-cli": "workspace:*", + "zod": "^4.3.6" + }, + "devDependencies": { + "@types/node": "^25.6.0", + "@types/ws": "^8.5.0", + "ws": "^8.18.0" + } +} diff --git a/packages/typescript/integrations/bridge-server/src/cli.ts b/packages/typescript/integrations/bridge-server/src/cli.ts new file mode 100644 index 00000000..9968c4b6 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/cli.ts @@ -0,0 +1,94 @@ +#!/usr/bin/env bun +import { readFileSync } from "node:fs"; +import { createBridgeServer } from "./http"; +import { readTokenRegistryFromEnv } from "./tokens"; + +const VERSION = readPackageVersion(); + +const log = { + info: (...args: unknown[]) => console.error("[slop-bridge]", ...args), + error: (...args: unknown[]) => console.error("[slop-bridge] ERROR:", ...args), +}; + +function printHelp(): void { + console.log(`slop-bridge-server ${VERSION} + +Run the hosted SLOP cloud relay bridge. + +Usage: + slop-bridge-server [--port ] [--host ] + slop-bridge-server --help + slop-bridge-server --version + +Environment: + SLOP_BRIDGE_USERS JSON map of user IDs to { mcpToken, relayToken, label? } + PORT Default port when --port is omitted +`); +} + +function readPackageVersion(): string { + try { + const json = JSON.parse(readFileSync(new URL("../package.json", import.meta.url), "utf8")) as { version?: unknown }; + return typeof json.version === "string" ? json.version : "0.1.0"; + } catch { + return "0.1.0"; + } +} + +function parseArgs(argv: string[]): { port: number; host?: string } | null { + let port = Number(process.env.PORT ?? 8080); + let host: string | undefined; + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + if (arg === "--help" || arg === "-h") { + printHelp(); + return null; + } + if (arg === "--version" || arg === "-v") { + console.log(VERSION); + return null; + } + if (arg === "--port") { + const value = argv[++index]; + if (!value) throw new Error("Missing value for --port"); + port = Number(value); + continue; + } + if (arg === "--host") { + const value = argv[++index]; + if (!value) throw new Error("Missing value for --host"); + host = value; + continue; + } + throw new Error(`Unknown argument: ${arg}`); + } + if (!Number.isInteger(port) || port <= 0) { + throw new Error(`Invalid port: ${port}`); + } + return { port, host }; +} + +async function main(): Promise { + const args = parseArgs(process.argv.slice(2)); + if (!args) return; + const tokens = readTokenRegistryFromEnv(); + const server = createBridgeServer({ + port: args.port, + hostname: args.host, + tokens, + logger: log, + }); + log.info(`listening on ${server.url}`); + + const stop = () => { + server.stop(); + process.exit(0); + }; + process.on("SIGINT", stop); + process.on("SIGTERM", stop); +} + +main().catch((error: unknown) => { + log.error("Fatal:", error instanceof Error ? error.message : String(error)); + process.exit(1); +}); diff --git a/packages/typescript/integrations/bridge-server/src/http.ts b/packages/typescript/integrations/bridge-server/src/http.ts new file mode 100644 index 00000000..5bd4edab --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/http.ts @@ -0,0 +1,232 @@ +import { randomUUID } from "node:crypto"; +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; +import { type BridgeMcpSession, createMcpSession } from "./mcp-session"; +import type { Up } from "./protocol"; +import { createRelayHub } from "./relay-hub"; +import { readBearerToken, type TokenRegistry } from "./tokens"; +import type { BridgeLogger, RelaySocketLike } from "./types"; + +export interface BridgeServerOptions { + port: number; + hostname?: string; + tokens: TokenRegistry; + logger?: BridgeLogger; + relayHub?: ReturnType; + idleTimeoutMs?: number; +} + +export interface BridgeServerHandle { + port: number; + url: string; + relayHub: ReturnType; + stop(): void; +} + +type BunServer = { + port: number; + url: URL; + stop(force?: boolean): void; + upgrade(request: Request, options?: { data?: RelaySocketData }): boolean; +}; + +type RelaySocketData = { userId: string }; + +type BunServerWebSocket = RelaySocketLike & { + data: RelaySocketData; +}; + +declare const Bun: { + serve(options: { + port: number; + hostname?: string; + fetch(request: Request, server: BunServer): Response | undefined | Promise; + websocket: { + open(socket: BunServerWebSocket): void; + message(socket: BunServerWebSocket, message: string | Buffer): void; + close(socket: BunServerWebSocket): void; + }; + }): BunServer; +}; + +const MCP_PATH = "/mcp"; +const RELAY_PATH = "/relay"; + +export function createBridgeServer(options: BridgeServerOptions): BridgeServerHandle { + const logger = options.logger ?? { info: console.error, error: console.error }; + const relayHub = options.relayHub ?? createRelayHub({ logger }); + const sessions = new Map(); + + function bearerUnauthorized(): Response { + return new Response("Unauthorized", { + status: 401, + headers: { "WWW-Authenticate": "Bearer" }, + }); + } + + function json(value: unknown, init?: ResponseInit): Response { + return new Response(JSON.stringify(value), { + ...init, + headers: { + "content-type": "application/json", + ...init?.headers, + }, + }); + } + + async function handleMcp(request: Request): Promise { + const token = readBearerToken(request); + const user = token ? options.tokens.resolveMcpToken(token) : null; + if (!user) return bearerUnauthorized(); + + const sessionId = request.headers.get("mcp-session-id"); + if (request.method === "POST") { + const session = sessionId ? sessions.get(sessionId) : undefined; + if (sessionId && (!session || session.userId !== user.userId)) { + return jsonRpcError(400, "Bad Request: No valid session ID provided"); + } + + if (!session) { + const body = await request + .clone() + .json() + .catch(() => null); + if (!isInitializeRequest(body)) { + return jsonRpcError(400, "Bad Request: No valid session ID provided"); + } + + let createdSession: BridgeMcpSession | null = null; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: (id) => { + if (createdSession) { + createdSession.id = id; + sessions.set(id, createdSession); + } + }, + onsessionclosed: async (id) => { + await closeSession(id); + }, + }); + createdSession = createMcpSession({ + id: "pending", + userId: user.userId, + transport, + relayHub, + idleTimeoutMs: options.idleTimeoutMs, + onIdle: (id) => { + void closeSession(id); + }, + }); + await createdSession.server.connect(transport); + const response = await transport.handleRequest(request); + const initializedId = transport.sessionId; + if (initializedId && createdSession) { + createdSession.id = initializedId; + sessions.set(initializedId, createdSession); + } + return response; + } + + session.touch(); + return session.transport.handleRequest(request); + } + + if (request.method === "GET" || request.method === "DELETE") { + if (!sessionId) return new Response("Invalid or missing session ID", { status: 400 }); + const session = sessions.get(sessionId); + if (!session || session.userId !== user.userId) { + return new Response("Invalid or missing session ID", { status: 400 }); + } + session.touch(); + const response = await session.transport.handleRequest(request); + if (request.method === "DELETE") { + await closeSession(sessionId); + } + return response; + } + + return new Response("Method Not Allowed", { status: 405 }); + } + + async function closeSession(sessionId: string): Promise { + const session = sessions.get(sessionId); + if (!session) return; + sessions.delete(sessionId); + await session.close(); + } + + function jsonRpcError(status: number, message: string): Response { + return json( + { + jsonrpc: "2.0", + error: { code: -32000, message }, + id: null, + }, + { status }, + ); + } + + const server = Bun.serve({ + port: options.port, + hostname: options.hostname ?? "0.0.0.0", + async fetch(request, bunServer) { + const url = new URL(request.url); + if (url.pathname === "/healthz") { + return json({ ok: true }); + } + if (url.pathname === "/debug/providers") { + const token = readBearerToken(request); + const user = token ? options.tokens.resolveMcpToken(token) : null; + if (!user) return bearerUnauthorized(); + const state = relayHub.getState(user.userId); + return json({ + userId: user.userId, + online: state?.online ?? false, + providers: state?.providers ?? [], + }); + } + if (url.pathname === MCP_PATH) { + return handleMcp(request); + } + if (url.pathname === RELAY_PATH) { + const token = readBearerToken(request); + const user = token ? options.tokens.resolveRelayToken(token) : null; + if (!user) return bearerUnauthorized(); + const upgraded = bunServer.upgrade(request, { data: { userId: user.userId } }); + if (!upgraded) return new Response("Upgrade failed", { status: 400 }); + return undefined; + } + return new Response("Not Found", { status: 404 }); + }, + websocket: { + open(socket) { + relayHub.acceptRelay(socket.data.userId, socket); + }, + message(socket, message) { + try { + const raw = typeof message === "string" ? message : message.toString(); + relayHub.handleRelayFrame(socket.data.userId, JSON.parse(raw) as Up); + } catch (error) { + logger.error("[slop-bridge] invalid relay frame", error instanceof Error ? error.message : String(error)); + socket.close(4001, "invalid relay frame"); + } + }, + close(socket) { + relayHub.closeRelay(socket.data.userId); + }, + }, + }); + + return { + port: server.port, + url: server.url.toString(), + relayHub, + stop() { + for (const sessionId of [...sessions.keys()]) { + void closeSession(sessionId); + } + server.stop(true); + }, + }; +} diff --git a/packages/typescript/integrations/bridge-server/src/index.ts b/packages/typescript/integrations/bridge-server/src/index.ts new file mode 100644 index 00000000..8622b3e2 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/index.ts @@ -0,0 +1,13 @@ +export { type BridgeServerHandle, type BridgeServerOptions, createBridgeServer } from "./http"; +export { type CreateRelayHubOptions, createRelayHub } from "./relay-hub"; +export { parseTokenRegistry, readBearerToken, readTokenRegistryFromEnv, type TokenRegistry } from "./tokens"; +export type { + ActionSummary, + BridgeLogger, + RelayHub, + RelayState, + SelectedProviderState, + StatePayload, + StateProviderSummary, + ToolResult, +} from "./types"; diff --git a/packages/typescript/integrations/bridge-server/src/mcp-session.ts b/packages/typescript/integrations/bridge-server/src/mcp-session.ts new file mode 100644 index 00000000..e58a79f6 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/mcp-session.ts @@ -0,0 +1,390 @@ +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import type { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +import { affordancesToTools, formatTree, type SlopNode, StateMirror } from "@slop-ai/consumer"; +import { z } from "zod"; +import { isPlainRecord } from "./protocol"; +import { RelayError, type RelayHub, type SelectedProviderState, type StatePayload, type ToolResult } from "./types"; + +const IDLE_TIMEOUT_MS = 30 * 60 * 1000; + +interface OpenAppArgs { + app?: string; +} + +interface AppActionArgs { + app: string; + path: string; + action: string; + params?: Record; +} + +interface AppActionBatchArgs { + app: string; + actions: { + path: string; + action: string; + params?: Record; + }[]; +} + +interface OpenedApp { + providerId: string; + providerName: string; + subId: string; + mirror: StateMirror; + stale?: RelayError; + updatedAt: number; +} + +export interface BridgeMcpSession { + id: string; + userId: string; + server: McpServer; + transport: WebStandardStreamableHTTPServerTransport; + openedApps: Map; + touch(): void; + close(): Promise; +} + +export interface CreateMcpSessionOptions { + id: string; + userId: string; + transport: WebStandardStreamableHTTPServerTransport; + relayHub: RelayHub; + onIdle: (sessionId: string) => void; + idleTimeoutMs?: number; +} + +export function createMcpSession(options: CreateMcpSessionOptions): BridgeMcpSession { + const openedApps = new Map(); + const server = new McpServer( + { name: "slop-bridge-server", version: "0.1.0" }, + { capabilities: { tools: {}, resources: {} } }, + ); + let idleTimer: ReturnType | null = null; + + const session: BridgeMcpSession = { + id: options.id, + userId: options.userId, + server, + transport: options.transport, + openedApps, + touch, + async close() { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + options.relayHub.markSessionClosed(options.userId, session.id); + await options.transport.close().catch(() => {}); + openedApps.clear(); + }, + }; + + function touch(): void { + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => { + options.onIdle(session.id); + }, options.idleTimeoutMs ?? IDLE_TIMEOUT_MS); + idleTimer.unref?.(); + } + + function statePayload(selectedApp?: string): StatePayload { + const state = options.relayHub.getState(options.userId); + const providers = (state?.providers ?? []).map((provider) => ({ + ...provider, + connected: state?.online === true && provider.status === "connected", + })); + const selected = selectedApp ? openedApps.get(selectedApp) : [...openedApps.values()].at(-1); + return { + updatedAt: Date.now(), + providers, + ...(selected && { selected: selectedState(selected) }), + }; + } + + function selectedState(app: OpenedApp): SelectedProviderState { + const tree = app.mirror.getTree(); + return { + id: app.providerId, + name: app.providerName, + tree, + formatted: formatTree(tree), + actions: describeActions(app.providerId, tree), + ...(app.stale && { stale: true }), + }; + } + + function jsonResult(payload: StatePayload): ToolResult { + return { + content: [{ type: "text", text: JSON.stringify(payload) }], + structuredContent: payload as unknown as Record, + }; + } + + function errorResult(message: string): ToolResult { + return { isError: true, content: [{ type: "text", text: message }] }; + } + + async function ensureOpenApp(appId: string): Promise { + const existing = openedApps.get(appId); + if (existing && !existing.stale) { + return existing; + } + + if (existing) { + options.relayHub.unsubscribe(options.userId, existing.subId); + openedApps.delete(appId); + } + + const { subId, snapshot } = await options.relayHub.subscribe({ + userId: options.userId, + sessionId: session.id, + providerId: appId, + onSnapshot: (frame) => { + const app = openedApps.get(appId); + if (!app) return; + app.mirror = new StateMirror({ + type: "snapshot", + id: frame.subId, + version: frame.version ?? 0, + seq: frame.seq, + tree: frame.tree, + }); + app.stale = undefined; + app.updatedAt = Date.now(); + }, + onPatch: (ops, version, seq) => { + const app = openedApps.get(appId); + if (!app) return; + app.mirror.applyPatch({ type: "patch", subscription: app.subId, version, seq, ops }); + app.updatedAt = Date.now(); + }, + onStale: (error) => { + const app = openedApps.get(appId); + if (app) app.stale = error; + }, + }); + + const providerName = + options.relayHub.getState(options.userId)?.providers.find((provider) => provider.id === appId)?.name ?? appId; + const opened: OpenedApp = { + providerId: appId, + providerName, + subId, + mirror: new StateMirror({ + type: "snapshot", + id: subId, + version: snapshot.version ?? 0, + seq: snapshot.seq, + tree: snapshot.tree, + }), + updatedAt: Date.now(), + }; + openedApps.set(appId, opened); + return opened; + } + + function summarizeState(payload: StatePayload): string { + if (payload.selected) { + const actions = payload.selected.actions + .map((action) => { + const location = action.path ?? (action.targets ? `${action.targets.length} targets` : "dynamic target"); + return `- ${action.action} on ${location}: ${action.description}`; + }) + .join("\n"); + return ( + `Opened ${payload.selected.name} (id: ${payload.selected.id}).\n\n` + + `Current state:\n\n${payload.selected.formatted}\n\n` + + `Available actions:\n${actions || "(none)"}` + ); + } + + if (payload.providers.length === 0) { + return "No SLOP relay providers are currently connected. Start slop-relay on the user's machine."; + } + + return `Available SLOP applications:\n${payload.providers + .map((provider) => `- ${provider.name} (id: ${provider.id}, ${provider.transport}, ${provider.status})`) + .join("\n")}`; + } + + server.registerTool( + "list_apps", + { + description: "List SLOP-enabled applications currently visible through the user's relay.", + inputSchema: {}, + annotations: { readOnlyHint: true, openWorldHint: false }, + }, + async () => { + touch(); + const payload = statePayload(); + return { + ...jsonResult(payload), + content: [{ type: "text", text: summarizeState(payload) }], + }; + }, + ); + + server.registerTool( + "open_app", + { + description: "Open and inspect a SLOP application by exact app id.", + inputSchema: { + app: z.string().describe("SLOP app id from list_apps."), + } as never, + annotations: { readOnlyHint: true, openWorldHint: false }, + }, + (async (args: OpenAppArgs) => { + touch(); + const app = typeof args.app === "string" ? args.app : ""; + if (!app) return errorResult("Missing app id."); + try { + await ensureOpenApp(app); + const payload = statePayload(app); + return { + ...jsonResult(payload), + content: [{ type: "text", text: summarizeState(payload) }], + }; + } catch (error) { + return errorFromUnknown(error); + } + }) as never, + ); + + server.registerTool( + "slop_get_state", + { + description: "Return the latest cached SLOP state for an opened app.", + inputSchema: { + app: z.string().optional().describe("SLOP app id. Defaults to the most recently opened app."), + } as never, + annotations: { readOnlyHint: true, openWorldHint: false }, + _meta: { ui: { visibility: ["app"] } }, + }, + (async (args: OpenAppArgs) => { + touch(); + const appId = typeof args.app === "string" ? args.app : [...openedApps.keys()].at(-1); + if (appId) { + const app = openedApps.get(appId); + if (app?.stale) { + try { + await ensureOpenApp(appId); + } catch {} + } + } + return jsonResult(statePayload(appId)); + }) as never, + ); + + server.registerTool( + "app_action", + { + description: "Perform a single SLOP affordance. Use action coordinates shown by open_app or slop_get_state.", + inputSchema: { + app: z.string().describe("SLOP app id."), + path: z.string().describe("Path to the target node."), + action: z.string().describe("Affordance/action name."), + params: z.record(z.string(), z.unknown()).optional().describe("Optional action parameters."), + } as never, + annotations: { readOnlyHint: false, openWorldHint: true }, + }, + (async (args: AppActionArgs) => { + touch(); + if (!isPlainRecord(args.params ?? {})) return errorResult("params must be an object when provided."); + const result = await invokeAction(args.app, args.path, args.action, args.params ?? {}); + return result; + }) as never, + ); + + server.registerTool( + "app_action_batch", + { + description: "Perform multiple SLOP affordances sequentially.", + inputSchema: { + app: z.string().describe("SLOP app id."), + actions: z.array( + z.object({ + path: z.string(), + action: z.string(), + params: z.record(z.string(), z.unknown()).optional(), + }), + ), + } as never, + annotations: { readOnlyHint: false, openWorldHint: true }, + }, + (async (args: AppActionBatchArgs) => { + touch(); + let failed = 0; + const lines: string[] = []; + for (const action of args.actions ?? []) { + const result = await invokeAction(args.app, action.path, action.action, action.params ?? {}); + if (result.isError) { + failed += 1; + lines.push(`FAIL: ${action.action} on ${action.path} - ${result.content[0]?.text ?? "unknown error"}`); + } else { + lines.push(`OK: ${action.action} on ${action.path}`); + } + } + return { isError: failed > 0, content: [{ type: "text", text: lines.join("\n") || "(no actions)" }] }; + }) as never, + ); + + async function invokeAction( + app: string, + path: string, + action: string, + params: Record, + ): Promise { + try { + const frame = await options.relayHub.invoke({ + userId: options.userId, + providerId: app, + path, + action, + params, + }); + if (!frame.ok) { + return errorResult(`[${frame.error.code}] ${frame.error.message}`); + } + return { + content: [ + { + type: "text", + text: `Done.${frame.data !== undefined ? ` Result: ${JSON.stringify(frame.data)}` : ""}`, + }, + ], + }; + } catch (error) { + return errorFromUnknown(error); + } + } + + function errorFromUnknown(error: unknown): ToolResult { + if (error instanceof RelayError) { + return errorResult(`[${error.code}] ${error.message}`); + } + return errorResult(error instanceof Error ? error.message : String(error)); + } + + touch(); + return session; +} + +function describeActions(providerId: string, tree: SlopNode) { + const toolSet = affordancesToTools(tree); + return toolSet.tools.flatMap((tool) => { + const resolved = toolSet.resolve(tool.function.name); + if (!resolved) return []; + return [ + { + name: `${providerId.replace(/[^a-zA-Z0-9]/g, "_")}__${tool.function.name}`, + description: tool.function.description, + path: resolved.path, + action: resolved.action, + targets: resolved.targets, + parameters: tool.function.parameters, + }, + ]; + }); +} diff --git a/packages/typescript/integrations/bridge-server/src/protocol.ts b/packages/typescript/integrations/bridge-server/src/protocol.ts new file mode 100644 index 00000000..2090ca80 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/protocol.ts @@ -0,0 +1,10 @@ +export { + DEFAULT_RELAY_URL, + type Down, + isDownFrame, + isPlainRecord, + type Logger, + type ProviderSummary, + RELAY_PROTOCOL_VERSION, + type Up, +} from "@slop-ai/relay-cli/protocol"; diff --git a/packages/typescript/integrations/bridge-server/src/relay-hub.ts b/packages/typescript/integrations/bridge-server/src/relay-hub.ts new file mode 100644 index 00000000..f0a8ffeb --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/relay-hub.ts @@ -0,0 +1,271 @@ +import { randomUUID } from "node:crypto"; +import { type Down, RELAY_PROTOCOL_VERSION, type Up } from "./protocol"; +import { type BridgeLogger, RelayError, type RelayHub, type RelaySocketLike, type RelayState } from "./types"; + +export interface CreateRelayHubOptions { + logger?: BridgeLogger; + requestTimeoutMs?: number; + openReadyState?: number; +} + +const DEFAULT_REQUEST_TIMEOUT_MS = 15_000; +const DEFAULT_OPEN_READY_STATE = 1; + +export function createRelayHub(options: CreateRelayHubOptions = {}): RelayHub { + const logger = options.logger ?? { info: () => {}, error: () => {} }; + const requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; + const openReadyState = options.openReadyState ?? DEFAULT_OPEN_READY_STATE; + const states = new Map(); + + function getOrCreateState(userId: string): RelayState { + const existing = states.get(userId); + if (existing) return existing; + const state: RelayState = { + userId, + socket: null, + providers: [], + pending: new Map(), + subs: new Map(), + online: false, + }; + states.set(userId, state); + return state; + } + + function sendFrame(state: RelayState, frame: Down): void { + if (!state.socket || state.socket.readyState !== openReadyState) { + throw new RelayError("relay_disconnected", "Relay is not connected."); + } + state.socket.send(JSON.stringify(frame)); + } + + function request( + state: RelayState, + frame: Extract, + timeoutMs = requestTimeoutMs, + ): Promise { + if (state.pending.has(frame.reqId)) { + throw new RelayError("duplicate_req_id", `Duplicate relay request id ${frame.reqId}.`); + } + + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + state.pending.delete(frame.reqId); + reject(new RelayError("relay_timeout", `Timed out waiting for relay response ${frame.reqId}.`)); + }, timeoutMs); + timer.unref?.(); + state.pending.set(frame.reqId, { + resolve, + reject, + timer, + }); + try { + sendFrame(state, frame); + } catch (error) { + clearTimeout(timer); + state.pending.delete(frame.reqId); + reject(error); + } + }); + } + + function rejectPending(state: RelayState, error: RelayError): void { + for (const [reqId, pending] of state.pending) { + clearTimeout(pending.timer); + pending.reject(error); + state.pending.delete(reqId); + } + } + + function markSubscriptionsStale(state: RelayState, error: RelayError): void { + for (const subscription of state.subs.values()) { + subscription.onStale(error); + } + } + + function closeSocket(socket: RelaySocketLike | null, code = 1000, reason = "closing"): void { + if (!socket) return; + try { + socket.close(code, reason); + } catch {} + } + + const hub: RelayHub = { + getState(userId) { + return states.get(userId) ?? null; + }, + + getOrCreateState, + + acceptRelay(userId, socket) { + const state = getOrCreateState(userId); + if (state.socket && state.socket !== socket) { + rejectPending(state, new RelayError("relay_replaced", "Relay was replaced by a newer connection.")); + markSubscriptionsStale(state, new RelayError("relay_replaced", "Relay was replaced by a newer connection.")); + closeSocket(state.socket, 4000, "relay replaced"); + } + state.socket = socket; + state.online = true; + state.providers = []; + state.relayVersion = undefined; + logger.info("[slop-bridge] relay connected", { userId }); + }, + + handleRelayFrame(userId, frame) { + const state = getOrCreateState(userId); + switch (frame.t) { + case "hello": { + if (frame.protocolVersion !== RELAY_PROTOCOL_VERSION) { + closeSocket(state.socket, 4002, "unsupported relay protocol"); + state.socket = null; + state.online = false; + throw new RelayError("unsupported_protocol", `Unsupported relay protocol ${frame.protocolVersion}.`); + } + state.relayVersion = frame.relayVersion; + logger.info("[slop-bridge] relay hello", { userId, relayVersion: frame.relayVersion }); + for (const subscription of state.subs.values()) { + const reqId = `req-${randomUUID()}`; + void request(state, { + t: "subscribe", + reqId, + subId: subscription.subId, + providerId: subscription.providerId, + }).catch((error) => { + subscription.onStale( + error instanceof RelayError + ? error + : new RelayError("resubscribe_failed", error instanceof Error ? error.message : String(error)), + ); + }); + } + break; + } + case "providers": { + state.providers = frame.list; + break; + } + case "snapshot": { + const pending = state.pending.get(frame.reqId); + if (pending) { + clearTimeout(pending.timer); + state.pending.delete(frame.reqId); + pending.resolve(frame); + } + state.subs.get(frame.subId)?.onSnapshot(frame); + break; + } + case "patch": { + state.subs.get(frame.subId)?.onPatch(frame.ops, frame.version, frame.seq); + break; + } + case "result": { + const pending = state.pending.get(frame.reqId); + if (!pending) return; + clearTimeout(pending.timer); + state.pending.delete(frame.reqId); + pending.resolve(frame); + break; + } + } + }, + + closeRelay(userId) { + const state = getOrCreateState(userId); + state.socket = null; + state.online = false; + rejectPending(state, new RelayError("relay_disconnected", "Relay disconnected.")); + markSubscriptionsStale(state, new RelayError("relay_disconnected", "Relay disconnected.")); + logger.info("[slop-bridge] relay disconnected", { userId }); + }, + + async subscribe(input) { + const state = getOrCreateState(input.userId); + const reqId = `req-${randomUUID()}`; + const subId = `sub-${randomUUID()}`; + state.subs.set(subId, { + userId: input.userId, + sessionId: input.sessionId, + providerId: input.providerId, + subId, + onSnapshot: input.onSnapshot, + onPatch: input.onPatch, + onStale: input.onStale, + }); + + try { + const frame = await request(state, { + t: "subscribe", + reqId, + subId, + providerId: input.providerId, + }); + if (frame.t === "result" && !frame.ok) { + throw new RelayError(frame.error.code, frame.error.message); + } + if (frame.t !== "snapshot") { + throw new RelayError("unexpected_relay_frame", `Expected snapshot, got ${frame.t}.`); + } + return { subId, snapshot: frame }; + } catch (error) { + state.subs.delete(subId); + throw error; + } + }, + + unsubscribe(userId, subId) { + const state = getOrCreateState(userId); + state.subs.delete(subId); + if (state.socket && state.socket.readyState === openReadyState) { + sendFrame(state, { t: "unsubscribe", subId }); + } + }, + + async invoke(input) { + const state = getOrCreateState(input.userId); + const reqId = `req-${randomUUID()}`; + const frame = await request(state, { + t: "invoke", + reqId, + providerId: input.providerId, + path: input.path, + action: input.action, + params: input.params, + }); + if (frame.t !== "result") { + throw new RelayError("unexpected_relay_frame", `Expected result, got ${frame.t}.`); + } + return frame; + }, + + markSessionClosed(userId, sessionId) { + const state = getOrCreateState(userId); + for (const [subId, subscription] of [...state.subs]) { + if (subscription.sessionId === sessionId) { + hub.unsubscribe(userId, subId); + } + } + }, + + resubscribeSession(userId, sessionId) { + const state = getOrCreateState(userId); + for (const subscription of [...state.subs.values()]) { + if (subscription.sessionId !== sessionId) continue; + const reqId = `req-${randomUUID()}`; + void request(state, { + t: "subscribe", + reqId, + subId: subscription.subId, + providerId: subscription.providerId, + }).catch((error) => { + subscription.onStale( + error instanceof RelayError + ? error + : new RelayError("resubscribe_failed", error instanceof Error ? error.message : String(error)), + ); + }); + } + }, + }; + + return hub; +} diff --git a/packages/typescript/integrations/bridge-server/src/tokens.ts b/packages/typescript/integrations/bridge-server/src/tokens.ts new file mode 100644 index 00000000..594f9039 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/tokens.ts @@ -0,0 +1,106 @@ +export interface UserTokenRecord { + userId: string; + label?: string; + mcpToken: string; + relayToken: string; +} + +export interface TokenRegistry { + users: UserTokenRecord[]; + resolveMcpToken(token: string): UserTokenRecord | null; + resolveRelayToken(token: string): UserTokenRecord | null; +} + +type EnvUserRecord = { + label?: unknown; + mcpToken?: unknown; + relayToken?: unknown; +}; + +const TOKEN_MIN_LENGTH = 16; + +export function readTokenRegistryFromEnv(env = process.env): TokenRegistry { + const raw = env.SLOP_BRIDGE_USERS; + if (!raw) { + throw new Error("Missing SLOP_BRIDGE_USERS. Expected JSON mapping user IDs to mcpToken and relayToken."); + } + return parseTokenRegistry(raw); +} + +export function parseTokenRegistry(raw: string): TokenRegistry { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (error) { + throw new Error(`SLOP_BRIDGE_USERS is not valid JSON: ${error instanceof Error ? error.message : String(error)}`); + } + + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new Error("SLOP_BRIDGE_USERS must be a JSON object keyed by user id."); + } + + const usedTokens = new Set(); + const users: UserTokenRecord[] = []; + + for (const [userId, value] of Object.entries(parsed as Record)) { + if (!isValidUserId(userId)) { + throw new Error(`Invalid bridge user id "${userId}".`); + } + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new Error(`Bridge user "${userId}" must be an object.`); + } + + const mcpToken = readToken(value.mcpToken, `mcpToken for "${userId}"`); + const relayToken = readToken(value.relayToken, `relayToken for "${userId}"`); + if (mcpToken === relayToken) { + throw new Error(`Bridge user "${userId}" must use different MCP and relay tokens.`); + } + for (const token of [mcpToken, relayToken]) { + if (usedTokens.has(token)) { + throw new Error("SLOP_BRIDGE_USERS contains duplicate tokens."); + } + usedTokens.add(token); + } + + users.push({ + userId, + label: typeof value.label === "string" ? value.label : undefined, + mcpToken, + relayToken, + }); + } + + if (users.length === 0) { + throw new Error("SLOP_BRIDGE_USERS must contain at least one user."); + } + + const mcpByToken = new Map(users.map((user) => [user.mcpToken, user])); + const relayByToken = new Map(users.map((user) => [user.relayToken, user])); + return { + users, + resolveMcpToken(token: string) { + return mcpByToken.get(token) ?? null; + }, + resolveRelayToken(token: string) { + return relayByToken.get(token) ?? null; + }, + }; +} + +export function readBearerToken(request: Request): string | null { + const header = request.headers.get("authorization"); + if (!header) return null; + const match = /^Bearer\s+(.+)$/i.exec(header.trim()); + return match?.[1] ?? null; +} + +function readToken(value: unknown, label: string): string { + if (typeof value !== "string" || value.length < TOKEN_MIN_LENGTH) { + throw new Error(`Invalid ${label}; expected a string with at least ${TOKEN_MIN_LENGTH} characters.`); + } + return value; +} + +function isValidUserId(value: string): boolean { + return /^[a-zA-Z0-9][a-zA-Z0-9._-]{0,127}$/.test(value); +} diff --git a/packages/typescript/integrations/bridge-server/src/types.ts b/packages/typescript/integrations/bridge-server/src/types.ts new file mode 100644 index 00000000..8ec10f13 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/src/types.ts @@ -0,0 +1,121 @@ +import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js"; +import type { PatchOp, SlopNode } from "@slop-ai/consumer"; +import type { Down, ProviderSummary, Up } from "./protocol"; + +export type ToolContent = { type: "text"; text: string }; +export type ToolResult = CallToolResult & { + content: ToolContent[]; + structuredContent?: Record; +}; + +export interface ActionSummary { + name: string; + description: string; + path: string | null; + action: string; + targets?: string[]; + parameters: Record; +} + +export interface StateProviderSummary extends ProviderSummary { + connected: boolean; +} + +export interface SelectedProviderState { + id: string; + name: string; + tree: SlopNode; + formatted: string; + actions: ActionSummary[]; + stale?: boolean; +} + +export interface StatePayload { + updatedAt: number; + providers: StateProviderSummary[]; + selected?: SelectedProviderState; +} + +export type PendingRequest = { + resolve: (frame: Up) => void; + reject: (error: RelayError) => void; + timer: ReturnType; +}; + +export class RelayError extends Error { + constructor( + public readonly code: string, + message: string, + ) { + super(message); + this.name = "RelayError"; + } +} + +export interface RelaySubscription { + userId: string; + sessionId: string; + providerId: string; + subId: string; + onSnapshot: (frame: Extract) => void; + onPatch: (ops: PatchOp[], version: number, seq?: number) => void; + onStale: (error: RelayError) => void; +} + +export interface RelaySocketLike { + readyState: number; + send(data: string): void; + close(code?: number, reason?: string): void; +} + +export interface RelayState { + userId: string; + socket: RelaySocketLike | null; + providers: ProviderSummary[]; + pending: Map; + subs: Map; + online: boolean; + relayVersion?: string; +} + +export type RelaySendOptions = { + expectReply?: boolean; + timeoutMs?: number; +}; + +export interface RelayHub { + getState(userId: string): RelayState | null; + getOrCreateState(userId: string): RelayState; + acceptRelay(userId: string, socket: RelaySocketLike): void; + handleRelayFrame(userId: string, frame: Up): void; + closeRelay(userId: string): void; + subscribe(input: { + userId: string; + sessionId: string; + providerId: string; + onSnapshot: RelaySubscription["onSnapshot"]; + onPatch: RelaySubscription["onPatch"]; + onStale: RelaySubscription["onStale"]; + }): Promise<{ subId: string; snapshot: Extract }>; + unsubscribe(userId: string, subId: string): void; + invoke(input: { + userId: string; + providerId: string; + path: string; + action: string; + params?: Record; + }): Promise>; + markSessionClosed(userId: string, sessionId: string): void; + resubscribeSession(userId: string, sessionId: string): void; +} + +export type BridgeLogger = { + info: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; +}; + +export type WebSocketReadyState = { + OPEN: number; +}; + +export type RelayFrame = Down | Up; diff --git a/packages/typescript/integrations/bridge-server/tsconfig.json b/packages/typescript/integrations/bridge-server/tsconfig.json new file mode 100644 index 00000000..9b6df668 --- /dev/null +++ b/packages/typescript/integrations/bridge-server/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ES2022", + "moduleResolution": "bundler", + "strict": true, + "declaration": true, + "lib": ["ES2022", "DOM"], + "types": ["node"], + "outDir": "dist", + "rootDir": "src", + "skipLibCheck": true + }, + "include": ["src"] +} diff --git a/packages/typescript/integrations/relay-cli/README.md b/packages/typescript/integrations/relay-cli/README.md new file mode 100644 index 00000000..02e7fd55 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/README.md @@ -0,0 +1,33 @@ +# @slop-ai/relay-cli + +Outbound relay agent for SLOP providers. + +`slop-relay` runs on a user's machine, discovers local SLOP apps with +`@slop-ai/discovery`, and connects outbound to a hosted bridge over WebSocket. +The hosted bridge is responsible for exposing those providers to remote MCP +hosts. + +## Usage + +```sh +slop-relay --token "$SLOP_RELAY_TOKEN" +slop-relay --url ws://localhost:9999 --token dev --verbose +``` + +Configuration precedence: + +1. `--url`, then `SLOP_RELAY_URL`, then `wss://bridge.slopai.dev/relay` +2. `--token`, then `SLOP_RELAY_TOKEN`, then `~/.slop/relay.json` + +`slop-relay login` is a placeholder in v1. Real OAuth login lands with the +hosted bridge. + +## Development + +```sh +bun install +cd packages/typescript/integrations/relay-cli +bun run dev -- --url ws://localhost:9999 --token dev --verbose +bun test +bun run build +``` diff --git a/packages/typescript/integrations/relay-cli/__tests__/mock-bridge.ts b/packages/typescript/integrations/relay-cli/__tests__/mock-bridge.ts new file mode 100644 index 00000000..5c6b71af --- /dev/null +++ b/packages/typescript/integrations/relay-cli/__tests__/mock-bridge.ts @@ -0,0 +1,51 @@ +import WebSocket, { WebSocketServer } from "ws"; +import type { Down, Up } from "../src/protocol"; + +export class MockRelayBridge { + private server: WebSocketServer | null = null; + private socket: WebSocket | null = null; + readonly frames: Up[] = []; + + constructor(private readonly port: number) {} + + get url(): string { + return `ws://127.0.0.1:${this.port}/relay`; + } + + async start(): Promise { + const server = new WebSocketServer({ host: "127.0.0.1", port: this.port, path: "/relay" }); + this.server = server; + server.on("connection", (socket) => { + this.socket = socket; + socket.on("message", (raw) => { + this.frames.push(JSON.parse(raw.toString()) as Up); + }); + }); + await new Promise((resolve) => server.once("listening", resolve)); + } + + send(frame: Down): void { + if (!this.socket || this.socket.readyState !== WebSocket.OPEN) { + throw new Error("Mock bridge has no open socket"); + } + this.socket.send(JSON.stringify(frame)); + } + + async close(): Promise { + for (const client of this.server?.clients ?? []) { + client.terminate(); + } + if (!this.server) return; + const server = this.server; + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, 100); + server.close((error) => { + clearTimeout(timer); + if (error) reject(error); + else resolve(); + }); + }); + this.server = null; + this.socket = null; + } +} diff --git a/packages/typescript/integrations/relay-cli/__tests__/relay.test.ts b/packages/typescript/integrations/relay-cli/__tests__/relay.test.ts new file mode 100644 index 00000000..9fe1279e --- /dev/null +++ b/packages/typescript/integrations/relay-cli/__tests__/relay.test.ts @@ -0,0 +1,298 @@ +import { describe, expect, test } from "bun:test"; +import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import WebSocket, { WebSocketServer } from "ws"; +import { createRelayBridge } from "../src/bridge"; +import type { Up } from "../src/protocol"; +import { createRelayClient } from "../src/relay-client"; +import { MockRelayBridge } from "./mock-bridge"; + +describe("relay bridge", () => { + test("sends hello and providers, then handles subscribe, patch, invoke, and unsubscribe", async () => { + const providersDir = createTempDir("slop-relay-providers"); + const bridgePort = await getFreePort(); + const providerPort = await getFreePort(); + const mockBridge = new MockRelayBridge(bridgePort); + const providerServer = await createMockSlopProvider({ port: providerPort }); + let relay: ReturnType | null = null; + + try { + writeDescriptor(providersDir, "fake-provider.json", { + id: "fake-provider", + name: "Fake Provider", + slop_version: "0.1", + transport: { type: "ws", url: providerServer.url }, + capabilities: ["state"], + }); + await mockBridge.start(); + + relay = createRelayBridge({ + url: mockBridge.url, + token: "dev", + relayVersion: "0.1.0", + discoveryOptions: { + providersDirs: [providersDir], + enableBridge: false, + watchProviders: false, + hostBridge: false, + scanIntervalMs: 50, + bridgeDialTimeoutMs: 20, + bridgeRetryDelayMs: 20, + }, + reconnectBaseDelayMs: 20, + reconnectMaxDelayMs: 40, + }); + relay.start(); + + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "hello")); + await waitUntil(() => + mockBridge.frames.some( + (frame) => frame.t === "providers" && frame.list.some((provider) => provider.id === "fake-provider"), + ), + ); + + mockBridge.send({ t: "subscribe", reqId: "req-sub", subId: "bridge-sub", providerId: "fake-provider" }); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "snapshot" && frame.subId === "bridge-sub")); + const snapshot = mockBridge.frames.find( + (frame): frame is Extract => frame.t === "snapshot", + ); + expect(snapshot?.tree.id).toBe("root"); + + providerServer.patch(); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "patch" && frame.subId === "bridge-sub")); + const patch = mockBridge.frames.find((frame): frame is Extract => frame.t === "patch"); + expect(patch?.ops[0]?.path).toBe("/properties/count"); + expect(patch?.version).toBe(2); + + mockBridge.send({ + t: "invoke", + reqId: "req-invoke", + providerId: "fake-provider", + path: "/", + action: "increment", + params: { by: 1 }, + }); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "result" && frame.reqId === "req-invoke")); + const result = mockBridge.frames.find( + (frame): frame is Extract => + frame.t === "result" && frame.reqId === "req-invoke", + ); + expect(result?.ok).toBe(true); + expect(result?.data).toEqual({ invoked: true }); + + mockBridge.send({ t: "unsubscribe", subId: "bridge-sub" }); + await delay(20); + const patchCount = mockBridge.frames.filter((frame) => frame.t === "patch").length; + providerServer.patch(); + await delay(50); + expect(mockBridge.frames.filter((frame) => frame.t === "patch")).toHaveLength(patchCount); + } finally { + relay?.stop(); + await providerServer.close(); + await mockBridge.close(); + removeTempDir(providersDir); + } + }); + + test("rejects non-object invoke params", async () => { + const bridgePort = await getFreePort(); + const mockBridge = new MockRelayBridge(bridgePort); + await mockBridge.start(); + const relay = createRelayBridge({ + url: mockBridge.url, + token: "dev", + relayVersion: "0.1.0", + discoveryOptions: { enableBridge: false, watchProviders: false, hostBridge: false }, + reconnectBaseDelayMs: 20, + reconnectMaxDelayMs: 40, + }); + + try { + relay.start(); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "hello")); + mockBridge.send({ + t: "invoke", + reqId: "bad", + providerId: "fake-provider", + path: "/", + action: "increment", + params: [] as never, + }); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "result" && frame.reqId === "bad")); + const result = mockBridge.frames.find( + (frame): frame is Extract => frame.t === "result" && frame.reqId === "bad", + ); + expect(result?.ok).toBe(false); + expect(result?.error.code).toBe("invalid_params"); + } finally { + relay.stop(); + await mockBridge.close(); + } + }); + + test("relay client retains only providers while disconnected", async () => { + const bridgePort = await getFreePort(); + const mockBridge = new MockRelayBridge(bridgePort); + const client = createRelayClient({ + url: mockBridge.url, + token: "dev", + relayVersion: "0.1.0", + reconnectBaseDelayMs: 20, + reconnectMaxDelayMs: 40, + }); + + client.send({ + t: "providers", + list: [ + { + id: "fake-provider", + name: "Fake Provider", + transport: "ws", + source: "local", + status: "disconnected", + }, + ], + }); + client.send({ t: "result", reqId: "stale-result", ok: true }); + client.send({ + t: "snapshot", + reqId: "stale-snapshot", + subId: "stale-sub", + tree: { id: "root", type: "app" }, + }); + client.send({ t: "patch", subId: "stale-sub", ops: [], version: 1 }); + + try { + await mockBridge.start(); + client.start(); + await waitUntil(() => mockBridge.frames.some((frame) => frame.t === "providers")); + + expect(mockBridge.frames.map((frame) => frame.t)).toEqual(["hello", "providers"]); + } finally { + client.stop(); + await mockBridge.close(); + } + }); +}); + +function createTempDir(prefix: string): string { + const dir = mkdtempSync(join(tmpdir(), `${prefix}-`)); + if (process.platform !== "win32") chmodSync(dir, 0o700); + return dir; +} + +function removeTempDir(path: string): void { + rmSync(path, { recursive: true, force: true }); +} + +function writeDescriptor(dir: string, fileName: string, descriptor: unknown): void { + const path = join(dir, fileName); + writeFileSync(path, JSON.stringify(descriptor, null, 2), { mode: 0o600 }); + if (process.platform !== "win32") chmodSync(path, 0o600); +} + +async function createMockSlopProvider(options: { port: number }) { + const clients = new Set(); + let subscriptionId: string | null = null; + let count = 0; + const server = new WebSocketServer({ host: "127.0.0.1", port: options.port, path: "/slop" }); + + server.on("connection", (socket) => { + clients.add(socket); + socket.send( + JSON.stringify({ + type: "hello", + provider: { id: "fake-provider", name: "Fake Provider", slop_version: "0.1", capabilities: ["state"] }, + }), + ); + socket.on("message", (raw) => { + const message = JSON.parse(raw.toString()) as Record; + if (message.type === "subscribe" && typeof message.id === "string") { + subscriptionId = message.id; + socket.send( + JSON.stringify({ + type: "snapshot", + id: message.id, + version: 1, + tree: { + id: "root", + type: "app", + properties: { count }, + affordances: [{ action: "increment", label: "Increment" }], + children: [], + }, + }), + ); + } + if (message.type === "invoke" && typeof message.id === "string") { + socket.send(JSON.stringify({ type: "result", id: message.id, status: "ok", data: { invoked: true } })); + } + }); + socket.on("close", () => clients.delete(socket)); + }); + + await new Promise((resolve) => server.once("listening", resolve)); + + return { + url: `ws://127.0.0.1:${options.port}/slop`, + patch() { + count += 1; + if (!subscriptionId) return; + for (const client of clients) { + if (client.readyState === WebSocket.OPEN) { + client.send( + JSON.stringify({ + type: "patch", + subscription: subscriptionId, + version: count + 1, + ops: [{ op: "replace", path: "/properties/count", value: count }], + }), + ); + } + } + }, + async close() { + for (const client of clients) client.terminate(); + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, 100); + server.close((error) => { + clearTimeout(timer); + if (error) reject(error); + else resolve(); + }); + }); + }, + }; +} + +async function getFreePort(): Promise { + const net = await import("node:net"); + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (!address || typeof address === "string") { + reject(new Error("Failed to allocate port")); + return; + } + server.close((error) => { + if (error) reject(error); + else resolve(address.port); + }); + }); + }); +} + +async function waitUntil(fn: () => boolean | Promise, timeoutMs = 1000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await fn()) return; + await delay(10); + } + throw new Error("Condition not met before timeout"); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/packages/typescript/integrations/relay-cli/package.json b/packages/typescript/integrations/relay-cli/package.json new file mode 100644 index 00000000..91be7bab --- /dev/null +++ b/packages/typescript/integrations/relay-cli/package.json @@ -0,0 +1,43 @@ +{ + "name": "@slop-ai/relay-cli", + "version": "0.1.0", + "type": "module", + "description": "Outbound relay agent that exposes local SLOP providers to a hosted bridge.", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "bin": { + "slop-relay": "dist/cli.js" + }, + "exports": { + ".": { + "import": "./dist/index.js", + "types": "./dist/index.d.ts" + }, + "./protocol": { + "import": "./dist/protocol.js", + "types": "./dist/protocol.d.ts" + } + }, + "files": ["src", "dist"], + "sideEffects": false, + "license": "MIT", + "repository": { + "type": "git", + "url": "https://github.com/devteapot/slop", + "directory": "packages/typescript/integrations/relay-cli" + }, + "scripts": { + "build": "bun build src/index.ts src/protocol.ts --outdir dist --format esm --target node --external '@slop-ai/*' --external 'ws' && bun build src/cli.ts --outdir dist --format esm --target node --external '@slop-ai/*' --external 'ws' && bunx tsc --emitDeclarationOnly --outDir dist", + "dev": "bun src/cli.ts", + "clean": "rm -rf dist" + }, + "dependencies": { + "@slop-ai/consumer": "workspace:*", + "@slop-ai/discovery": "workspace:*", + "ws": "^8.18.0" + }, + "devDependencies": { + "@types/node": "^25.6.0", + "@types/ws": "^8.5.0" + } +} diff --git a/packages/typescript/integrations/relay-cli/src/bridge.ts b/packages/typescript/integrations/relay-cli/src/bridge.ts new file mode 100644 index 00000000..bd0280b9 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/src/bridge.ts @@ -0,0 +1,225 @@ +import type { PatchOp, ResultMessage } from "@slop-ai/consumer"; +import { + type ConnectedProvider, + createDiscoveryService, + type DiscoveryOptions, + type DiscoveryService, + type ProviderDescriptor, +} from "@slop-ai/discovery/service"; +import { type Down, isPlainRecord, type Logger, type ProviderSummary, type Up } from "./protocol"; +import { createRelayClient, type RelayClient, type RelayClientOptions } from "./relay-client"; + +export interface RelayBridgeOptions { + url: string; + token: string; + relayVersion: string; + discoveryOptions?: DiscoveryOptions; + discovery?: DiscoveryService; + client?: RelayClient; + logger?: Logger; + reconnectBaseDelayMs?: number; + reconnectMaxDelayMs?: number; + stableConnectionMs?: number; +} + +export interface RelayBridge { + start(): void; + stop(): void; +} + +type PatchListener = (subscriptionId: string, ops: PatchOp[], version: number) => void; + +interface ActiveSubscription { + provider: ConnectedProvider; + listener: PatchListener; +} + +const noopLogger: Logger = { info: () => {}, error: () => {} }; + +export function createRelayBridge(options: RelayBridgeOptions): RelayBridge { + const logger = options.logger ?? noopLogger; + const discovery = + options.discovery ?? + createDiscoveryService({ + autoConnect: false, + watchProviders: true, + ...options.discoveryOptions, + logger, + }); + const client = + options.client ?? + createRelayClient({ + url: options.url, + token: options.token, + relayVersion: options.relayVersion, + logger, + reconnectBaseDelayMs: options.reconnectBaseDelayMs, + reconnectMaxDelayMs: options.reconnectMaxDelayMs, + stableConnectionMs: options.stableConnectionMs, + } satisfies RelayClientOptions); + const subscriptions = new Map(); + + const sendProviders = () => { + client.send({ t: "providers", list: summarizeProviders(discovery) }); + }; + + const onFrame = (frame: Down) => { + void handleFrame(frame).catch((error) => { + logger.error("[slop-relay] Bridge frame failed:", error instanceof Error ? error.message : String(error)); + }); + }; + + async function handleFrame(frame: Down): Promise { + switch (frame.t) { + case "subscribe": + await handleSubscribe(frame); + break; + case "unsubscribe": + handleUnsubscribe(frame.subId); + break; + case "invoke": + await handleInvoke(frame); + break; + case "ping": + break; + } + } + + async function handleSubscribe(frame: Extract): Promise { + handleUnsubscribe(frame.subId); + const provider = await discovery.ensureConnected(frame.providerId); + if (!provider) { + sendError(frame.reqId, "provider_unavailable", `Provider "${frame.providerId}" is not available.`); + return; + } + + const tree = provider.consumer.getTree(provider.subscriptionId); + if (!tree) { + sendError(frame.reqId, "snapshot_unavailable", "Provider has not delivered an initial snapshot yet."); + return; + } + + client.send({ t: "snapshot", reqId: frame.reqId, subId: frame.subId, tree }); + + const listener: PatchListener = (subscriptionId, ops, version) => { + if (subscriptionId !== provider.subscriptionId) return; + client.send({ t: "patch", subId: frame.subId, ops, version }); + }; + provider.consumer.on("patch", listener); + subscriptions.set(frame.subId, { provider, listener }); + } + + function handleUnsubscribe(subId: string): void { + const active = subscriptions.get(subId); + if (!active) return; + active.provider.consumer.off("patch", active.listener); + subscriptions.delete(subId); + } + + async function handleInvoke(frame: Extract): Promise { + if (frame.params !== undefined && !isPlainRecord(frame.params)) { + sendError(frame.reqId, "invalid_params", "Invoke params must be an object when provided."); + return; + } + + const provider = await discovery.ensureConnected(frame.providerId); + if (!provider) { + sendError(frame.reqId, "provider_unavailable", `Provider "${frame.providerId}" is not available.`); + return; + } + + try { + const result = await provider.consumer.invoke(frame.path, frame.action, frame.params ?? {}); + client.send(resultToFrame(frame.reqId, result)); + } catch (error) { + sendError(frame.reqId, "invoke_failed", error instanceof Error ? error.message : String(error)); + } + } + + function sendError(reqId: string, code: string, message: string): void { + client.send({ t: "result", reqId, ok: false, error: { code, message } }); + } + + return { + start() { + discovery.onStateChange(sendProviders); + client.on("frame", onFrame); + client.on("open", sendProviders); + discovery.start(); + sendProviders(); + client.start(); + }, + stop() { + for (const subId of [...subscriptions.keys()]) { + handleUnsubscribe(subId); + } + client.off("frame", onFrame); + client.off("open", sendProviders); + client.stop(); + discovery.stop(); + }, + }; +} + +export function summarizeProviders( + discovery: Pick, +): ProviderSummary[] { + const connected = discovery.getProviders(); + const connectedById = new Map(connected.map((provider) => [provider.id, provider])); + const summaries: ProviderSummary[] = []; + const seen = new Set(); + + for (const descriptor of discovery.getDiscovered()) { + const provider = connectedById.get(descriptor.id); + summaries.push(summaryFromDescriptor(descriptor, provider)); + seen.add(descriptor.id); + } + + for (const provider of connected) { + if (seen.has(provider.id)) continue; + summaries.push({ + id: provider.id, + name: provider.name, + transport: provider.descriptor.transport.type, + source: provider.descriptor.source ?? "local", + status: provider.status, + capabilities: provider.descriptor.capabilities, + }); + } + + return summaries; +} + +function summaryFromDescriptor( + descriptor: ProviderDescriptor, + provider: ConnectedProvider | undefined, +): ProviderSummary { + return { + id: descriptor.id, + name: provider?.name ?? descriptor.name, + transport: descriptor.transport.type, + source: descriptor.source ?? "local", + status: provider?.status ?? "disconnected", + capabilities: descriptor.capabilities, + }; +} + +function resultToFrame(reqId: string, result: ResultMessage): Up { + if (result.status === "error") { + return { + t: "result", + reqId, + ok: false, + error: { + code: result.error?.code ?? "error", + message: result.error?.message ?? "(no message)", + }, + }; + } + return { + t: "result", + reqId, + ok: true, + ...(result.data !== undefined && { data: result.data }), + }; +} diff --git a/packages/typescript/integrations/relay-cli/src/cli.ts b/packages/typescript/integrations/relay-cli/src/cli.ts new file mode 100644 index 00000000..b48dd3f1 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/src/cli.ts @@ -0,0 +1,138 @@ +#!/usr/bin/env node +import { existsSync, readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { createRelayBridge } from "./bridge"; +import { DEFAULT_RELAY_URL } from "./protocol"; + +const VERSION = readPackageVersion(); + +type CliOptions = { + command?: "login"; + url: string; + token?: string; + verbose: boolean; +}; + +const log = { + info: (...args: unknown[]) => console.error("[slop-relay]", ...args), + error: (...args: unknown[]) => console.error("[slop-relay] ERROR:", ...args), +}; + +function printHelp(): void { + console.log(`slop-relay ${VERSION} + +Run a local SLOP relay agent that connects outbound to a hosted bridge. + +Usage: + slop-relay [--url ] [--token ] [--verbose] + slop-relay login + slop-relay --help + slop-relay --version + +Configuration: + --url Bridge WebSocket URL. Defaults to ${DEFAULT_RELAY_URL} + --token Relay token. Falls back to SLOP_RELAY_TOKEN or ~/.slop/relay.json + --verbose Print connection lifecycle logs +`); +} + +function readPackageVersion(): string { + try { + const json = JSON.parse(readFileSync(new URL("../package.json", import.meta.url), "utf8")) as { version?: unknown }; + return typeof json.version === "string" ? json.version : "0.1.0"; + } catch { + return "0.1.0"; + } +} + +function parseArgs(argv: string[]): CliOptions | null { + let command: CliOptions["command"]; + let url = process.env.SLOP_RELAY_URL ?? DEFAULT_RELAY_URL; + let token = process.env.SLOP_RELAY_TOKEN; + let verbose = false; + + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + if (arg === "--help" || arg === "-h") { + printHelp(); + return null; + } + if (arg === "--version" || arg === "-v") { + console.log(VERSION); + return null; + } + if (arg === "login") { + command = "login"; + continue; + } + if (arg === "--verbose") { + verbose = true; + continue; + } + if (arg === "--url") { + const value = argv[++index]; + if (!value) throw new Error("Missing value for --url"); + url = value; + continue; + } + if (arg === "--token") { + const value = argv[++index]; + if (!value) throw new Error("Missing value for --token"); + token = value; + continue; + } + throw new Error(`Unknown argument: ${arg}`); + } + + return { command, url, token: token ?? readTokenFile(), verbose }; +} + +function readTokenFile(): string | undefined { + const path = join(homedir(), ".slop", "relay.json"); + if (!existsSync(path)) return undefined; + try { + const json = JSON.parse(readFileSync(path, "utf8")) as { token?: unknown }; + return typeof json.token === "string" && json.token ? json.token : undefined; + } catch { + return undefined; + } +} + +async function main(): Promise { + const options = parseArgs(process.argv.slice(2)); + if (!options) return; + + if (options.command === "login") { + console.error("Relay login is coming soon. For now, set SLOP_RELAY_TOKEN or pass --token."); + return; + } + + if (!options.token) { + console.error("No token. Run slop-relay login (coming soon) or set SLOP_RELAY_TOKEN."); + process.exit(1); + } + + const logger = options.verbose ? log : { info: () => {}, error: log.error }; + const bridge = createRelayBridge({ + url: options.url, + token: options.token, + relayVersion: VERSION, + logger, + }); + + bridge.start(); + logger.info(`Connected relay agent starting for ${options.url}`); + + const stop = () => { + bridge.stop(); + process.exit(0); + }; + process.on("SIGINT", stop); + process.on("SIGTERM", stop); +} + +main().catch((error: unknown) => { + log.error("Fatal:", error instanceof Error ? error.message : String(error)); + process.exit(1); +}); diff --git a/packages/typescript/integrations/relay-cli/src/index.ts b/packages/typescript/integrations/relay-cli/src/index.ts new file mode 100644 index 00000000..9c2155d3 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/src/index.ts @@ -0,0 +1,12 @@ +export { createRelayBridge, type RelayBridge, type RelayBridgeOptions, summarizeProviders } from "./bridge"; +export { + DEFAULT_RELAY_URL, + type Down, + isDownFrame, + isPlainRecord, + type Logger, + type ProviderSummary, + RELAY_PROTOCOL_VERSION, + type Up, +} from "./protocol"; +export { createRelayClient, RelayClient, type RelayClientOptions } from "./relay-client"; diff --git a/packages/typescript/integrations/relay-cli/src/protocol.ts b/packages/typescript/integrations/relay-cli/src/protocol.ts new file mode 100644 index 00000000..f0b51b77 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/src/protocol.ts @@ -0,0 +1,58 @@ +import type { PatchOp, SlopNode } from "@slop-ai/consumer"; + +export const RELAY_PROTOCOL_VERSION = 1; +export const DEFAULT_RELAY_URL = "wss://bridge.slopai.dev/relay"; + +export interface ProviderSummary { + id: string; + name: string; + transport: "unix" | "ws" | "stdio" | "relay"; + source: "local" | "bridge"; + status: "connected" | "connecting" | "disconnected"; + capabilities?: string[]; +} + +export type Up = + | { t: "hello"; relayVersion: string; protocolVersion: typeof RELAY_PROTOCOL_VERSION } + | { t: "providers"; list: ProviderSummary[] } + | { t: "snapshot"; reqId: string; subId: string; tree: SlopNode; version?: number; seq?: number } + | { t: "patch"; subId: string; ops: PatchOp[]; version: number; seq?: number } + | { t: "result"; reqId: string; ok: true; data?: unknown } + | { t: "result"; reqId: string; ok: false; error: { code: string; message: string } }; + +export type Down = + | { t: "subscribe"; reqId: string; subId: string; providerId: string; path?: string; depth?: number } + | { t: "unsubscribe"; subId: string } + | { t: "invoke"; reqId: string; providerId: string; path: string; action: string; params?: Record } + | { t: "ping" }; + +export type Logger = { + info: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; +}; + +export function isDownFrame(value: unknown): value is Down { + if (typeof value !== "object" || value === null) return false; + const frame = value as Record; + switch (frame.t) { + case "subscribe": + return typeof frame.reqId === "string" && typeof frame.subId === "string" && typeof frame.providerId === "string"; + case "unsubscribe": + return typeof frame.subId === "string"; + case "invoke": + return ( + typeof frame.reqId === "string" && + typeof frame.providerId === "string" && + typeof frame.path === "string" && + typeof frame.action === "string" + ); + case "ping": + return true; + default: + return false; + } +} + +export function isPlainRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/packages/typescript/integrations/relay-cli/src/relay-client.ts b/packages/typescript/integrations/relay-cli/src/relay-client.ts new file mode 100644 index 00000000..3fecd348 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/src/relay-client.ts @@ -0,0 +1,193 @@ +import WebSocket from "ws"; +import { type Down, isDownFrame, type Logger, RELAY_PROTOCOL_VERSION, type Up } from "./protocol"; + +export interface RelayClientOptions { + url: string; + token: string; + relayVersion: string; + logger?: Logger; + reconnectBaseDelayMs?: number; + reconnectMaxDelayMs?: number; + stableConnectionMs?: number; +} + +const DEFAULT_RECONNECT_BASE_DELAY_MS = 1000; +const DEFAULT_RECONNECT_MAX_DELAY_MS = 30_000; +const DEFAULT_STABLE_CONNECTION_MS = 60_000; + +export class RelayClient { + private readonly logger: Logger; + private readonly openListeners = new Set<() => void>(); + private readonly closeListeners = new Set<() => void>(); + private readonly frameListeners = new Set<(frame: Down) => void>(); + private readonly errorListeners = new Set<(error: Error) => void>(); + private socket: WebSocket | null = null; + private stopped = true; + private attempts = 0; + private reconnectTimer: ReturnType | null = null; + private stableTimer: ReturnType | null = null; + private latestProviders: Extract | null = null; + + constructor(private readonly options: RelayClientOptions) { + this.logger = options.logger ?? { info: () => {}, error: () => {} }; + } + + on(event: "open", listener: () => void): void; + on(event: "close", listener: () => void): void; + on(event: "frame", listener: (frame: Down) => void): void; + on(event: "error", listener: (error: Error) => void): void; + on( + event: "open" | "close" | "frame" | "error", + listener: (() => void) | ((frame: Down) => void) | ((error: Error) => void), + ): void { + if (event === "open") this.openListeners.add(listener as () => void); + if (event === "close") this.closeListeners.add(listener as () => void); + if (event === "frame") this.frameListeners.add(listener as (frame: Down) => void); + if (event === "error") this.errorListeners.add(listener as (error: Error) => void); + } + + off(event: "open", listener: () => void): void; + off(event: "close", listener: () => void): void; + off(event: "frame", listener: (frame: Down) => void): void; + off(event: "error", listener: (error: Error) => void): void; + off( + event: "open" | "close" | "frame" | "error", + listener: (() => void) | ((frame: Down) => void) | ((error: Error) => void), + ): void { + if (event === "open") this.openListeners.delete(listener as () => void); + if (event === "close") this.closeListeners.delete(listener as () => void); + if (event === "frame") this.frameListeners.delete(listener as (frame: Down) => void); + if (event === "error") this.errorListeners.delete(listener as (error: Error) => void); + } + + start(): void { + if (!this.stopped) return; + this.stopped = false; + this.connect(); + } + + stop(): void { + this.stopped = true; + this.clearTimers(); + const socket = this.socket; + this.socket = null; + if (socket && socket.readyState !== WebSocket.CLOSED) { + socket.close(); + } + } + + send(frame: Up): void { + if (frame.t === "providers") { + this.latestProviders = frame; + } + if (!this.socket || this.socket.readyState !== WebSocket.OPEN) { + return; + } + this.write(frame); + } + + private connect(): void { + if (this.stopped) return; + + const socket = new WebSocket(this.options.url, { + headers: { Authorization: `Bearer ${this.options.token}` }, + }); + this.socket = socket; + + socket.on("open", () => { + this.write({ + t: "hello", + relayVersion: this.options.relayVersion, + protocolVersion: RELAY_PROTOCOL_VERSION, + }); + if (this.latestProviders) { + this.write(this.latestProviders); + } + this.emitOpen(); + this.stableTimer = setTimeout(() => { + this.attempts = 0; + }, this.options.stableConnectionMs ?? DEFAULT_STABLE_CONNECTION_MS); + this.stableTimer.unref?.(); + }); + + socket.on("message", (raw) => { + try { + const parsed = JSON.parse(raw.toString()) as unknown; + if (!isDownFrame(parsed)) { + this.logger.error("[slop-relay] Ignoring invalid bridge frame"); + return; + } + this.emitFrame(parsed); + } catch (error) { + this.emitError(error instanceof Error ? error : new Error(String(error))); + } + }); + + socket.on("close", () => { + if (this.socket === socket) { + this.socket = null; + } + this.clearStableTimer(); + this.emitClose(); + this.scheduleReconnect(); + }); + + socket.on("error", (error) => { + this.emitError(error); + }); + } + + private write(frame: Up): void { + if (!this.socket || this.socket.readyState !== WebSocket.OPEN) return; + this.socket.send(JSON.stringify(frame)); + } + + private scheduleReconnect(): void { + if (this.stopped || this.reconnectTimer) return; + this.attempts += 1; + const base = this.options.reconnectBaseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS; + const cap = this.options.reconnectMaxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS; + const delay = Math.min(base * 2 ** Math.max(0, this.attempts - 1), cap); + const jittered = Math.round(delay * (0.75 + Math.random() * 0.5)); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.connect(); + }, jittered); + this.reconnectTimer.unref?.(); + } + + private clearTimers(): void { + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + this.clearStableTimer(); + } + + private clearStableTimer(): void { + if (this.stableTimer) { + clearTimeout(this.stableTimer); + this.stableTimer = null; + } + } + + private emitOpen(): void { + for (const listener of this.openListeners) listener(); + } + + private emitClose(): void { + for (const listener of this.closeListeners) listener(); + } + + private emitFrame(frame: Down): void { + for (const listener of this.frameListeners) listener(frame); + } + + private emitError(error: Error): void { + for (const listener of this.errorListeners) listener(error); + } +} + +export function createRelayClient(options: RelayClientOptions): RelayClient { + return new RelayClient(options); +} diff --git a/packages/typescript/integrations/relay-cli/tsconfig.json b/packages/typescript/integrations/relay-cli/tsconfig.json new file mode 100644 index 00000000..a7ff2eb1 --- /dev/null +++ b/packages/typescript/integrations/relay-cli/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ES2022", + "moduleResolution": "bundler", + "strict": true, + "declaration": true, + "lib": ["ES2022"], + "types": ["node"], + "outDir": "dist", + "rootDir": "src", + "skipLibCheck": true + }, + "include": ["src"] +}