feat(chat): LLM failover chain + show which model answered #83
@@ -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: <llm> (<model>)` per turn, and
|
||||
`⚠ failed over → answered by <llm> (<model>)` 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.
|
||||
|
||||
@@ -198,7 +198,8 @@ async function runOneShot(
|
||||
// `thread: <id>` — 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<string, unknown>,
|
||||
): 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
|
||||
|
||||
@@ -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<ChatResult> {
|
||||
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<ChatStreamChunk> {
|
||||
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<PreparedChatContext> {
|
||||
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<PoolCandidate[]> {
|
||||
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 {
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
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<string, Record<string, unknown>> = {
|
||||
'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/);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user