stack/packages/plugins/model-provider/fixture/endpoint.mjs

// 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); });