197 lines
12 KiB
TypeScript
197 lines
12 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { registerPushRoutes, type AgentPushCoordinator, type ReadState } from './agent-push.js';
|
|
import type { IncomingMessage } from 'node:http';
|
|
import type { Duplex } from 'node:stream';
|
|
import type { FastifyInstance } from 'fastify';
|
|
import { WebSocket, WebSocketServer } from 'ws';
|
|
import { z } from 'zod';
|
|
import { requireAdmin } from './auth.js';
|
|
import { prisma } from './db.js';
|
|
import { env } from './env.js';
|
|
import { AgentAttachmentReferences, registerAgentAttachmentRoutes, requireAgentAttachments } from './agent-attachments.js';
|
|
|
|
const Cursor = z.string().regex(/^\d{1,16}$/).refine(value => Number.isSafeInteger(Number(value)), 'Invalid cursor')
|
|
.transform(value => String(Number(value)));
|
|
const PageQuery = z.object({ after: Cursor.optional(), before: Cursor.optional(), limit: z.coerce.number().int().min(1).max(200).optional() })
|
|
.refine(value => !(value.after && value.before), 'Use after or before, not both');
|
|
const MessageBody = z.object({ content: z.string().max(40_000), requestId: z.string().min(1).max(256), attachments: AgentAttachmentReferences.optional() })
|
|
.refine(value => value.content.trim().length > 0 || Boolean(value.attachments?.length), 'content or attachments are required');
|
|
const ReadBody = z.object({ cursor: Cursor, sourceId: z.string().uuid().optional() });
|
|
const AuthFrame = z.object({ type: z.literal('auth'), token: z.string().nullable().optional(), after: Cursor.optional(), clientKind: z.enum(['web', 'native']).optional(), presenceSessionId: z.string().uuid().optional() });
|
|
const SubscribeFrame = z.object({ type: z.literal('subscribe'), after: Cursor.optional() });
|
|
const SubmissionReceipt = z.object({ id: z.number().int().positive().max(Number.MAX_SAFE_INTEGER),
|
|
status: z.enum(['queued', 'placed', 'done', 'unanswered']), messageId: z.string(), attachmentIds: z.array(z.string()).optional() });
|
|
|
|
type AgentOptions = { baseUrl?: string; apiKey?: string };
|
|
const parse = <T>(schema: z.ZodType<T>, value: unknown): T => {
|
|
const result = schema.safeParse(value);
|
|
if (!result.success) throw Object.assign(new Error(result.error.issues.map(issue => issue.message).join('; ')), { statusCode: 400 });
|
|
return result.data;
|
|
};
|
|
|
|
/** Sybil Agent owns the conversation; the chat server stores only its shared read cursor. */
|
|
export async function registerAgentRoutes(app: FastifyInstance, options: AgentOptions = {}) {
|
|
const baseUrl = options.baseUrl || env.SYBIL_AGENT_BASE_URL;
|
|
const apiKey = options.apiKey ?? env.SYBIL_AGENT_API_KEY;
|
|
const readListeners = new Set<(state: ReadState) => void>();
|
|
const push = await registerPushRoutes(app, prisma, { baseUrl, apiKey }, state => { for (const send of readListeners) send(state); });
|
|
// An open development proxy must not expose a token-protected attachment runtime's
|
|
// history/model context using its server-side credential.
|
|
const guardOpenProxy = async (signal?: AbortSignal) => {
|
|
if (env.ADMIN_TOKEN) return;
|
|
try {
|
|
const response = await fetch(new URL('/api/conversations/root', baseUrl), {
|
|
headers: { 'X-Sybil-Client': 'backend-proxy-v1', Authorization: `Bearer ${apiKey}` }, signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(15_000)]) : AbortSignal.timeout(15_000), redirect: 'error',
|
|
});
|
|
const descriptor = await response.json() as { capabilities?: { attachments?: unknown } };
|
|
if (!response.ok || descriptor?.capabilities?.attachments) throw new Error('Private runtime');
|
|
} catch { throw Object.assign(new Error('Authenticated Sybil access is required for this Agent runtime.'), { statusCode: 503, code: 'attachments_unavailable' }); }
|
|
};
|
|
const proxyJson = async (path: string, method = 'GET', body?: unknown, signal?: AbortSignal) => {
|
|
if (!apiKey) throw app.httpErrors.serviceUnavailable('Sybil Agent is not configured');
|
|
await guardOpenProxy(signal);
|
|
let response: Response;
|
|
try {
|
|
response = await fetch(new URL(path, baseUrl), { method,
|
|
headers: { 'X-Sybil-Client': 'backend-proxy-v1', Authorization: `Bearer ${apiKey}`, ...(body === undefined ? {} : { 'Content-Type': 'application/json' }) },
|
|
...(body === undefined ? {} : { body: JSON.stringify(body) }), signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(15_000)]) : AbortSignal.timeout(15_000),
|
|
});
|
|
} catch { throw app.httpErrors.badGateway('Sybil Agent is unavailable'); }
|
|
let data: unknown;
|
|
try { data = await response.json(); }
|
|
catch { throw app.httpErrors.badGateway('Sybil Agent returned invalid JSON'); }
|
|
return { status: response.status, data };
|
|
};
|
|
|
|
app.get('/v1/agent', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store');
|
|
requireAdmin(req);
|
|
const result = await proxyJson('/api/conversations/root');
|
|
reply.code(result.status);
|
|
if (result.status !== 200) return result.data;
|
|
const descriptor: Record<string, unknown> = { ...result.data as Record<string, unknown>, websocket: '/v1/agent/ws' };
|
|
if (!env.ADMIN_TOKEN) delete descriptor.capabilities;
|
|
return descriptor;
|
|
});
|
|
app.get('/v1/agent/messages', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store');
|
|
requireAdmin(req);
|
|
const query = parse(PageQuery, req.query);
|
|
const params = new URLSearchParams();
|
|
for (const [name, value] of Object.entries(query)) if (value !== undefined) params.set(name, String(value));
|
|
const result = await proxyJson(`/api/conversations/root/messages?${params}`);
|
|
reply.code(result.status);
|
|
return result.data;
|
|
});
|
|
app.post('/v1/agent/messages', { bodyLimit: 256 * 1024 }, async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store');
|
|
requireAdmin(req);
|
|
const body = parse(MessageBody, req.body);
|
|
await push.notePrompt();
|
|
if (body.attachments?.length) await requireAgentAttachments(proxyJson);
|
|
const result = await proxyJson('/api/conversations/root/messages', 'POST', { ...body, channel: 'sybil', whenBusy: 'followUp', wait: false });
|
|
if (result.status >= 200 && result.status < 300) {
|
|
const receipt = SubmissionReceipt.safeParse(result.data);
|
|
if (!receipt.success || receipt.data.messageId !== `inbound:sybil:${body.requestId}`
|
|
|| (body.attachments?.length && JSON.stringify(receipt.data.attachmentIds) !== JSON.stringify(body.attachments.map(item => item.id)))) {
|
|
throw app.httpErrors.badGateway('Sybil Agent returned an invalid message receipt');
|
|
}
|
|
}
|
|
reply.code(result.status);
|
|
return result.data;
|
|
});
|
|
app.get('/v1/agent/read-state', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store');
|
|
requireAdmin(req);
|
|
return push.readState();
|
|
});
|
|
app.put('/v1/agent/read-state', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store');
|
|
requireAdmin(req);
|
|
const { cursor, sourceId } = parse(ReadBody, req.body);
|
|
return push.advanceRead(cursor, sourceId);
|
|
});
|
|
|
|
await registerAgentAttachmentRoutes(app, { baseUrl, apiKey, proxyJson });
|
|
attachAgentSocket(app, baseUrl, apiKey, guardOpenProxy, push, readListeners);
|
|
}
|
|
|
|
function attachAgentSocket(app: FastifyInstance, baseUrl: string, apiKey: string | undefined, guardOpenProxy: () => Promise<void>, push: AgentPushCoordinator, readListeners: Set<(state: ReadState) => void>) {
|
|
const sockets = new WebSocketServer({ noServer: true, maxPayload: 16 * 1024 });
|
|
const cleanup = new Set<() => void>();
|
|
const upgrade = (request: IncomingMessage, socket: Duplex, head: Buffer) => {
|
|
let url: URL;
|
|
try { url = new URL(request.url || '/', 'http://localhost'); }
|
|
catch { socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n'); return; }
|
|
if (url.pathname !== '/v1/agent/ws') { socket.end('HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n'); return; }
|
|
// Credentials must never enter URLs and access logs.
|
|
if (url.search) { socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n'); return; }
|
|
sockets.handleUpgrade(request, socket, head, browser => {
|
|
let upstream: WebSocket | undefined;
|
|
const connectionId = randomUUID();
|
|
const sendRead = (state: ReadState) => send({ type: 'read_state', ...state });
|
|
let authenticated = false, authenticating = false, disposed = false, alive = true;
|
|
const send = (value: unknown) => {
|
|
if (browser.readyState !== WebSocket.OPEN) return;
|
|
if (browser.bufferedAmount > 2 * 1024 * 1024) { browser.close(1013, 'Slow consumer; reconnect with cursor'); return; }
|
|
browser.send(typeof value === 'string' ? value : JSON.stringify(value));
|
|
};
|
|
const authTimeout = setTimeout(() => browser.close(4401, 'Authentication required'), 5000);
|
|
const heartbeat = setInterval(() => { if (!alive) browser.terminate(); else { alive = false; browser.ping(); } }, 25_000);
|
|
browser.on('pong', () => { alive = true; if (authenticated && env.ADMIN_TOKEN) void push.heartbeat(connectionId).catch(() => {}); });
|
|
const dispose = () => {
|
|
if (disposed) return;
|
|
disposed = true; clearTimeout(authTimeout); clearInterval(heartbeat); cleanup.delete(dispose);
|
|
readListeners.delete(sendRead);
|
|
void push.disconnect(connectionId).catch(() => {});
|
|
upstream?.terminate(); browser.terminate();
|
|
};
|
|
cleanup.add(dispose);
|
|
browser.on('close', dispose); browser.on('error', dispose);
|
|
browser.on('message', async (raw, isBinary) => {
|
|
try {
|
|
if (isBinary) throw new Error('JSON text frames are required');
|
|
const frame = JSON.parse(raw.toString());
|
|
if (!authenticated) {
|
|
if (authenticating) { browser.close(4400, 'Authentication is still pending'); return; }
|
|
authenticating = true;
|
|
const auth = parse(AuthFrame, frame);
|
|
if (env.ADMIN_TOKEN && auth.token !== env.ADMIN_TOKEN) { browser.close(4401, 'Invalid token'); return; }
|
|
if (!apiKey) { browser.close(4503, 'Sybil Agent is not configured'); return; }
|
|
try { await guardOpenProxy(); } catch { browser.close(4503, 'Authenticated Sybil access is required'); return; }
|
|
if (disposed || browser.readyState !== WebSocket.OPEN) return;
|
|
authenticated = true; clearTimeout(authTimeout);
|
|
readListeners.add(sendRead);
|
|
if (env.ADMIN_TOKEN) await push.connect(connectionId, auth.clientKind, auth.presenceSessionId).catch(() => {});
|
|
if (disposed) { void push.disconnect(connectionId).catch(() => {}); return; }
|
|
const target = new URL('/api/ws', baseUrl); target.protocol = target.protocol === 'https:' ? 'wss:' : 'ws:';
|
|
upstream = new WebSocket(target, { headers: { 'X-Sybil-Client': 'backend-proxy-v1', Authorization: `Bearer ${apiKey}` }, handshakeTimeout: 10_000, maxPayload: 16 * 1024 * 1024 });
|
|
upstream.on('open', () => upstream!.send(JSON.stringify({ type: 'subscribe', ...(auth.after === undefined ? {} : { after: auth.after }) })));
|
|
upstream.on('message', (data, binary) => {
|
|
if (binary) { browser.close(1011, 'Invalid Sybil Agent frame'); return; }
|
|
send(data.toString());
|
|
});
|
|
upstream.on('close', () => { if (!disposed) browser.close(1012, 'Sybil Agent disconnected; reconnect with cursor'); });
|
|
upstream.on('error', () => { send({ type: 'error', error: 'Sybil Agent is unavailable' }); browser.close(1011, 'Sybil Agent is unavailable'); });
|
|
return;
|
|
}
|
|
if (frame.type === 'ping') { send({ type: 'pong' }); return; }
|
|
const subscribe = parse(SubscribeFrame, frame);
|
|
if (upstream?.readyState !== WebSocket.OPEN) { send({ type: 'error', error: 'Sybil Agent is still connecting' }); return; }
|
|
upstream.send(JSON.stringify(subscribe));
|
|
} catch (error) {
|
|
if (!authenticated) browser.close(4400, 'Invalid authentication frame');
|
|
else send({ type: 'error', error: error instanceof Error ? error.message : 'Invalid frame' });
|
|
}
|
|
});
|
|
});
|
|
};
|
|
app.server.on('upgrade', upgrade);
|
|
app.addHook('preClose', async () => {
|
|
app.server.off('upgrade', upgrade);
|
|
for (const stop of cleanup) stop();
|
|
for (const client of sockets.clients) client.terminate();
|
|
sockets.close();
|
|
});
|
|
}
|