From 18524c1c57341c1ffaa951e170ebc0cb4cd09b40 Mon Sep 17 00:00:00 2001 From: Michal Date: Tue, 21 Jul 2026 23:39:56 +0100 Subject: [PATCH] feat(chat): LLM failover chain + show which model answered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chat is LLM-essential but not model-specific — instead of failing when the pinned model is down/drifted, it now fails over across an ordered chain and reports which model actually answered. - Ordered fallback: an Llm declares `extraConfig.fallbacks: string[]`; the dispatcher builds primary-pool → fallback-pool(s) candidates and tries them in order (resolveCandidatesWithFallbacks). - Fail over on real failures, not just transport: runOneInference now advances on a non-2xx status (e.g. a 400 from a drifted model) or an empty/invalid completion, not only thrown transport errors. Streaming fails over pre-first-chunk (already threw on 4xx). - Transparency: ChatResult + the SSE `final` frame carry {llm, model, failedOver}; the CLI prints `model: ()` each turn and `⚠ failed over → answered by …` when a fallback was used. - Exhaustion names the last model + upstream body (not "no choice"). Tests: 3 failover unit tests (primary 400 → fallback answers + model reported; primary answers → failedOver=false; all fail → clear aggregated error). mcpd 948 + CLI 508 green; tsc + lint clean. docs/reliability.md updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/reliability.md | 26 +++++-- src/cli/src/commands/chat.ts | 30 ++++++-- src/mcpd/src/services/chat.service.ts | 101 +++++++++++++++++++++++--- src/mcpd/tests/chat-service.test.ts | 61 ++++++++++++++++ 4 files changed, 194 insertions(+), 24 deletions(-) diff --git a/docs/reliability.md b/docs/reliability.md index 22e8b49..60e3eca 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -28,10 +28,24 @@ Note: the gate's prompt-ranking uses the **heavy client provider's own model** it deliberately does *not* force the project's vLLM model onto it (doing so made every selection fail silently when the model wasn't anthropic-servable). -## LLM-*essential* operations +## LLM-*essential* operations — failover chain -Chat needs the LLM — it can't fall back. It must **fail fast with a clear, -actionable message** instead of a vague one. Chat surfaces the upstream status + -body (which names the model + reason), e.g. -`LLM returned no completion (HTTP 400): {"error":"model deepseek-v4-flash not found"}`, -rather than "Adapter returned no choice". +Chat needs *an* LLM but not a *specific* one. Instead of failing when the pinned +model is down, chat **fails over across an ordered chain** and **reports which +model actually answered**. + +- **Chain:** an `Llm` declares fallbacks in `extraConfig.fallbacks: string[]` + (Llm names, in order). The dispatcher builds an ordered candidate list — + the primary's pool, then each fallback's pool — and tries them in order. +- **Fails over on real failures**, not just transport: a non-2xx status + (e.g. a drifted model's `400`) or an empty/invalid completion now advances to + the next candidate (`chat.service.ts` `runOneInference`). Streaming fails over + pre-first-chunk. +- **Transparency:** `ChatResult` / the SSE `final` frame carry `llm`, `model`, + and `failedOver`. The CLI prints `model: ()` per turn, and + `⚠ failed over → answered by ()` when a fallback was used. +- **Exhaustion is clear:** if every candidate fails, the error names the last + model + upstream status/body (not "no choice"). + +Homelab chain: `vllm-current` (the served vLLM) → an `anthropic-fallback` server +Llm as the always-up last resort. diff --git a/src/cli/src/commands/chat.ts b/src/cli/src/commands/chat.ts index 3d04069..628642b 100644 --- a/src/cli/src/commands/chat.ts +++ b/src/cli/src/commands/chat.ts @@ -198,7 +198,8 @@ async function runOneShot( // `thread: ` — single space after the colon, matching the streaming // path (line 160 below) so any tooling/regex that watches one form picks // up the other too. - process.stderr.write(styleStats(`(${String(words)}w · ${(words / sec).toFixed(1)} w/s · ${sec.toFixed(1)}s)`) + ` thread: ${res.threadId}\n`); + const answered = formatAnswered(res.llm, res.model, res.failedOver); + process.stderr.write(styleStats(`(${String(words)}w · ${(words / sec).toFixed(1)} w/s · ${sec.toFixed(1)}s)`) + ` thread: ${res.threadId}${answered !== '' ? ` ${answered}` : ''}\n`); return; } const bar = installStatusBar(); @@ -258,6 +259,8 @@ async function runRepl( const res = await chatRequestNonStream(deps, subject, body); threadId = res.threadId; process.stdout.write(`${res.assistant}\n`); + const answered = formatAnswered(res.llm, res.model, res.failedOver); + if (answered !== '') process.stderr.write(`${styleStats(`(${answered})`)}\n`); } else { threadId = await streamOnce(deps, subject, line, threadId, overrides, bar); process.stdout.write('\n'); @@ -455,7 +458,7 @@ async function chatRequestNonStream( deps: ChatCommandDeps, subject: ChatSubject, body: Record, -): Promise<{ assistant: string; threadId: string; turnIndex: number }> { +): Promise<{ assistant: string; threadId: string; turnIndex: number; llm?: string; model?: string; failedOver?: boolean }> { const url = new URL(`${deps.baseUrl}/api/v1/${subject.path}/chat`); const payload = JSON.stringify(body); return new Promise((resolve, reject) => { @@ -482,7 +485,7 @@ async function chatRequestNonStream( return; } try { - resolve(JSON.parse(raw) as { assistant: string; threadId: string; turnIndex: number }); + resolve(JSON.parse(raw) as { assistant: string; threadId: string; turnIndex: number; llm?: string; model?: string; failedOver?: boolean }); } catch (err) { reject(new Error(`malformed response: ${(err as Error).message}`)); } @@ -550,6 +553,7 @@ async function streamOnce( } let buf = ''; let resolvedThread = threadId ?? ''; + let answered = ''; res.setEncoding('utf-8'); res.on('data', (chunk: string) => { buf += chunk; @@ -590,6 +594,7 @@ async function streamOnce( break; case 'final': if (evt.threadId !== undefined) resolvedThread = evt.threadId; + answered = formatAnswered(evt.llm, evt.model, evt.failedOver); break; case 'error': process.stderr.write(`\n[error: ${evt.message ?? ''}]\n`); @@ -606,10 +611,11 @@ async function streamOnce( const final = formatStats(stats, false); // Inline final-stats footer (permanent record): goes through the // scroll region, so a copy-pasted transcript captures it. - if (final !== '' && STDERR_IS_TTY) { - process.stderr.write(`\n${styleStats(`(${final})`)}`); - } else if (final !== '') { - process.stderr.write(`\n(${final})`); + const footer = [final, answered].filter((s) => s !== '').join(' · '); + if (footer !== '' && STDERR_IS_TTY) { + process.stderr.write(`\n${styleStats(`(${footer})`)}`); + } else if (footer !== '') { + process.stderr.write(`\n(${footer})`); } // Live status bar: pin the final value so it stays visible between // turns (answers "how fast was the last one?"). @@ -637,6 +643,16 @@ interface ChatStreamFrame { threadId?: string; turnIndex?: number; message?: string; + llm?: string; + model?: string; + failedOver?: boolean; +} + +/** One-line "which model answered" footer, dim; loud when it failed over. */ +function formatAnswered(llm?: string, model?: string, failedOver?: boolean): string { + if (llm === undefined && model === undefined) return ''; + const label = model !== undefined && model !== '' ? `${llm ?? '?'} (${model})` : (llm ?? '?'); + return failedOver === true ? `⚠ failed over → answered by ${label}` : `model: ${label}`; } // ANSI codes for the reasoning sidebar. Dim + italic visually separates diff --git a/src/mcpd/src/services/chat.service.ts b/src/mcpd/src/services/chat.service.ts index 16d8d07..c02e7f9 100644 --- a/src/mcpd/src/services/chat.service.ts +++ b/src/mcpd/src/services/chat.service.ts @@ -111,6 +111,10 @@ export interface ChatStreamChunk { threadId?: string; turnIndex?: number; message?: string; + /** On `final`: which Llm/model produced the reply, and whether it failed over. */ + llm?: string; + model?: string; + failedOver?: boolean; } /** @@ -151,6 +155,12 @@ export interface ChatResult { threadId: string; assistant: string; turnIndex: number; + /** Which Llm actually produced the reply (after any failover). */ + llm: string; + /** The model string that answered (e.g. 'qwen3-thinking' or 'claude-opus-4-...'). */ + model: string; + /** True when the reply came from a fallback rather than the primary Llm. */ + failedOver: boolean; } /** @@ -173,7 +183,12 @@ interface PreparedChatContext { threadId: string; history: OpenAiMessage[]; systemBlock: string; + /** Ordered dispatch list: the primary Llm's pool first, then each fallback + * Llm's pool (from the primary's extraConfig.fallbacks). The loop tries them + * in order and fails over on error/4xx/no-completion. */ poolCandidates: PoolCandidate[]; + /** Name of the intended primary Llm (to detect + report a failover). */ + primaryLlm: string; mergedParams: AgentChatParams; toolList: ChatTool[]; projectId: string | null; @@ -249,9 +264,11 @@ export class ChatService { private async runChatLoop(ctx: PreparedChatContext): Promise { let assistantFinal = ''; let lastTurnIndex = ctx.startingTurnIndex; + let winner: PoolCandidate | undefined; try { for (let i = 0; i < ctx.maxIterations; i += 1) { const result = await this.runOneInference(ctx); + winner = result.candidate; const choice = extractChoice(result.body); if (choice === null) { // Surface the upstream body (LiteLLM/vLLM names the model + reason, @@ -312,7 +329,14 @@ export class ChatService { assistantFinal = assistantText; lastTurnIndex = finalMsg.turnIndex; await this.chatRepo.touchThread(ctx.threadId); - return { threadId: ctx.threadId, assistant: assistantFinal, turnIndex: lastTurnIndex }; + return { + threadId: ctx.threadId, + assistant: assistantFinal, + turnIndex: lastTurnIndex, + llm: winner?.llmName ?? ctx.primaryLlm, + model: winner?.modelOverride ?? '', + failedOver: winner !== undefined && winner.llmName !== ctx.primaryLlm, + }; } throw new Error(`Chat loop exceeded ${String(ctx.maxIterations)} iterations without a terminal turn`); } catch (err) { @@ -333,6 +357,7 @@ export class ChatService { /** Shared streaming turn loop. Persists rows in lockstep. */ private async *runChatStreamLoop(ctx: PreparedChatContext): AsyncGenerator { + const winnerRef: { candidate: PoolCandidate | null } = { candidate: null }; try { for (let i = 0; i < ctx.maxIterations; i += 1) { // `reasoning` is accumulated alongside `content` so we can fall back @@ -348,7 +373,7 @@ export class ChatService { toolCalls: [], }; let finishReason: string | null = null; - for await (const chunk of this.streamInference(ctx)) { + for await (const chunk of this.streamInference(ctx, winnerRef)) { if (chunk.done === true) break; if (chunk.data === '[DONE]') break; const evt = parseStreamingChunk(chunk.data); @@ -452,7 +477,14 @@ export class ChatService { content: persistedContent, }); await this.chatRepo.touchThread(ctx.threadId); - yield { type: 'final', threadId: ctx.threadId, turnIndex: finalMsg.turnIndex }; + yield { + type: 'final', + threadId: ctx.threadId, + turnIndex: finalMsg.turnIndex, + llm: winnerRef.candidate?.llmName ?? ctx.primaryLlm, + model: winnerRef.candidate?.modelOverride ?? '', + failedOver: winnerRef.candidate !== null && winnerRef.candidate.llmName !== ctx.primaryLlm, + }; return; } throw new Error(`Chat loop exceeded ${String(ctx.maxIterations)} iterations without a terminal turn`); @@ -482,10 +514,14 @@ export class ChatService { systemBlock: string; toolList: ChatTool[]; mergedParams: AgentChatParams; - }): AsyncGenerator<{ data: string; done?: boolean }> { + }, winnerRef?: { candidate: PoolCandidate | null }): AsyncGenerator<{ data: string; done?: boolean }> { let lastErr: Error | null = null; for (const c of ctx.poolCandidates) { try { + // Record the candidate we're committing to; if it throws before its + // first chunk the loop overwrites this on the next iteration, so after + // a successful stream this holds the Llm that actually answered. + if (winnerRef) winnerRef.candidate = c; yield* this.streamOneCandidate(c, ctx); return; } catch (err) { @@ -585,17 +621,26 @@ export class ChatService { systemBlock: string; toolList: ChatTool[]; mergedParams: AgentChatParams; - }): Promise<{ status: number; body: unknown }> { + }): Promise<{ status: number; body: unknown; candidate: PoolCandidate }> { let lastErr: Error | null = null; for (const c of ctx.poolCandidates) { try { - return await this.dispatchOneCandidate(c, ctx); + const result = await this.dispatchOneCandidate(c, ctx); + // Fail over on an upstream error status or an empty/invalid completion — + // NOT just transport errors. This is what lets a 400 (drifted/down model) + // fall through to the next candidate and, ultimately, the anthropic fallback. + if (result.status >= 400 || extractChoice(result.body) === null) { + const detail = typeof result.body === 'string' ? result.body : JSON.stringify(result.body); + lastErr = new Error(`LLM '${c.llmName}' (model '${c.modelOverride}') HTTP ${String(result.status)}: ${detail.slice(0, 300)}`); + continue; + } + return { ...result, candidate: c }; } catch (err) { - lastErr = err as Error; - // Try the next pool member. + lastErr = new Error(`LLM '${c.llmName}' (model '${c.modelOverride}'): ${(err as Error).message}`); + // Try the next candidate. } } - throw lastErr ?? new Error('chat dispatch exhausted: no pool candidates'); + throw lastErr ?? new Error('chat dispatch exhausted: no candidates'); } private async dispatchOneCandidate( @@ -634,7 +679,7 @@ export class ChatService { private async prepareContext(args: ChatRequestArgs): Promise { const agent = await this.agents.getByName(args.agentName); const pinned = await this.llms.getByName(agent.llm.name); - const poolCandidates = await this.resolvePoolCandidates(pinned); + const poolCandidates = await this.resolveCandidatesWithFallbacks(pinned); if (poolCandidates.length === 0) { // Pool exists but every member is inactive (publishers offline, // public Llm explicitly disabled, etc.). Surface a clear error @@ -713,6 +758,7 @@ export class ChatService { history, systemBlock, poolCandidates, + primaryLlm: pinned.name, mergedParams, toolList: filteredTools, projectId, @@ -774,7 +820,7 @@ export class ChatService { } const pinned = await this.llms.getByName(project.llmProvider); // project.llmModel overrides the pinned Llm's own model for this chat. - const poolCandidates = await this.resolvePoolCandidates(pinned, project.llmModel ?? undefined); + const poolCandidates = await this.resolveCandidatesWithFallbacks(pinned, project.llmModel ?? undefined); if (poolCandidates.length === 0) { throw new NotFoundError( `No active Llm in pool '${effectivePoolName(pinned)}' (project '${project.name}' → ${pinned.name})`, @@ -824,6 +870,7 @@ export class ChatService { history, systemBlock, poolCandidates, + primaryLlm: pinned.name, mergedParams, toolList: filteredTools, projectId, @@ -892,6 +939,30 @@ export class ChatService { })); } + /** + * Build the ordered dispatch list: the primary Llm's pool, then each fallback + * Llm's pool (declared on the primary via `extraConfig.fallbacks: string[]`). + * The dispatch loop tries them in order and fails over on error/4xx/no-completion. + * A missing/misconfigured fallback is skipped, never fatal. + */ + private async resolveCandidatesWithFallbacks( + pinned: { name: string; poolName: string | null; extraConfig?: unknown }, + modelOverride?: string, + ): Promise { + const candidates = await this.resolvePoolCandidates(pinned, modelOverride); + for (const name of extractFallbackNames(pinned.extraConfig)) { + if (name === pinned.name) continue; // no self-reference loops + try { + const fb = await this.llms.getByName(name); + // A fallback uses its OWN configured model — not the primary's override. + candidates.push(...(await this.resolvePoolCandidates(fb))); + } catch { + // Fallback missing/inactive — skip it, keep the rest of the chain. + } + } + return candidates; + } + /** * Resolves a personality (request override → agent default) and returns * its bound prompt contents in `PersonalityPrompt.priority` desc order. @@ -1099,6 +1170,14 @@ function safeParseJson(s: string): unknown { } } +/** Read an ordered fallback-Llm list from a primary Llm's `extraConfig.fallbacks`. */ +function extractFallbackNames(extraConfig: unknown): string[] { + if (extraConfig === null || typeof extraConfig !== 'object') return []; + const raw = (extraConfig as { fallbacks?: unknown }).fallbacks; + if (!Array.isArray(raw)) return []; + return raw.filter((x): x is string => typeof x === 'string' && x.length > 0); +} + /** The gated virtual secret-read tool offered to project chat with --allow-secrets. */ function secretReadTool(): ChatTool { return { diff --git a/src/mcpd/tests/chat-service.test.ts b/src/mcpd/tests/chat-service.test.ts index 2b14031..42f4f9d 100644 --- a/src/mcpd/tests/chat-service.test.ts +++ b/src/mcpd/tests/chat-service.test.ts @@ -1080,3 +1080,64 @@ function mockPersonalityRepo( findBinding: vi.fn(async () => null), }; } + +// ── Failover chain + "which model answered" reporting ── +describe('ChatService — failover chain', () => { + const NOW2 = new Date(); + function llmRow(name: string, model: string, fallbacks: string[] = []): Record { + return { + id: `id-${name}`, name, type: 'openai', model, url: '', tier: 'fast', description: '', + apiKeySecretId: null, apiKeySecretKey: null, + extraConfig: fallbacks.length > 0 ? { fallbacks } : {}, + poolName: null, kind: 'public', providerSessionId: null, status: 'active', + lastHeartbeatAt: null, inactiveSince: null, version: 1, createdAt: NOW2, updatedAt: NOW2, + }; + } + function failoverLlms(): LlmService { + const rows: Record> = { + 'primary-llm': llmRow('primary-llm', 'primary-model', ['fallback-llm']), + 'fallback-llm': llmRow('fallback-llm', 'fallback-model'), + }; + return { + getByName: vi.fn(async (name: string) => ({ ...rows[name], apiKeyRef: null })), + findByPoolName: vi.fn(async (pool: string) => [rows[pool]]), + resolveApiKey: vi.fn(async () => 'k'), + } as unknown as LlmService; + } + function agentPinnedTo(llmName: string): AgentService { + return { + getByName: vi.fn(async (name: string) => ({ + id: `agent-${name}`, name, description: '', systemPrompt: 'sys', + llm: { id: 'x', name: llmName }, project: null, defaultPersonality: null, + proxyModelName: null, defaultParams: {}, extras: {}, ownerId: 'o', version: 1, createdAt: NOW2, updatedAt: NOW2, + })), + } as unknown as AgentService; + } + const status400 = (): NonStreamingResult => ({ status: 400, body: { error: 'model not served' } }); + + it('fails over to the fallback Llm on a primary 400 and reports the answering model', async () => { + const adapter = scriptedAdapter([status400(), chatCompletion('fallback answer')]); + const svc = new ChatService(agentPinnedTo('primary-llm'), failoverLlms(), adapterRegistry(adapter), mockChatRepo(), mockPromptRepo(), mockTools()); + const res = await svc.chat({ agentName: 'a', userMessage: 'hi', ownerId: 'o' }); + expect(res.assistant).toBe('fallback answer'); + expect(res.failedOver).toBe(true); + expect(res.llm).toBe('fallback-llm'); + expect(res.model).toBe('fallback-model'); + }); + + it('reports the primary model + failedOver=false when the primary answers', async () => { + const adapter = scriptedAdapter([chatCompletion('primary answer')]); + const svc = new ChatService(agentPinnedTo('primary-llm'), failoverLlms(), adapterRegistry(adapter), mockChatRepo(), mockPromptRepo(), mockTools()); + const res = await svc.chat({ agentName: 'a', userMessage: 'hi', ownerId: 'o' }); + expect(res.assistant).toBe('primary answer'); + expect(res.failedOver).toBe(false); + expect(res.llm).toBe('primary-llm'); + expect(res.model).toBe('primary-model'); + }); + + it('throws a clear aggregated error naming the model when every candidate fails', async () => { + const adapter = scriptedAdapter([status400(), status400()]); + const svc = new ChatService(agentPinnedTo('primary-llm'), failoverLlms(), adapterRegistry(adapter), mockChatRepo(), mockPromptRepo(), mockTools()); + await expect(svc.chat({ agentName: 'a', userMessage: 'hi', ownerId: 'o' })).rejects.toThrow(/HTTP 400/); + }); +});