// Loopback endpoint for the fixture provider. It answers the Anthropic // messages API with FIXTURE_WORD for FIXTURE_MODEL, only on 127.0.0.1, only // with the published fixture key, and forwards nothing anywhere. // Run as a process: prints one JSON line {"ready":true,"port":N} and serves // until SIGTERM/SIGINT or its lifetime (MOSAIC_FIXTURE_TTL_S) ends. import { createServer } from 'node:http'; import { fileURLToPath } from 'node:url'; import { FIXTURE_KEY, FIXTURE_MODEL, FIXTURE_WORD } from './index.mjs'; export const HOST = '127.0.0.1'; const BODY_MAX = 1024 * 1024; export const TTL_DEFAULT_S = 14400; export const TTL_MIN_S = 60; // The longest delay a Node timer holds; a longer one fires after 1 ms. export const TTL_MAX_S = 2147483; // Seconds the process lives, from MOSAIC_FIXTURE_TTL_S. Unset, empty or not a number is the // default; 0 is no expiry (`mosaic dev fixture init --ttl 0`, a named dev instance only); // anything else is held to [TTL_MIN_S, TTL_MAX_S]. export function lifetimeSeconds(raw) { const n = raw === undefined || raw === '' ? TTL_DEFAULT_S : Number(raw); if (!Number.isFinite(n)) return TTL_DEFAULT_S; return n === 0 ? 0 : Math.min(TTL_MAX_S, Math.max(TTL_MIN_S, n)); } export function createEndpoint({ log = () => {}, ttlSeconds = null, expiresAt = null } = {}) { const counts = { messages: 0, refused: 0 }; const server = createServer((req, res) => { const reply = (status, body, type = 'application/json') => { res.writeHead(status, { 'content-type': type }); res.end(typeof body === 'string' ? body : JSON.stringify(body)); }; const error = (status, type, message) => { counts.refused++; log(`${req.method} ${req.url} ${status} ${type}`); reply(status, { type: 'error', error: { type, message } }); }; if (req.method === 'GET' && req.url === '/health') return reply(200, { ok: true, model: FIXTURE_MODEL, upstream: 'none', counts, ttlSeconds, expiresAt }); if (req.method !== 'POST' || !/^\/v1\/messages(\?|$)/.test(req.url ?? '')) return error(404, 'not_found_error', 'fixture serves POST /v1/messages only'); if (req.headers['x-api-key'] !== FIXTURE_KEY) return error(401, 'authentication_error', 'fixture accepts only the fixture key'); let size = 0; const chunks = []; req.on('data', c => { size += c.length; if (size > BODY_MAX) req.destroy(); else chunks.push(c); }); req.on('end', () => { let body; try { body = JSON.parse(Buffer.concat(chunks).toString('utf8')); } catch { return error(400, 'invalid_request_error', 'body is not JSON'); } if (body?.model !== FIXTURE_MODEL) return error(404, 'not_found_error', `fixture serves ${FIXTURE_MODEL} only`); counts.messages++; log(`POST /v1/messages 200 ${FIXTURE_MODEL}`); const message = { id: `msg_fixture_${counts.messages}`, type: 'message', role: 'assistant', model: FIXTURE_MODEL, stop_sequence: null }; const usage = { input_tokens: 1, output_tokens: 1 }; if (body.stream !== true) return reply(200, { ...message, content: [{ type: 'text', text: FIXTURE_WORD }], stop_reason: 'end_turn', usage }); res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' }); const event = (type, data) => res.write(`event: ${type}\ndata: ${JSON.stringify({ type, ...data })}\n\n`); event('message_start', { message: { ...message, content: [], stop_reason: null, usage: { input_tokens: 1, output_tokens: 0 } } }); event('content_block_start', { index: 0, content_block: { type: 'text', text: '' } }); event('content_block_delta', { index: 0, delta: { type: 'text_delta', text: FIXTURE_WORD } }); event('content_block_stop', { index: 0 }); event('message_delta', { delta: { stop_reason: 'end_turn', stop_sequence: null }, usage: { output_tokens: 1 } }); event('message_stop', {}); res.end(); }); }); return { server, counts, listen: (port = 0) => new Promise((resolve, reject) => { server.once('error', reject); server.listen(port, HOST, () => { const a = server.address(); if (a.address !== HOST) { server.close(); reject(new Error('endpoint is not on loopback')); } else resolve({ port: a.port, baseUrl: `http://${HOST}:${a.port}` }); }); }), close: () => new Promise(resolve => { server.closeAllConnections?.(); server.close(() => resolve()); }), }; } // Listens on loopback and runs `expire` once ttlSeconds pass; ttlSeconds 0 never expires. export async function serve({ ttlSeconds, log, expire }) { const expiresAt = ttlSeconds ? new Date(Date.now() + ttlSeconds * 1000).toISOString() : null; const endpoint = createEndpoint({ log, ttlSeconds, expiresAt }); const address = await endpoint.listen(0); // A bounded lifetime, so an endpoint a crashed caller left behind still ends. const timer = ttlSeconds ? setTimeout(expire, ttlSeconds * 1000) : null; timer?.unref(); return { endpoint, ...address, expiresAt, timer }; } async function main() { let endpoint; // Expiry and SIGTERM end the process the same way, leaving any record its caller wrote. const stop = async () => { await endpoint?.close(); process.exit(0); }; const served = await serve({ ttlSeconds: lifetimeSeconds(process.env.MOSAIC_FIXTURE_TTL_S), log: line => process.stderr.write(`${new Date().toISOString()} ${line}\n`), expire: stop }); endpoint = served.endpoint; process.stdout.write(`${JSON.stringify({ ready: true, port: served.port, pid: process.pid })}\n`); process.on('SIGTERM', stop); process.on('SIGINT', stop); } if (process.argv[1] && fileURLToPath(import.meta.url) === process.argv[1]) main().catch(e => { process.stderr.write(`fixture endpoint: ${e.message}\n`); process.exit(1); });