spb/claude-remote

1/** Cluster endpoints + local model probes. No model is ever invented: the list is CR_LOCAL_MODELS. */2import type { FastifyInstance } from 'fastify';3import { z } from 'zod';4import { ClusterNode, ClusterOverview, LocalModel, type Health } from '@cr/shared';5import { authOf, notFound, parse, type Ctx } from '../../http.js';67const NodeParams = z.object({ id: z.string().min(1).max(40).regex(/^[A-Za-z0-9._-]+$/) });8const MetricsQuery = z.object({9  node: z.string().max(40).regex(/^[A-Za-z0-9._-]+$/).optional(),10  minutes: z.coerce.number().int().min(1).max(14 * 24 * 60).default(60),11});12const MetricPoint = z.object({13  at: z.string(), online: z.boolean(), cpuPct: z.number().nullable(), memUsedMb: z.number().nullable(),14  diskFreeGb: z.number().nullable(), load1: z.number().nullable(),15});16const NodeDetail = z.object({ node: ClusterNode, history: z.array(MetricPoint) });17const MetricsResponse = z.object({ since: z.string(), series: z.record(z.string(), z.array(MetricPoint)) });18const ModelTest = z.object({19  ok: z.boolean(), checkedAt: z.string(), latencyMs: z.number().nullable(), tokensPerSec: z.number().nullable(), error: z.string().nullable(),20});2122interface MetricRow {23  node_id: string;24  at: Date;25  online: boolean;26  cpu_pct: number | null;27  mem_used_mb: number | null;28  disk_free_gb: number | null;29  load1: number | null;30}31const point = (r: MetricRow): z.infer<typeof MetricPoint> => ({32  at: r.at.toISOString(), online: r.online, cpuPct: r.cpu_pct, memUsedMb: r.mem_used_mb, diskFreeGb: r.disk_free_gb, load1: r.load1,33});3435export default async function clusterPlugin(app: FastifyInstance, ctx: Ctx): Promise<void> {36  const { db, config, audit } = ctx;37  const probes = new Map<string, { health: Health; checkedAt: string; latencyMs: number | null; tokensPerSec: number | null }>();3839  app.get('/cluster/nodes', { config: { doc: { summary: 'Cluster overview (agentd, or host only)', response: ClusterOverview, responseName: 'ClusterOverview' } } }, () => ctx.cluster.overview());4041  app.get('/cluster/nodes/:id', { config: { doc: { summary: 'One node + its recorded metrics (24 h)', response: NodeDetail, responseName: 'ClusterNodeDetail' } } }, async (req) => {42    const { id } = parse(NodeParams, req.params);43    const node = (await ctx.cluster.overview()).nodes.find((n) => n.id === id);44    if (!node) throw notFound('node');45    const rows = await db.q<MetricRow>("SELECT * FROM node_metrics WHERE node_id = $1 AND at > now() - interval '24 hours' ORDER BY at", [id]);46    return { node, history: rows.map(point) };47  });4849  app.get('/cluster/metrics', { config: { doc: { summary: 'Recorded per-node metrics (1/min)', query: MetricsQuery, response: MetricsResponse, responseName: 'ClusterMetrics' } } }, async (req) => {50    const q = parse(MetricsQuery, req.query);51    const since = new Date(ctx.now() - q.minutes * 60_000);52    const rows = await db.q<MetricRow>(53      'SELECT * FROM node_metrics WHERE at > $1 AND ($2::text IS NULL OR node_id = $2) ORDER BY at LIMIT 50000', [since, q.node ?? null],54    );55    const series: Record<string, z.infer<typeof MetricPoint>[]> = {};56    for (const r of rows) (series[r.node_id] ??= []).push(point(r));57    return { since: since.toISOString(), series };58  });5960  /** Health = a real GET `${endpoint}/v1/models` (5 s). Results are cached 30 s. */61  async function probe(m: (typeof config.localModels)[number]): Promise<LocalModel> {62    const hit = probes.get(m.id);63    let p = hit && ctx.now() - Date.parse(hit.checkedAt) < 30_000 ? hit : null;64    if (!p) {65      let health: Health = 'OFFLINE';66      try {67        const res = await ctx.fetch(`${m.endpoint.replace(/\/$/, '')}/v1/models`, { signal: AbortSignal.timeout(5000) });68        health = res.ok ? 'ONLINE' : 'DEGRADED';69        await res.body?.cancel().catch(() => undefined);70      } catch {71        health = 'OFFLINE';72      }73      p = { health, checkedAt: new Date(ctx.now()).toISOString(), latencyMs: hit?.latencyMs ?? null, tokensPerSec: hit?.tokensPerSec ?? null };74      probes.set(m.id, p);75    }76    // `endpoint` is reduced to its host label: internal URLs are not sent to the browser.77    let label = m.host;78    try {79      label = new URL(m.endpoint).port ? `${m.host}:${new URL(m.endpoint).port}` : m.host;80    } catch {81      /* keep host */82    }83    return { id: m.id, name: m.name, host: m.host, engine: m.engine, endpoint: label, ...p };84  }8586  app.get('/cluster/models', { config: { doc: { summary: 'Configured local models with probed health', response: z.array(LocalModel), responseName: 'LocalModelList' } } }, () =>87    Promise.all(config.localModels.map(probe)),88  );8990  app.post('/cluster/models/:id/test', { config: { doc: { summary: 'Real small chat completion: latency and tokens/s when measurable', response: ModelTest, responseName: 'LocalModelTest' } } }, async (req) => {91    authOf(req);92    const { id } = parse(z.object({ id: z.string().min(1).max(80) }), req.params);93    const m = config.localModels.find((x) => x.id === id);94    if (!m) throw notFound('model');95    const started = performance.now();96    let ok = false;97    let latencyMs: number | null = null;98    let tokensPerSec: number | null = null;99    let error: string | null = null;100    try {101      const res = await ctx.fetch(`${m.endpoint.replace(/\/$/, '')}/v1/chat/completions`, {102        method: 'POST',103        headers: { 'content-type': 'application/json' },104        body: JSON.stringify({ model: m.id, messages: [{ role: 'user', content: 'Reply with the single word: ok' }], max_tokens: 32, stream: false }),105        signal: AbortSignal.timeout(60_000),106      });107      latencyMs = Math.round(performance.now() - started);108      if (res.ok) {109        const body = (await res.json()) as { usage?: { completion_tokens?: unknown } };110        ok = true;111        const tokens = body.usage?.completion_tokens;112        // Only when the server reports usage; otherwise it is not measurable and stays null.113        if (typeof tokens === 'number' && tokens > 0 && latencyMs > 0) tokensPerSec = Math.round((tokens / (latencyMs / 1000)) * 10) / 10;114      } else {115        error = `HTTP ${res.status}`;116        await res.body?.cancel().catch(() => undefined);117      }118    } catch (err) {119      error = (err as Error).name === 'TimeoutError' ? 'timeout' : 'unreachable';120    }121    const checkedAt = new Date(ctx.now()).toISOString();122    probes.set(m.id, { health: ok ? 'ONLINE' : error?.startsWith('HTTP') ? 'DEGRADED' : 'OFFLINE', checkedAt, latencyMs: ok ? latencyMs : null, tokensPerSec });123    await audit.req(req, 'cluster.model.test', { target: m.id, ok });124    return { ok, checkedAt, latencyMs: ok ? latencyMs : null, tokensPerSec, error };125  });126}127