adminandClaude Opus 5.5
fb981b5b0f
The agent built every part again that already stood in a graph, node by node, while the editor has Ctrl+C and Ctrl+V for it. The new command `copy_nodes` in the area graphs copies nodes into the same graph or another one, in one undo step with the rest of the call. - The links between the copied nodes come along, and so do groups and comments that are only about copied nodes. A node belongs to a group by its middle, as when it is dragged. `boxesAbout` in annotationFollow.ts decides this. - `made` names each copy by the id of its original, so later steps link to it with `$copy.<id>`. `$copy` alone is the first one. - Without x and y the copies are arranged at the end of the call like new nodes. With them they keep their layout, top left corner there. - Nodes the target graph does not take are refused by name, not left out as Ctrl+V does. The rule is the graph tab's: `fitsGraph` and `fittingPart` moved from clipboard.ts to paletteEntries.ts. - A copied Variables node takes the variables of its behavior along into the target behavior (`CarryVariablesCommand`), with default, value, Keep and Private. A variable of that name of another kind there is refused. - The relay's instructions name `copy_nodes`. Found on the way: - Groups made by the agent can take in a node of another group once the nodes are arranged. lint_graph does not count that. - Undo after a paste leaves an empty `groups: []`, as AddAnnotationCommand does. It is the same as no list. Tests: - agent-copy (new, 3 tests) with the agent, relay, clipboard and palette tests: 118/118. - Counter-probes: 6 bendings, all compile, all red. - e2e 71 (atx): ALL PASSED, 26 checks. - Gate (atx): CHECK EXIT 0, 327 files, 5024 tests. - Live in the MCP Test Project: 5 nodes of Arrow keys walk into a new behavior, 6 links, group, comment and speed (300) along, lint_graph 0 everywhere, taken back with undo. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
656 lines
25 KiB
JavaScript
656 lines
25 KiB
JavaScript
import { randomUUID } from 'node:crypto';
|
||
import { createServer, request as httpRequest } from 'node:http';
|
||
import { acceptWebSocket } from './websocket.mjs';
|
||
|
||
/**
|
||
* The MCP relay (#429): an agent speaks MCP over Streamable HTTP, the user's editor tab is on a
|
||
* WebSocket, and the relay passes the agent's tool calls to the tab of the same account. The tools
|
||
* live in the editor; the relay knows their list (`tools`) and nothing of what they do.
|
||
*
|
||
* Who is who is the PHP API's call (`api`): an agent's token is asked about on every request, so a
|
||
* revoked one stops at the next, and a tab hands in a ticket it got with its session. One tab per
|
||
* account is on at a time; the next one takes over.
|
||
*
|
||
* With `backend` it also stands in front of a server of plain HTTP and passes everything but its own
|
||
* path on, which is how CI puts it beside `php -S` on one origin (nginx does that on stein2).
|
||
*/
|
||
|
||
/** Newest first; an unknown version asked for gets the newest. */
|
||
export const PROTOCOL_VERSIONS = ['2025-11-25', '2025-06-18', '2025-03-26', '2024-11-05'];
|
||
|
||
export const VIEW_URI = 'feuerwerk://view';
|
||
|
||
const VIEW_RESOURCE = {
|
||
uri: VIEW_URI,
|
||
name: 'view',
|
||
title: 'What the user sees',
|
||
description:
|
||
"The user's editor right now: open tab and document, selection, camera, play state, open dialog. " +
|
||
'Subscribe to hear when the user switches tab or selection.',
|
||
mimeType: 'application/json',
|
||
};
|
||
|
||
const INSTRUCTIONS = [
|
||
"Feuerwerk is a 2D game editor in the browser. You work inside the user's open editor tab,",
|
||
'live: the user sees every change you make and can take it back with Ctrl+Z.',
|
||
'Start with get_view to learn what the user is looking at; screenshot shows it as a picture.',
|
||
'Read the project with list_assets, get_asset and get_graph. Change it with run_commands: its steps',
|
||
'are commands from catalog topic "commands" (by area; ask with query for their arguments), run as',
|
||
'one undo step, and a step can name what an earlier one made ("as": "x", then "$x"). Nodes for a',
|
||
'graph come from catalog topic "nodes". A Variables node reads or writes a variable once add_port',
|
||
'names it. A behavior acts on the object it is placed under: place it with parent set to that object.',
|
||
'Give new nodes no x and y: the editor arranges them when the call ends, next to what they link to',
|
||
'and into its group; the nodes already there stay, or move right to make room. Keep a graph',
|
||
'readable: a value needed far away comes from a second Get or Owner node next to its user, not',
|
||
'over a long link. Check with lint_graph, the crossings too: when there are more than the links',
|
||
'need, arrange_graph orders the whole graph for the fewest. Put nodes that belong together into a',
|
||
'group with add_group, and say what they do with add_comment; a comment a node lies on',
|
||
'(coveredNotes) goes back above its nodes with fit_group. A part already built elsewhere is',
|
||
'copied with copy_nodes, into the same graph or another, rather than built again. Music plays',
|
||
'nothing without an instrument, and an instrument sounds only the notes its Notes nodes take',
|
||
'in, C4 to B4 in a new one: get_asset lists under silent what of a pattern, song or composition',
|
||
'is not heard.',
|
||
'apply_preset gives an instrument that sounds every note in one step. Pictures, fonts, sounds',
|
||
'and videos are files: list_files lists those the server holds, add_file_asset makes an asset',
|
||
'of one. Before deleting or reworking an asset, get_asset says under usedBy what names it.',
|
||
'The editor follows your edits: after run_commands the user sees the graph, scene or asset it worked',
|
||
'on, with what it touched selected. Use open and select to show anything else.',
|
||
'Test what you built yourself: play, then play_input sends keys and mouse clicks into the game,',
|
||
'and get_runtime reads where the objects are, their variables and the console. A game needs a',
|
||
'camera in its scene to show anything; get_view says when it has none.',
|
||
'When a tool answers that no editor is connected, ask the user to open the editor and switch on',
|
||
'"Agent" in its account menu.',
|
||
].join(' ');
|
||
|
||
const NOT_CONNECTED =
|
||
'No Feuerwerk editor is connected for this account. Ask the user to open the editor and switch on ' +
|
||
'"Agent" in the account menu (top right), then try again.';
|
||
|
||
/** How a tab says goodbye on the WebSocket (4000–4999 are the application's). */
|
||
export const CLOSE = { replaced: 4000, badHello: 4400, refused: 4401, helloTimeout: 4408 };
|
||
|
||
/**
|
||
* @param {object} options
|
||
* @param {number} [options.port] 0 for any free one
|
||
* @param {string} [options.host]
|
||
* @param {{verifyToken(token: string, address: string): Promise<ApiAnswer>, redeemTicket(ticket: string): Promise<ApiAnswer>}} options.api
|
||
* @param {() => Promise<object[]>} options.tools the tool list, asked for on every use
|
||
* @param {string[]} options.origins the editor's origins: a tab has to come from one, an agent may
|
||
* send none, but not another
|
||
* @param {boolean} [options.trustProxy] take the agent's address from `X-Real-IP` (nginx sets it)
|
||
* @param {string} [options.backend] pass everything off the relay's path on to this server
|
||
* @param {(line: string) => void} [options.log]
|
||
*/
|
||
export async function startRelay(options) {
|
||
const {
|
||
api,
|
||
tools,
|
||
origins,
|
||
trustProxy = false,
|
||
backend,
|
||
log = () => {},
|
||
callTimeoutMs = 30_000,
|
||
waitTimeoutMs = 120_000,
|
||
helloTimeoutMs = 10_000,
|
||
pingMs = 25_000,
|
||
maxMessageBytes = 32 * 1024 * 1024,
|
||
maxBodyBytes = 4 * 1024 * 1024,
|
||
sessionIdleMs = 12 * 60 * 60 * 1000,
|
||
agentRecentMs = 2 * 60 * 1000,
|
||
maxSessionsPerAccount = 20,
|
||
} = options;
|
||
|
||
/** @type {Map<string, Tab>} by account, lower case */
|
||
const tabs = new Map();
|
||
/** @type {Map<string, Session>} by session id */
|
||
const sessions = new Map();
|
||
|
||
const accountKey = (user) => user.toLowerCase();
|
||
const sessionsOf = (key) => [...sessions.values()].filter((session) => session.key === key);
|
||
|
||
const server = createServer((req, res) => {
|
||
handle(req, res).catch((error) => {
|
||
log(`error: ${error instanceof Error ? error.message : String(error)}`);
|
||
if (!res.headersSent) reply(res, 500, { error: 'Internal relay error' });
|
||
else res.end();
|
||
});
|
||
});
|
||
server.on('upgrade', (req, socket, head) => upgrade(req, socket, head));
|
||
|
||
const keepAlive = setInterval(() => {
|
||
for (const tab of tabs.values()) tab.conn.ping();
|
||
const now = Date.now();
|
||
for (const session of sessions.values()) {
|
||
for (const stream of session.streams) stream.write(': ping\n\n');
|
||
if (session.streams.size === 0 && now - session.seenAt > sessionIdleMs) endSession(session);
|
||
}
|
||
// An agent that went quiet is counted out with time, not with a request.
|
||
for (const key of tabs.keys()) tellAgents(key);
|
||
}, pingMs);
|
||
|
||
async function handle(req, res) {
|
||
if (!isRelayPath(req.url)) {
|
||
if (backend) return passOn(req, res, backend);
|
||
return reply(res, 404, { error: 'Not found' });
|
||
}
|
||
const origin = req.headers.origin;
|
||
if (origin !== undefined && !origins.includes(origin)) {
|
||
return reply(res, 403, { error: 'Requests from other sites are refused' });
|
||
}
|
||
if (req.method === 'POST') return post(req, res);
|
||
if (req.method === 'GET') return stream(req, res);
|
||
if (req.method === 'DELETE') return remove(req, res);
|
||
return reply(res, 405, { error: 'Method not allowed' }, { Allow: 'GET, POST, DELETE' });
|
||
}
|
||
|
||
/** The account behind the agent's token, or null once the refusal has been sent. */
|
||
async function agent(req, res) {
|
||
const match = /^Bearer\s+(\S+)$/i.exec(String(req.headers.authorization ?? ''));
|
||
if (!match) {
|
||
reply(res, 401, { error: 'An agent token is needed: Authorization: Bearer fwa_…' }, BEARER);
|
||
return null;
|
||
}
|
||
let answer;
|
||
try {
|
||
answer = await api.verifyToken(match[1], addressOf(req));
|
||
} catch (error) {
|
||
log(`api unreachable: ${error instanceof Error ? error.message : String(error)}`);
|
||
reply(res, 502, { error: 'The editor server did not answer' });
|
||
return null;
|
||
}
|
||
if (answer.status === 200 && typeof answer.user === 'string') return answer.user;
|
||
if (answer.status === 429) {
|
||
reply(
|
||
res,
|
||
429,
|
||
{ error: answer.error ?? 'Too many failed attempts' },
|
||
{ 'Retry-After': answer.retryAfter ?? '900' },
|
||
);
|
||
return null;
|
||
}
|
||
reply(res, 401, { error: 'Unknown, expired or revoked agent token' }, BEARER);
|
||
return null;
|
||
}
|
||
|
||
function addressOf(req) {
|
||
const forwarded = req.headers['x-real-ip'];
|
||
return trustProxy && typeof forwarded === 'string' && forwarded !== ''
|
||
? forwarded
|
||
: (req.socket.remoteAddress ?? '');
|
||
}
|
||
|
||
/** The session a request names, if it is this account's; 404 otherwise, which tells a client to start over. */
|
||
function sessionOf(req, res, user) {
|
||
const session = sessions.get(String(req.headers['mcp-session-id'] ?? ''));
|
||
if (!session || session.key !== accountKey(user)) {
|
||
reply(res, 404, { jsonrpc: '2.0', id: null, error: { code: -32001, message: 'Session not found' } });
|
||
return null;
|
||
}
|
||
session.seenAt = Date.now();
|
||
return session;
|
||
}
|
||
|
||
async function post(req, res) {
|
||
const user = await agent(req, res);
|
||
if (user === null) return;
|
||
const body = await readBody(req, maxBodyBytes);
|
||
if (body === null) return reply(res, 413, { error: 'Request body too large' });
|
||
let parsed;
|
||
try {
|
||
parsed = JSON.parse(body);
|
||
} catch {
|
||
return reply(res, 400, rpcError(null, -32700, 'Parse error'));
|
||
}
|
||
const messages = Array.isArray(parsed) ? parsed : [parsed];
|
||
if (messages.length === 0) return reply(res, 400, rpcError(null, -32600, 'Invalid request'));
|
||
|
||
let session;
|
||
if (messages.some((message) => message?.method === 'initialize')) {
|
||
if (messages.length !== 1) return reply(res, 400, rpcError(null, -32600, 'initialize comes alone'));
|
||
session = openSession(user);
|
||
} else {
|
||
session = sessionOf(req, res, user);
|
||
if (!session) return;
|
||
}
|
||
|
||
const replies = [];
|
||
for (const message of messages) {
|
||
if (message === null || typeof message !== 'object' || message.jsonrpc !== '2.0') {
|
||
replies.push(rpcError(null, -32600, 'Invalid request'));
|
||
} else if (typeof message.method === 'string' && message.id !== undefined && message.id !== null) {
|
||
replies.push(await dispatch(session, message));
|
||
}
|
||
// Notifications and the client's answers need nothing back.
|
||
}
|
||
const headers = { 'Mcp-Session-Id': session.id };
|
||
if (replies.length === 0) return reply(res, 202, undefined, headers);
|
||
reply(res, 200, Array.isArray(parsed) ? replies : replies[0], headers);
|
||
}
|
||
|
||
function openSession(user) {
|
||
const key = accountKey(user);
|
||
const own = sessionsOf(key).sort((a, b) => a.seenAt - b.seenAt);
|
||
while (own.length >= maxSessionsPerAccount) endSession(own.shift());
|
||
/** @type {Session} */
|
||
const session = {
|
||
id: randomUUID(),
|
||
user,
|
||
key,
|
||
protocolVersion: PROTOCOL_VERSIONS[0],
|
||
subscriptions: new Set(),
|
||
streams: new Set(),
|
||
seenAt: Date.now(),
|
||
calledAt: 0,
|
||
lastSummary: null,
|
||
};
|
||
sessions.set(session.id, session);
|
||
log(`agent session ${session.id.slice(0, 8)} for ${user}`);
|
||
tellAgents(key);
|
||
return session;
|
||
}
|
||
|
||
function endSession(session) {
|
||
for (const stream of session.streams) stream.end();
|
||
session.streams.clear();
|
||
sessions.delete(session.id);
|
||
tellAgents(session.key);
|
||
}
|
||
|
||
async function dispatch(session, message) {
|
||
const { id, method } = message;
|
||
const params = message.params !== null && typeof message.params === 'object' ? message.params : {};
|
||
const ok = (result) => ({ jsonrpc: '2.0', id, result });
|
||
switch (method) {
|
||
case 'initialize': {
|
||
const asked = params.protocolVersion;
|
||
session.protocolVersion = PROTOCOL_VERSIONS.includes(asked) ? asked : PROTOCOL_VERSIONS[0];
|
||
return ok({
|
||
protocolVersion: session.protocolVersion,
|
||
capabilities: { tools: {}, resources: { subscribe: true } },
|
||
serverInfo: { name: 'feuerwerk', title: 'Feuerwerk', version: '1.0.0' },
|
||
instructions: INSTRUCTIONS,
|
||
});
|
||
}
|
||
case 'ping':
|
||
return ok({});
|
||
case 'tools/list':
|
||
return ok({ tools: await tools() });
|
||
case 'tools/call': {
|
||
const name = params.name;
|
||
const args = params.arguments ?? {};
|
||
if (typeof name !== 'string' || !(await tools()).some((tool) => tool.name === name)) {
|
||
return rpcError(id, -32602, `Unknown tool: ${String(name)}`);
|
||
}
|
||
if (args === null || typeof args !== 'object' || Array.isArray(args)) {
|
||
return rpcError(id, -32602, 'arguments has to be an object');
|
||
}
|
||
return ok(await callTool(session, name, args));
|
||
}
|
||
case 'resources/list':
|
||
return ok({ resources: [VIEW_RESOURCE] });
|
||
case 'resources/templates/list':
|
||
return ok({ resourceTemplates: [] });
|
||
case 'resources/read': {
|
||
if (params.uri !== VIEW_URI) return rpcError(id, -32002, 'Resource not found', { uri: params.uri });
|
||
const tab = tabs.get(session.key);
|
||
let view = tab?.view ?? null;
|
||
if (tab && view === null) view = (await tab.call('get_view', {})).structuredContent ?? null;
|
||
if (tab) session.lastSummary = tab.summary;
|
||
const text = JSON.stringify(view ?? { connected: false, hint: NOT_CONNECTED });
|
||
return ok({ contents: [{ uri: VIEW_URI, mimeType: 'application/json', text }] });
|
||
}
|
||
case 'resources/subscribe':
|
||
case 'resources/unsubscribe':
|
||
if (params.uri !== VIEW_URI) return rpcError(id, -32002, 'Resource not found', { uri: params.uri });
|
||
if (method === 'resources/subscribe') session.subscriptions.add(VIEW_URI);
|
||
else session.subscriptions.delete(VIEW_URI);
|
||
return ok({});
|
||
default:
|
||
return rpcError(id, -32601, `Method not found: ${method}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Hands a call to the account's tab. An agent that has not looked since the user changed what they
|
||
* see gets told so with the answer: not every client follows a subscription.
|
||
*/
|
||
async function callTool(session, name, args) {
|
||
session.calledAt = Date.now();
|
||
tellAgents(session.key);
|
||
const tab = tabs.get(session.key);
|
||
if (!tab) return { content: [{ type: 'text', text: NOT_CONNECTED }], isError: true };
|
||
const seen = session.lastSummary;
|
||
const result = await tab.call(name, args);
|
||
if (name === 'get_view') {
|
||
session.lastSummary = tab.summary;
|
||
} else if (tab.summary !== null && seen !== tab.summary) {
|
||
session.lastSummary = tab.summary;
|
||
result.content = [
|
||
...result.content,
|
||
{ type: 'text', text: `The user's view has changed since you last looked: ${tab.summary}` },
|
||
];
|
||
}
|
||
return result;
|
||
}
|
||
|
||
async function stream(req, res) {
|
||
if (!String(req.headers.accept ?? '').includes('text/event-stream')) {
|
||
return reply(res, 405, { error: 'Ask for text/event-stream' }, { Allow: 'GET, POST, DELETE' });
|
||
}
|
||
// A client may drop the stream while its token is being checked: its close is over by the time
|
||
// the stream would be kept, and a stream kept then would count an agent that is gone, for good.
|
||
let dropped = false;
|
||
res.once('close', () => (dropped = true));
|
||
const user = await agent(req, res);
|
||
if (user === null || dropped) return;
|
||
const session = sessionOf(req, res, user);
|
||
if (!session) return;
|
||
res.writeHead(200, {
|
||
'Content-Type': 'text/event-stream',
|
||
'Cache-Control': 'no-cache, no-transform',
|
||
'X-Accel-Buffering': 'no',
|
||
'Mcp-Session-Id': session.id,
|
||
});
|
||
res.write(': open\n\n');
|
||
session.streams.add(res);
|
||
log(`stream open on ${session.id.slice(0, 8)}`);
|
||
tellAgents(session.key);
|
||
res.on('close', () => {
|
||
session.streams.delete(res);
|
||
log(`stream closed on ${session.id.slice(0, 8)}`);
|
||
tellAgents(session.key);
|
||
});
|
||
}
|
||
|
||
async function remove(req, res) {
|
||
const user = await agent(req, res);
|
||
if (user === null) return;
|
||
const session = sessionOf(req, res, user);
|
||
if (!session) return;
|
||
log(`agent session ${session.id.slice(0, 8)} ended by the agent`);
|
||
endSession(session);
|
||
reply(res, 204);
|
||
}
|
||
|
||
/** Tells the subscribed agents of an account that its view changed. */
|
||
function viewChanged(key) {
|
||
const note = `event: message\ndata: ${JSON.stringify({
|
||
jsonrpc: '2.0',
|
||
method: 'notifications/resources/updated',
|
||
params: { uri: VIEW_URI },
|
||
})}\n\n`;
|
||
for (const session of sessionsOf(key)) {
|
||
if (!session.subscriptions.has(VIEW_URI)) continue;
|
||
for (const out of session.streams) out.write(note);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* How many agents an account has on, for its tab to show: a session counts while its event stream
|
||
* is open or it called a tool lately. A client that went away without a DELETE leaves its
|
||
* session behind, and so does a revoked token until it is used again. Claude Code opens two
|
||
* spare sessions next to the one it keeps whenever it connects: they never call a tool.
|
||
*/
|
||
function tellAgents(key) {
|
||
const tab = tabs.get(key);
|
||
if (!tab) return;
|
||
const now = Date.now();
|
||
const count = sessionsOf(key).filter(
|
||
(session) => session.streams.size > 0 || now - session.calledAt < agentRecentMs,
|
||
).length;
|
||
if (count === tab.agents) return;
|
||
tab.agents = count;
|
||
const seen = sessionsOf(key).map(
|
||
(session) =>
|
||
`${session.id.slice(0, 8)} streams ${session.streams.size} idle ${Math.round((now - session.seenAt) / 1000)}s`,
|
||
);
|
||
log(`agents for ${tab.user}: ${count} [${seen.join('; ')}]`);
|
||
tab.conn.send(JSON.stringify({ type: 'agents', count }));
|
||
}
|
||
|
||
function upgrade(req, socket, head) {
|
||
socket.on('error', () => socket.destroy());
|
||
if (!isRelayPath(req.url)) return socket.destroy();
|
||
const origin = req.headers.origin;
|
||
if (typeof origin !== 'string' || !origins.includes(origin)) {
|
||
socket.end('HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n');
|
||
return;
|
||
}
|
||
const conn = acceptWebSocket(req, socket, head, { maxMessageBytes });
|
||
if (!conn) return;
|
||
/** @type {Tab | null} */
|
||
let tab = null;
|
||
let greeted = false;
|
||
const hello = setTimeout(() => conn.close(CLOSE.helloTimeout, 'hello first'), helloTimeoutMs);
|
||
conn.onmessage = async (text) => {
|
||
let message;
|
||
try {
|
||
message = JSON.parse(text);
|
||
} catch {
|
||
return conn.close(CLOSE.badHello, 'not JSON');
|
||
}
|
||
if (tab) return tab.receive(message);
|
||
if (greeted) return;
|
||
if (message?.type !== 'hello' || typeof message.ticket !== 'string') {
|
||
return conn.close(CLOSE.badHello, 'hello first');
|
||
}
|
||
greeted = true;
|
||
clearTimeout(hello);
|
||
let answer;
|
||
try {
|
||
answer = await api.redeemTicket(message.ticket);
|
||
} catch {
|
||
answer = { status: 502 };
|
||
}
|
||
if (conn.closed) return;
|
||
if (answer.status !== 200 || typeof answer.user !== 'string') {
|
||
return conn.close(CLOSE.refused, 'ticket refused');
|
||
}
|
||
tab = attach(answer.user, conn);
|
||
};
|
||
conn.onclose = () => {
|
||
clearTimeout(hello);
|
||
if (tab) detach(tab, 'The editor tab went away');
|
||
};
|
||
}
|
||
|
||
function attach(user, conn) {
|
||
const key = accountKey(user);
|
||
const old = tabs.get(key);
|
||
if (old) {
|
||
old.conn.send(JSON.stringify({ type: 'replaced' }));
|
||
detach(old, 'Another editor tab took over');
|
||
old.conn.close(CLOSE.replaced, 'replaced');
|
||
}
|
||
const tab = new Tab(user, conn, callTimeoutMs, waitTimeoutMs);
|
||
tab.onView = () => viewChanged(key);
|
||
tabs.set(key, tab);
|
||
conn.send(JSON.stringify({ type: 'welcome', user }));
|
||
log(`tab on for ${user}`);
|
||
tellAgents(key);
|
||
return tab;
|
||
}
|
||
|
||
function detach(tab, why) {
|
||
const key = accountKey(tab.user);
|
||
tab.abandon(why);
|
||
if (tabs.get(key) !== tab) return;
|
||
tabs.delete(key);
|
||
log(`tab off for ${tab.user}`);
|
||
viewChanged(key);
|
||
}
|
||
|
||
await new Promise((resolve, reject) => {
|
||
server.once('error', reject);
|
||
server.listen(options.port ?? 0, options.host ?? '127.0.0.1', () => resolve(undefined));
|
||
});
|
||
|
||
return {
|
||
port: /** @type {import('node:net').AddressInfo} */ (server.address()).port,
|
||
/** Accounts with a tab on, and sessions per account: for the tests. */
|
||
stats: () => ({
|
||
tabs: [...tabs.values()].map((tab) => tab.user),
|
||
sessions: [...sessions.values()].map((session) => session.user),
|
||
}),
|
||
close: () =>
|
||
new Promise((resolve) => {
|
||
clearInterval(keepAlive);
|
||
for (const tab of tabs.values()) tab.conn.close(1001, 'relay stops');
|
||
for (const session of sessions.values()) for (const out of session.streams) out.end();
|
||
server.close(() => resolve(undefined));
|
||
server.closeAllConnections();
|
||
}),
|
||
};
|
||
}
|
||
|
||
/** One editor tab on the relay: the calls it has open and the last view it reported. */
|
||
class Tab {
|
||
view = null;
|
||
/** One line of what the user sees, which the agent is told about when it changes. */
|
||
summary = null;
|
||
agents = -1;
|
||
onView = () => {};
|
||
#pending = new Map();
|
||
#next = 1;
|
||
|
||
constructor(user, conn, timeoutMs, waitTimeoutMs) {
|
||
this.user = user;
|
||
this.conn = conn;
|
||
this.timeoutMs = timeoutMs;
|
||
this.waitTimeoutMs = waitTimeoutMs;
|
||
}
|
||
|
||
/** @returns {Promise<{content: object[], isError?: boolean, structuredContent?: object}>} */
|
||
call(tool, args) {
|
||
const id = this.#next++;
|
||
return new Promise((resolve) => {
|
||
this.#pending.set(id, { resolve, timer: this.#timer(id, this.timeoutMs) });
|
||
this.conn.send(JSON.stringify({ type: 'call', id, tool, args }));
|
||
});
|
||
}
|
||
|
||
#timer(id, ms) {
|
||
return setTimeout(() => {
|
||
const waiting = this.#pending.get(id);
|
||
this.#pending.delete(id);
|
||
waiting?.resolve(failure(`The editor did not answer within ${Math.round(ms / 1000)} s`));
|
||
}, ms);
|
||
}
|
||
|
||
receive(message) {
|
||
if (message?.type === 'wait') {
|
||
// The tab is asking the user something: the call gets the longer time, once.
|
||
const waiting = this.#pending.get(message.id);
|
||
if (!waiting || waiting.extended) return;
|
||
clearTimeout(waiting.timer);
|
||
waiting.timer = this.#timer(message.id, this.waitTimeoutMs);
|
||
waiting.extended = true;
|
||
} else if (message?.type === 'result') {
|
||
const waiting = this.#pending.get(message.id);
|
||
if (!waiting) return;
|
||
this.#pending.delete(message.id);
|
||
clearTimeout(waiting.timer);
|
||
waiting.resolve(toolResult(message));
|
||
} else if (message?.type === 'view' && message.view !== null && typeof message.view === 'object') {
|
||
this.view = message.view;
|
||
this.summary = typeof message.summary === 'string' ? message.summary.slice(0, 500) : null;
|
||
this.onView();
|
||
}
|
||
}
|
||
|
||
abandon(why) {
|
||
for (const { resolve, timer } of this.#pending.values()) {
|
||
clearTimeout(timer);
|
||
resolve(failure(why));
|
||
}
|
||
this.#pending.clear();
|
||
}
|
||
}
|
||
|
||
/** What a tab answered, as a tool result: only content the protocol knows goes on to the agent. */
|
||
function toolResult(message) {
|
||
const content = Array.isArray(message.content)
|
||
? message.content.filter(
|
||
(item) =>
|
||
(item?.type === 'text' && typeof item.text === 'string') ||
|
||
(item?.type === 'image' && typeof item.data === 'string' && typeof item.mimeType === 'string'),
|
||
)
|
||
: [];
|
||
if (content.length === 0) return failure('The editor answered with nothing');
|
||
const result = { content, isError: message.isError === true };
|
||
if (message.structuredContent !== null && typeof message.structuredContent === 'object') {
|
||
result.structuredContent = message.structuredContent;
|
||
}
|
||
return result;
|
||
}
|
||
|
||
function failure(text) {
|
||
return { content: [{ type: 'text', text }], isError: true };
|
||
}
|
||
|
||
const BEARER = { 'WWW-Authenticate': 'Bearer realm="feuerwerk"' };
|
||
|
||
function isRelayPath(url) {
|
||
const path = new URL(url ?? '/', 'http://relay').pathname;
|
||
return path === '/mcp' || path.endsWith('/mcp');
|
||
}
|
||
|
||
function rpcError(id, code, message, data) {
|
||
return { jsonrpc: '2.0', id, error: data === undefined ? { code, message } : { code, message, data } };
|
||
}
|
||
|
||
function reply(res, status, body, headers = {}) {
|
||
const text = body === undefined ? '' : JSON.stringify(body);
|
||
res.writeHead(status, {
|
||
...(body === undefined ? {} : { 'Content-Type': 'application/json' }),
|
||
'Cache-Control': 'no-store',
|
||
...headers,
|
||
});
|
||
res.end(text);
|
||
}
|
||
|
||
/** The request body as text, or null when it is larger than `max` bytes. */
|
||
function readBody(req, max) {
|
||
return new Promise((resolve, reject) => {
|
||
const chunks = [];
|
||
let size = 0;
|
||
req.on('data', (chunk) => {
|
||
size += chunk.length;
|
||
if (size > max) {
|
||
resolve(null);
|
||
req.removeAllListeners('data');
|
||
req.resume();
|
||
return;
|
||
}
|
||
chunks.push(chunk);
|
||
});
|
||
req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
|
||
req.on('error', reject);
|
||
});
|
||
}
|
||
|
||
/** Front mode: one request through to the backend, streamed both ways (Range requests included). */
|
||
function passOn(req, res, backend) {
|
||
const target = new URL(req.url ?? '/', backend);
|
||
const out = httpRequest(
|
||
target,
|
||
{ method: req.method, headers: { ...req.headers, host: target.host } },
|
||
(answer) => {
|
||
res.writeHead(answer.statusCode ?? 502, answer.headers);
|
||
answer.pipe(res);
|
||
},
|
||
);
|
||
out.on('error', () => {
|
||
if (!res.headersSent) res.writeHead(502);
|
||
res.end();
|
||
});
|
||
req.pipe(out);
|
||
}
|
||
|
||
/**
|
||
* @typedef {{status: number, user?: string, role?: string, error?: string, retryAfter?: string | null}} ApiAnswer
|
||
* @typedef {{id: string, user: string, key: string, protocolVersion: string, subscriptions: Set<string>,
|
||
* streams: Set<import('node:http').ServerResponse>, seenAt: number, calledAt: number,
|
||
* lastSummary: string | null}} Session
|
||
*/
|