feat(sites): per-site model probe with streaming logs and auto-disable

- Move post-refresh probe settings from global config to per-site (sites table) - Add manual 'probe now' button in site editor with real-time SSE log stream - Use fetch+ReadableStream instead of EventSource to support Authorization header - Concurrent probe execution (default 10 workers) via worker-pool pattern - Add stop button to cancel in-flight probe while preserving completed results - Add latency threshold: auto-disable models exceeding response time limit - Treat inconclusive probe results as unsupported (auto-disable) - Show specific failure reason in logs (timeout, no token, model not found, etc.) - Refresh and display post-probe model status list (green=available, red=disabled) - Remove probe settings from global Settings page and RuntimeSettingsPayload
2026-04-23 21:42:48 +08:00
parent 262651a708
commit 46f23ddbb3
7 changed files with 669 additions and 0 deletions
@@ -15,6 +15,9 @@ export const sites = sqliteTable('sites', {
  sortOrder: integer('sort_order').default(0),
  globalWeight: real('global_weight').default(1),
  apiKey: text('api_key'),
+  postRefreshProbeEnabled: integer('post_refresh_probe_enabled', { mode: 'boolean' }).default(false),
+  postRefreshProbeModel: text('post_refresh_probe_model').default(''),
+  postRefreshProbeScope: text('post_refresh_probe_scope').default('single'),
  createdAt: text('created_at').default(sql`(datetime('now'))`),
  updatedAt: text('updated_at').default(sql`(datetime('now'))`),
 }, (table) => ({
@@ -702,6 +702,10 @@ function applyImportedSettingToRuntime(key: string, value: unknown) {
      config.tokenRouterFailureCooldownMaxSec = normalized;
      return;
    }
+    case 'post_refresh_probe_enabled':
+    case 'post_refresh_probe_model':
+    case 'post_refresh_probe_scope':
+      return;
    default:
      return;
  }
@@ -18,6 +18,11 @@ import {
 import { getSiteInitializationPreset } from '../../../shared/siteInitializationPresets.js';
 import { normalizeSiteApiEndpointBaseUrl } from '../../services/siteApiEndpointService.js';
 import { analyzePrimarySiteUrl } from '../../../shared/sitePrimaryUrl.js';
+import { probeSiteModels } from '../../services/modelService.js';
+
+function sseWrite(raw: import('http').ServerResponse, event: string, data: unknown) {
+  try { raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); } catch { /* ignore */ }
+}

 function normalizeSiteStatus(input: unknown): 'active' | 'disabled' | null {
  if (input === undefined || input === null) return null;
@@ -683,6 +688,10 @@ export async function sitesRoutes(app: FastifyInstance) {
    if (body.isPinned !== undefined) updates.isPinned = normalizedPinned;
    if (body.sortOrder !== undefined) updates.sortOrder = normalizedSortOrder;
    if (body.globalWeight !== undefined) updates.globalWeight = normalizedGlobalWeight;
+    const anyBody = body as Record<string, unknown>;
+    if (anyBody.postRefreshProbeEnabled !== undefined) updates.postRefreshProbeEnabled = Boolean(anyBody.postRefreshProbeEnabled);
+    if (anyBody.postRefreshProbeModel !== undefined) updates.postRefreshProbeModel = String(anyBody.postRefreshProbeModel || '').trim();
+    if (anyBody.postRefreshProbeScope !== undefined) updates.postRefreshProbeScope = anyBody.postRefreshProbeScope === 'all' ? 'all' : 'single';
    updates.updatedAt = new Date().toISOString();
    try {
      await db.transaction(async (tx) => {
@@ -887,6 +896,58 @@ export async function sitesRoutes(app: FastifyInstance) {
    return { siteId: id, models };
  });

+  // Manually probe site models now (one-shot JSON)
+  app.post<{ Params: { id: string }; Body: unknown }>('/api/sites/:id/probe-now', async (request, reply) => {
+    const id = parseInt(request.params.id);
+    if (Number.isNaN(id)) {
+      return reply.code(400).send({ error: 'Invalid site id' });
+    }
+    const body = request.body as Record<string, unknown> | null;
+    const scope = body?.scope === 'all' ? 'all' : body?.scope === 'single' ? 'single' : undefined;
+    const modelName = typeof body?.modelName === 'string' ? body.modelName.trim() : undefined;
+    const result = await probeSiteModels(id, { scope, modelName });
+    if (!result.success) {
+      return reply.code(422).send({ error: result.error });
+    }
+    return result;
+  });
+
+  // Streaming probe via SSE
+  app.get<{ Params: { id: string }; Querystring: { scope?: string; modelName?: string; latencyThresholdMs?: string } }>(
+    '/api/sites/:id/probe-stream',
+    async (request, reply) => {
+      reply.hijack();
+      reply.raw.writeHead(200, {
+        'Content-Type': 'text/event-stream',
+        'Cache-Control': 'no-cache',
+        'Connection': 'keep-alive',
+      });
+
+      const id = parseInt(request.params.id);
+      if (Number.isNaN(id)) {
+        sseWrite(reply.raw, 'error', { message: 'Invalid site id' });
+        reply.raw.end();
+        return;
+      }
+
+      const q = request.query;
+      const scope = q.scope === 'all' ? 'all' : q.scope === 'single' ? 'single' : undefined;
+      const modelName = q.modelName?.trim() || undefined;
+      const parsedThreshold = parseInt(q.latencyThresholdMs ?? '', 10);
+      const latencyThresholdMs = Number.isFinite(parsedThreshold) && parsedThreshold > 0 ? parsedThreshold : undefined;
+
+      try {
+        const result = await probeSiteModels(id, { scope, modelName, latencyThresholdMs }, (ev) => {
+          sseWrite(reply.raw, ev.type, ev);
+        });
+        sseWrite(reply.raw, 'complete', result);
+      } catch (err: any) {
+        sseWrite(reply.raw, 'error', { message: err?.message || '探测失败' });
+      }
+      reply.raw.end();
+    },
+  );
+
  // Detect platform for a URL
  app.post<{ Body: unknown }>('/api/sites/detect', async (request, reply) => {
    const parsedBody = parseSiteDetectPayload(request.body);
@@ -758,6 +758,9 @@ function buildAllApiHubV2AccountsSection(data: RawBackupData): {
      sortOrder: section.sites.length,
      globalWeight: 1,
      apiKey: null,
+      postRefreshProbeEnabled: false,
+      postRefreshProbeModel: '',
+      postRefreshProbeScope: 'single',
      createdAt: input.createdAt,
      updatedAt: input.updatedAt,
    });
@@ -1001,6 +1004,9 @@ function buildAccountsSectionFromRefBackup(data: RawBackupData): AccountsBackupS
        sortOrder: sites.length,
        globalWeight: 1,
        apiKey: null,
+        postRefreshProbeEnabled: false,
+        postRefreshProbeModel: '',
+        postRefreshProbeScope: 'single',
        createdAt: toIsoString(item.created_at),
        updatedAt: toIsoString(item.updated_at),
      });
@@ -34,6 +34,7 @@ import {
  discoverCodexModelsFromCloud,
  validateGeminiCliOauthConnection,
 } from './platformDiscoveryRegistry.js';
+import { probeRuntimeModel, type RuntimeModelProbeStatus } from './runtimeModelProbe.js';

 const API_TOKEN_DISCOVERY_TIMEOUT_MS = 8_000;
 const MODEL_DISCOVERY_TIMEOUT_MS = 12_000;
@@ -101,6 +102,16 @@ export type ModelRefreshSuccessResult = {
  tokenScanned: number;
  discoveredByCredential: boolean;
  discoveredApiToken: boolean;
+  postProbeResult?: {
+    scope: 'single' | 'all';
+    probed: number;
+    unsupported: number;
+    details: Array<{
+      modelName: string;
+      status: RuntimeModelProbeStatus;
+      latencyMs: number | null;
+    }>;
+  };
 };

 export type ModelRefreshResult =
@@ -310,6 +321,7 @@ function buildSuccessfulRefreshResult(input: {
  tokenScanned: number;
  discoveredByCredential: boolean;
  discoveredApiToken: boolean;
+  postProbeResult?: ModelRefreshSuccessResult['postProbeResult'];
 }): ModelRefreshSuccessResult {
  return {
    accountId: input.accountId,
@@ -322,6 +334,7 @@ function buildSuccessfulRefreshResult(input: {
    tokenScanned: input.tokenScanned,
    discoveredByCredential: input.discoveredByCredential,
    discoveredApiToken: input.discoveredApiToken,
+    postProbeResult: input.postProbeResult,
  };
 }

@@ -368,6 +381,215 @@ async function retryOauthModelDiscoveryWithRefresh<T>(input: {
  }
 }

+export type ProbeSiteModelsResult = {
+  success: boolean;
+  error?: string;
+  scope: 'single' | 'all';
+  probed: number;
+  unsupported: number;
+  details: Array<{ modelName: string; status: RuntimeModelProbeStatus; latencyMs: number | null; reason?: string }>;
+};
+
+export type ProbeSiteModelsProgress =
+  | { type: 'start'; scope: 'single' | 'all'; modelsCount: number; modelsToProbe: string[] }
+  | { type: 'model'; modelName: string; status: RuntimeModelProbeStatus; latencyMs: number | null; latencyExceeded?: true; reason?: string }
+  | { type: 'action'; modelName: string; action: 'disabled' };
+
+export async function probeSiteModels(
+  siteId: number,
+  options?: { scope?: 'single' | 'all'; modelName?: string; concurrency?: number; latencyThresholdMs?: number },
+  onProgress?: (event: ProbeSiteModelsProgress) => void,
+): Promise<ProbeSiteModelsResult> {
+  const empty = (scope: 'single' | 'all', error: string): ProbeSiteModelsResult =>
+    ({ success: false, error, scope, probed: 0, unsupported: 0, details: [] });
+
+  const site = await db.select().from(schema.sites).where(eq(schema.sites.id, siteId)).get();
+  if (!site) return empty('single', '站点不存在');
+
+  const account = await db.select().from(schema.accounts)
+    .where(and(eq(schema.accounts.siteId, siteId), eq(schema.accounts.status, 'active')))
+    .get();
+  if (!account) return empty('single', '该站点没有可用的活跃账号');
+
+  const modelRows = await db.select({ modelName: schema.modelAvailability.modelName })
+    .from(schema.modelAvailability)
+    .where(and(
+      eq(schema.modelAvailability.accountId, account.id),
+      eq(schema.modelAvailability.available, true),
+    ))
+    .all();
+
+  const scope = (options?.scope ?? (site.postRefreshProbeScope === 'all' ? 'all' : 'single')) as 'single' | 'all';
+  const availableModels = modelRows.map((r) => r.modelName.trim()).filter((m) => m.length > 0);
+  if (availableModels.length === 0) {
+    return empty(scope, '该站点暂无已发现模型，请先刷新模型列表');
+  }
+
+  let modelsToProbe: string[];
+  if (scope === 'all') {
+    modelsToProbe = availableModels;
+  } else {
+    const configModel = ((options?.modelName ?? site.postRefreshProbeModel) || '').trim().toLowerCase();
+    const found = configModel
+      ? (availableModels.find((m) => m.toLowerCase() === configModel) ?? availableModels[0])
+      : availableModels[0];
+    modelsToProbe = [found];
+  }
+
+  onProgress?.({ type: 'start', scope, modelsCount: modelsToProbe.length, modelsToProbe });
+
+  // Probe models concurrently, limited by modelAvailabilityProbeConcurrency
+  const concurrency = Math.max(1, options?.concurrency ?? 10);
+  const detailsMap = new Map<string, { modelName: string; status: RuntimeModelProbeStatus; latencyMs: number | null }>();
+
+  let cursor = 0;
+  async function worker() {
+    while (cursor < modelsToProbe.length) {
+      const modelName = modelsToProbe[cursor++];
+      try {
+        const result = await probeRuntimeModel({
+          site, account, modelName, timeoutMs: config.modelAvailabilityProbeTimeoutMs,
+        });
+        const threshold = options?.latencyThresholdMs ?? 0;
+        const latencyExceeded = (
+          result.status === 'supported'
+          && threshold > 0
+          && result.latencyMs != null
+          && result.latencyMs > threshold
+        );
+        const effectiveStatus: RuntimeModelProbeStatus = latencyExceeded ? 'unsupported' : result.status;
+        const effectiveReason = latencyExceeded
+          ? `响应延迟 ${result.latencyMs}ms 超过阈值 ${threshold}ms`
+          : result.reason;
+        detailsMap.set(modelName, { modelName, status: effectiveStatus, latencyMs: result.latencyMs, reason: effectiveReason });
+        onProgress?.(latencyExceeded
+          ? { type: 'model', modelName, status: effectiveStatus, latencyMs: result.latencyMs, latencyExceeded: true, reason: effectiveReason }
+          : { type: 'model', modelName, status: effectiveStatus, latencyMs: result.latencyMs, reason: effectiveReason },
+        );
+      } catch (err) {
+        const errReason = err instanceof Error ? err.message : '探测异常';
+        console.warn(`[probe-site-now] probe failed for site ${siteId} model ${modelName}`, err);
+        detailsMap.set(modelName, { modelName, status: 'inconclusive', latencyMs: null, reason: errReason });
+        onProgress?.({ type: 'model', modelName, status: 'inconclusive', latencyMs: null, reason: errReason });
+      }
+    }
+  }
+  await Promise.all(Array.from({ length: Math.min(concurrency, modelsToProbe.length) }, worker));
+
+  // Restore original model order for the final details list
+  const details = modelsToProbe.map((m) => detailsMap.get(m)!);
+
+  const unsupportedModels = details.filter((d) => d.status === 'unsupported' || d.status === 'inconclusive').map((d) => d.modelName);
+  if (unsupportedModels.length > 0) {
+    const checkedAt = new Date().toISOString();
+    for (const modelName of unsupportedModels) {
+      await db.update(schema.modelAvailability)
+        .set({ available: false, checkedAt })
+        .where(and(
+          eq(schema.modelAvailability.accountId, account.id),
+          eq(schema.modelAvailability.modelName, modelName),
+        ))
+        .run();
+      await db.insert(schema.siteDisabledModels)
+        .values({ siteId, modelName })
+        .onConflictDoNothing()
+        .run();
+      onProgress?.({ type: 'action', modelName, action: 'disabled' });
+    }
+    const reason = unsupportedModels.length === 1
+      ? `手动探测失败：模型 ${unsupportedModels[0]} 不可用`
+      : `手动探测失败：${unsupportedModels.length} 个模型不可用（${unsupportedModels.slice(0, 3).join('、')}${unsupportedModels.length > 3 ? '…' : ''}）`;
+    await setAccountRuntimeHealth(account.id, { state: 'unhealthy', reason, source: 'manual-probe', checkedAt });
+    rebuildTokenRoutesFromAvailability().catch((err) => {
+      console.warn('[probe-site-now] route rebuild failed', err);
+    });
+  }
+
+  return { success: true, scope, probed: details.length, unsupported: unsupportedModels.length, details };
+}
+
+async function runPostRefreshProbeIfEnabled(params: {
+  account: typeof schema.accounts.$inferSelect;
+  site: typeof schema.sites.$inferSelect;
+  discoveredModels: string[];
+}): Promise<ModelRefreshSuccessResult['postProbeResult']> {
+  if (!params.site.postRefreshProbeEnabled) return undefined;
+  if (params.discoveredModels.length === 0) return undefined;
+
+  const scope = (params.site.postRefreshProbeScope === 'all' ? 'all' : 'single') as 'single' | 'all';
+
+  // Determine which models to probe
+  let modelsToProbe: string[];
+  if (scope === 'all') {
+    modelsToProbe = params.discoveredModels;
+  } else {
+    const configModel = (params.site.postRefreshProbeModel || '').trim().toLowerCase();
+    const found = configModel
+      ? (params.discoveredModels.find((m) => m.toLowerCase() === configModel) ?? params.discoveredModels[0])
+      : params.discoveredModels[0];
+    modelsToProbe = [found];
+  }
+
+  // Probe each model sequentially
+  const details: Array<{ modelName: string; status: RuntimeModelProbeStatus; latencyMs: number | null }> = [];
+  for (const modelName of modelsToProbe) {
+    try {
+      const result = await probeRuntimeModel({
+        site: params.site,
+        account: params.account,
+        modelName,
+        timeoutMs: config.modelAvailabilityProbeTimeoutMs,
+      });
+      details.push({ modelName, status: result.status, latencyMs: result.latencyMs });
+    } catch (err) {
+      console.warn(`[post-refresh-probe] probe failed for account ${params.account.id} model ${modelName}`, err);
+      details.push({ modelName, status: 'inconclusive', latencyMs: null });
+    }
+  }
+
+  // Handle unsupported models
+  const unsupportedModels = details.filter((d) => d.status === 'unsupported' || d.status === 'inconclusive').map((d) => d.modelName);
+  if (unsupportedModels.length > 0) {
+    const checkedAt = new Date().toISOString();
+    for (const modelName of unsupportedModels) {
+      // Mark model as unavailable
+      await db.update(schema.modelAvailability)
+        .set({ available: false, checkedAt })
+        .where(and(
+          eq(schema.modelAvailability.accountId, params.account.id),
+          eq(schema.modelAvailability.modelName, modelName),
+        ))
+        .run();
+      // Add to site-level disabled models
+      await db.insert(schema.siteDisabledModels)
+        .values({ siteId: params.site.id, modelName })
+        .onConflictDoNothing()
+        .run();
+    }
+    // Update account health
+    const reason = unsupportedModels.length === 1
+      ? `刷新后探测失败：模型 ${unsupportedModels[0]} 不可用`
+      : `刷新后探测失败：${unsupportedModels.length} 个模型不可用（${unsupportedModels.slice(0, 3).join('、')}${unsupportedModels.length > 3 ? '…' : ''}）`;
+    await setAccountRuntimeHealth(params.account.id, {
+      state: 'unhealthy',
+      reason,
+      source: 'post-refresh-probe',
+      checkedAt,
+    });
+    // Single route rebuild for all changes
+    rebuildTokenRoutesFromAvailability().catch((err) => {
+      console.warn('[post-refresh-probe] route rebuild failed', err);
+    });
+  }
+
+  return {
+    scope,
+    probed: details.length,
+    unsupported: unsupportedModels.length,
+    details,
+  };
+}
+
 export async function refreshModelsForAccount(
  accountId: number,
  options?: { allowInactive?: boolean },
@@ -510,6 +732,11 @@ export async function refreshModelsForAccount(
        source: 'model-discovery',
        checkedAt,
      });
+      const codexPostProbeResult = await runPostRefreshProbeIfEnabled({
+        account: discoveryAccount,
+        site,
+        discoveredModels: codexModels,
+      });
      return buildSuccessfulRefreshResult({
        accountId,
        modelCount: codexModels.length,
@@ -517,6 +744,7 @@ export async function refreshModelsForAccount(
        tokenScanned: 0,
        discoveredByCredential: true,
        discoveredApiToken: false,
+        postProbeResult: codexPostProbeResult,
      });
    } catch (err) {
      discoveryAccount = getRefreshedOauthAccountFromError(err) || discoveryAccount;
@@ -590,6 +818,11 @@ export async function refreshModelsForAccount(
        source: 'model-discovery',
        checkedAt,
      });
+      const claudePostProbeResult = await runPostRefreshProbeIfEnabled({
+        account: discoveryAccount,
+        site,
+        discoveredModels: claudeModels,
+      });
      return buildSuccessfulRefreshResult({
        accountId,
        modelCount: claudeModels.length,
@@ -597,6 +830,7 @@ export async function refreshModelsForAccount(
        tokenScanned: 0,
        discoveredByCredential: true,
        discoveredApiToken: false,
+        postProbeResult: claudePostProbeResult,
      });
    } catch (err) {
      discoveryAccount = getRefreshedOauthAccountFromError(err) || discoveryAccount;
@@ -684,6 +918,11 @@ export async function refreshModelsForAccount(
        source: 'model-discovery',
        checkedAt,
      });
+      const geminiPostProbeResult = await runPostRefreshProbeIfEnabled({
+        account: discoveryAccount,
+        site,
+        discoveredModels: GEMINI_CLI_STATIC_MODELS,
+      });
      return buildSuccessfulRefreshResult({
        accountId,
        modelCount: GEMINI_CLI_STATIC_MODELS.length,
@@ -691,6 +930,7 @@ export async function refreshModelsForAccount(
        tokenScanned: 0,
        discoveredByCredential: true,
        discoveredApiToken: false,
+        postProbeResult: geminiPostProbeResult,
      });
    } catch (err) {
      const rawMessage = (err as { message?: string })?.message || 'gemini cli oauth validation failed';
@@ -764,6 +1004,11 @@ export async function refreshModelsForAccount(
        source: 'model-discovery',
        checkedAt,
      });
+      const antigravityPostProbeResult = await runPostRefreshProbeIfEnabled({
+        account: discoveryAccount,
+        site,
+        discoveredModels: antigravityModels,
+      });
      return buildSuccessfulRefreshResult({
        accountId,
        modelCount: antigravityModels.length,
@@ -771,6 +1016,7 @@ export async function refreshModelsForAccount(
        tokenScanned: 0,
        discoveredByCredential: true,
        discoveredApiToken: false,
+        postProbeResult: antigravityPostProbeResult,
      });
    } catch (err) {
      discoveryAccount = getRefreshedOauthAccountFromError(err) || discoveryAccount;
@@ -1021,6 +1267,11 @@ export async function refreshModelsForAccount(
  });

  const modelsPreview = Array.from(accountModels.values()).slice(0, 10);
+  const standardPostProbeResult = await runPostRefreshProbeIfEnabled({
+    account,
+    site,
+    discoveredModels: Array.from(accountModels.values()),
+  });
  return buildSuccessfulRefreshResult({
    accountId,
    modelCount: accountModels.size,
@@ -1028,6 +1279,7 @@ export async function refreshModelsForAccount(
    tokenScanned: scannedTokenCount,
    discoveredByCredential,
    discoveredApiToken: !!discoveredApiToken,
+    postProbeResult: standardPostProbeResult,
  });
 }