diff --git a/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts b/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts index a755efb98..6654b0ba2 100644 --- a/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts +++ b/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts @@ -24,6 +24,7 @@ import type { HostLlmCompletion, } from "../../core/llm/host-bridge.js"; import type { LlmMessage, LlmUsage } from "../../core/llm/types.js"; +import { waitForRetry } from "../../core/util/retry-after.js"; const HOST_LLM_BRIDGE_ID = "deepseek-harness.host.v1"; const HOST_LLM_TIMEOUT_CODE = "MEMOS_DSH_HOST_LLM_TIMEOUT"; @@ -31,6 +32,19 @@ const HOST_LLM_MESSAGE_SOURCE = "memos-local-memory"; const NO_REASONING_EFFORT = ReasoningEffortId("off"); const UNSUPPORTED_REASONING_EFFORT = "UNSUPPORTED_REASONING_EFFORT"; const MODEL_CAPABILITY_TTL_MS = 10 * 60 * 1_000; +/** + * Bounded retry budget for one auxiliary host LLM call. + * + * The gateway in front of the host Messages API intermittently answers 502, + * which is a transient transport failure, not a logical one. The interactive + * chat path retries; auxiliary stages (recall filtering, capture + * summarization) silently degrade instead of retrying, so a single blip drops + * the whole stage. Retry a bounded number of times, bounded by the caller's + * per-call deadline. + */ +const HOST_LLM_MAX_ATTEMPTS = 3; +/** `250 ms × attempt` backoff between auxiliary retries. */ +const HOST_LLM_RETRY_BASE_DELAY_MS = 250; /** Atomic provider/model route captured from the DSH agent that owns a turn. */ export interface DeepSeekHarnessLlmRoute { @@ -148,55 +162,17 @@ export function createDeepSeekHarnessHostLlmBridge( ); try { - callDeadline.signal.throwIfAborted(); - const prepared = await prepareAuxiliaryCall( + const result = await completeAuxiliaryWithRetry( options.llm, input, route, callDeadline.signal, capabilityCache, ); - const request = createGenerateOptions(input, route, prepared.config); - request.signal = callDeadline.signal; - const assembler = new BlockAssembler(); - for await (const chunk of prepared.stream(request)) { - callDeadline.signal.throwIfAborted(); - assembler.push(chunk); - } - callDeadline.signal.throwIfAborted(); - - assertSuccessfulFinish(assembler.finish); - const blocks = assembler.blocks(); - if (blocks.some((block) => block.type === "tool-call")) { - throw outputError( - "DeepSeek Harness host LLM returned tool calls for a text-only MemOS request", - "TOOL_CALLS", - ); - } - if (blocks.some((block) => block.type !== "text" && block.type !== "reasoning")) { - throw outputError( - "DeepSeek Harness host LLM returned unsupported non-text content", - "UNSUPPORTED_CONTENT", - ); - } - - const text = blocks - .filter((block): block is Extract<(typeof blocks)[number], { type: "text" }> => ( - block.type === "text" - )) - .map((block) => block.text) - .join(""); - if (!text.trim()) { - throw outputError( - "DeepSeek Harness host LLM produced no text content", - "EMPTY_TEXT", - ); - } - return { - text, + text: result.text, model: route.model, - ...(assembler.usage === undefined ? {} : { usage: mapUsage(assembler.usage) }), + ...(result.usage === undefined ? {} : { usage: result.usage }), durationMs: Date.now() - startedAt, }; } catch (error) { @@ -216,6 +192,113 @@ export function createDeepSeekHarnessHostLlmBridge( }; } +/** + * Dispatch one auxiliary host LLM call, retrying a bounded number of transient + * failures. + * + * A prepared DSH call is single-use, so each attempt re-prepares through the + * same registration-bound path (the model-capability lookup is cached, so this + * costs no extra round trip). The whole loop is bounded by the caller's + * deadline: `signal.throwIfAborted()` before every attempt and a cancellable + * backoff make the caller's abort authoritative, so a deadline abort never + * turns into another request. + */ +async function completeAuxiliaryWithRetry( + llm: DeepSeekHarnessLlmLike, + input: HostLlmCompleteInput, + route: DeepSeekHarnessLlmRoute, + signal: AbortSignal, + capabilityCache: CapabilityCache, +): Promise<{ text: string; usage?: LlmUsage }> { + if (HOST_LLM_MAX_ATTEMPTS <= 0) { + throw new Error(`HOST_LLM_MAX_ATTEMPTS must be >= 1, got ${HOST_LLM_MAX_ATTEMPTS}`); + } + let lastError: unknown; + for (let attempt = 1; attempt <= HOST_LLM_MAX_ATTEMPTS; attempt++) { + signal.throwIfAborted(); + try { + const prepared = await prepareAuxiliaryCall(llm, input, route, signal, capabilityCache); + const request = createGenerateOptions(input, route, prepared.config); + request.signal = signal; + const assembler = new BlockAssembler(); + for await (const chunk of prepared.stream(request)) { + signal.throwIfAborted(); + assembler.push(chunk); + } + signal.throwIfAborted(); + return assembleAuxiliaryText(assembler); + } catch (error) { + if (attempt >= HOST_LLM_MAX_ATTEMPTS || !isRetryableAuxiliaryError(error)) { + throw error; + } + lastError = error; + } + await waitForRetry(HOST_LLM_RETRY_BASE_DELAY_MS * attempt, signal); + } + // Unreachable: the guard at the top of the loop rejects an empty attempt + // range, so every iteration either returns or throws before reaching here. + throw lastError; +} + +/** + * Whether an auxiliary-call failure is a transient transport failure worth + * retrying. + * + * Only an unusable provider is retried, and only when it carries no HTTP + * status (a transport failure) or a server-error status (`>= 500`). Logical + * outcomes are never retried: `401`/`402`/`403` (a bad or exhausted + * credential fails identically on every attempt), provider aborts + * (`dshCode: "ABORTED"`, a cancellation), rate limits and timeouts (their own + * codes), malformed output, and deadline aborts. + */ +function isRetryableAuxiliaryError(error: unknown): boolean { + if (!(error instanceof MemosError) || error.code !== ERROR_CODES.LLM_UNAVAILABLE) { + return false; + } + // A provider-side abort is a cancellation, not a transient failure. + if (error.details?.dshCode === "ABORTED") return false; + const status = error.details?.status; + if (status === undefined) return true; + return typeof status === "number" && status >= 500; +} + +/** Validate one completed attempt's blocks and extract its text. */ +function assembleAuxiliaryText( + assembler: BlockAssembler, +): { text: string; usage?: LlmUsage } { + assertSuccessfulFinish(assembler.finish); + const blocks = assembler.blocks(); + if (blocks.some((block) => block.type === "tool-call")) { + throw outputError( + "DeepSeek Harness host LLM returned tool calls for a text-only MemOS request", + "TOOL_CALLS", + ); + } + if (blocks.some((block) => block.type !== "text" && block.type !== "reasoning")) { + throw outputError( + "DeepSeek Harness host LLM returned unsupported non-text content", + "UNSUPPORTED_CONTENT", + ); + } + + const text = blocks + .filter((block): block is Extract<(typeof blocks)[number], { type: "text" }> => ( + block.type === "text" + )) + .map((block) => block.text) + .join(""); + if (!text.trim()) { + throw outputError( + "DeepSeek Harness host LLM produced no text content", + "EMPTY_TEXT", + ); + } + return { + text, + ...(assembler.usage === undefined ? {} : { usage: mapUsage(assembler.usage) }), + }; +} + function createGenerateOptions( input: HostLlmCompleteInput, route: DeepSeekHarnessLlmRoute, diff --git a/apps/memos-local-plugin/tests/unit/adapters/deepseek-harness-host-llm.test.ts b/apps/memos-local-plugin/tests/unit/adapters/deepseek-harness-host-llm.test.ts index 493fba9e7..23f2f65d0 100644 --- a/apps/memos-local-plugin/tests/unit/adapters/deepseek-harness-host-llm.test.ts +++ b/apps/memos-local-plugin/tests/unit/adapters/deepseek-harness-host-llm.test.ts @@ -6,6 +6,7 @@ import LlmRuntime, { resolveRetryPolicy, type GenerateOptions, type LlmCallConfig, + type LlmFailure, type LlmResolvedModelInfo, type PreparedLlmCall, type StreamChunk, @@ -703,4 +704,224 @@ describe("DeepSeek Harness host LLM bridge", () => { }); expect(observedSignal?.aborted).toBe(true); }); + + describe("bounded retry on transient host LLM failures", () => { + /** + * One scripted attempt outcome: `undefined` succeeds, an `LlmFailure` + * fails the attempt. Each attempt re-prepares a single-use call, so the + * recorded attempt count is exactly the number of dispatches. + */ + function scriptedFailures(script: readonly (LlmFailure | undefined)[]): { + llm: DeepSeekHarnessLlmLike; + attempts: () => number; + } { + let attempt = 0; + const llm: DeepSeekHarnessLlmLike = { + resolveModelInfo: (provider, model) => Promise.resolve({ + provider, + id: model, + name: model, + reasoning: { efforts: [{ id: ReasoningEffortId("off"), name: "Off" }] }, + }), + prepareCall: (config) => { + const outcome = script[Math.min(attempt, script.length - 1)]; + attempt++; + return Promise.resolve(preparedCall(config, () => (async function* () { + if (outcome) { + yield { type: "finish", reason: { kind: "error", failure: outcome } } as StreamChunk; + return; + } + yield { type: "text-delta", index: 0, text: "recovered" } as StreamChunk; + yield { type: "finish", reason: { kind: "stop" } } as StreamChunk; + })())); + }, + }; + return { llm, attempts: () => attempt }; + } + + function completeWith(llm: DeepSeekHarnessLlmLike, timeoutMs?: number) { + const routes = new DeepSeekHarnessLlmRouteContext(); + const bridge = createDeepSeekHarnessHostLlmBridge({ llm, routes }); + return routes.run(ROUTE, () => bridge.complete({ + messages: [{ role: "user", content: "hello" }], + ...(timeoutMs === undefined ? {} : { timeoutMs }), + })); + } + + // The gateway 502 the issue reports: an UNAVAILABLE failure carrying a + // server-error status. `[502,502,ok]` recovers on the third dispatch. + const gateway502: LlmFailure = { message: "DeepSeek Messages request failed (502)", code: "UNAVAILABLE", status: 502 }; + + it("retries a transient 502 and succeeds within the attempt bound", async () => { + const { llm, attempts } = scriptedFailures([gateway502, gateway502, undefined]); + + const result = await completeWith(llm, 5_000); + + expect(result.text).toBe("recovered"); + expect(attempts()).toBe(3); + }); + + it("stops retrying as soon as an attempt succeeds", async () => { + const { llm, attempts } = scriptedFailures([gateway502, undefined]); + + const result = await completeWith(llm, 5_000); + + expect(result.text).toBe("recovered"); + expect(attempts()).toBe(2); + }); + + it("gives up after the attempt bound and rethrows the last failure", async () => { + const { llm, attempts } = scriptedFailures([gateway502, gateway502, gateway502]); + + await expect(completeWith(llm, 5_000)).rejects.toMatchObject({ + code: "llm_unavailable", + details: { dshCode: "UNAVAILABLE", status: 502 }, + }); + expect(attempts()).toBe(3); + }); + + it("does not retry a successful first attempt", async () => { + const { llm, attempts } = scriptedFailures([undefined]); + + await expect(completeWith(llm, 5_000)).resolves.toMatchObject({ text: "recovered" }); + expect(attempts()).toBe(1); + }); + + it.each([ + { name: "credential rejection", failure: { message: "unauthorized", code: "AUTH", status: 401 } as LlmFailure }, + { name: "payment required", failure: { message: "insufficient balance", code: "QUOTA", status: 402 } as LlmFailure }, + { name: "forbidden", failure: { message: "forbidden", code: "AUTH", status: 403 } as LlmFailure }, + ])("never retries $name", async ({ failure }) => { + const { llm, attempts } = scriptedFailures([failure, undefined]); + + await expect(completeWith(llm, 5_000)).rejects.toMatchObject({ + code: "llm_unavailable", + details: { status: failure.status }, + }); + expect(attempts()).toBe(1); + }); + + it("retries a statusless transient transport failure", async () => { + const { llm, attempts } = scriptedFailures([ + { message: "fetch failed", code: "UNAVAILABLE" }, + undefined, + ]); + + await expect(completeWith(llm, 5_000)).resolves.toMatchObject({ text: "recovered" }); + expect(attempts()).toBe(2); + }); + + it("does not retry a provider-side abort", async () => { + // A provider `aborted` finish maps to LLM_UNAVAILABLE with no status; it + // is a cancellation, not a transient failure, and must not be retried. + const { llm, attempts } = scriptedFailures([ + { message: "request cancelled", code: "ABORTED" }, + undefined, + ]); + + await expect(completeWith(llm, 5_000)).rejects.toMatchObject({ + code: "llm_unavailable", + details: { dshCode: "ABORTED" }, + }); + expect(attempts()).toBe(1); + }); + + it("does not retry a rate limit, which carries its own code", async () => { + const { llm, attempts } = scriptedFailures([ + { message: "too many requests", code: "RATE_LIMIT", status: 429 }, + undefined, + ]); + + await expect(completeWith(llm, 5_000)).rejects.toMatchObject({ + code: "llm_rate_limited", + details: { dshCode: "RATE_LIMIT" }, + }); + expect(attempts()).toBe(1); + }); + + it("does not retry malformed output", async () => { + let attempt = 0; + const llm: DeepSeekHarnessLlmLike = { + resolveModelInfo: (provider, model) => Promise.resolve({ + provider, + id: model, + name: model, + reasoning: { efforts: [{ id: ReasoningEffortId("off"), name: "Off" }] }, + }), + prepareCall: (config) => { + attempt++; + return Promise.resolve(preparedCall(config, () => (async function* () { + yield { type: "finish", reason: { kind: "max-tokens" } } as StreamChunk; + })())); + }, + }; + + await expect(completeWith(llm, 5_000)).rejects.toMatchObject({ + code: "llm_output_malformed", + details: { dshCode: "MAX_TOKENS" }, + }); + expect(attempt).toBe(1); + }); + + it("does not retry when the bounded deadline aborts the call", async () => { + let attempt = 0; + const llm: DeepSeekHarnessLlmLike = { + resolveModelInfo: (provider, model) => Promise.resolve({ provider, id: model, name: model }), + prepareCall(config, signal) { + attempt++; + return Promise.resolve(preparedCall(config, () => (async function* () { + await new Promise((resolve) => { + signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + yield { type: "text-delta", index: 0, text: "late" } as StreamChunk; + yield { type: "finish", reason: { kind: "stop" } } as StreamChunk; + })())); + }, + }; + + await expect(completeWith(llm, 5)).rejects.toMatchObject({ + code: "llm_timeout", + details: { dshCode: "MEMOS_DSH_HOST_LLM_TIMEOUT" }, + }); + expect(attempt).toBe(1); + }); + + it("does not start a retry when the caller aborts during backoff", async () => { + let attempt = 0; + const controller = new AbortController(); + const llm: DeepSeekHarnessLlmLike = { + resolveModelInfo: (provider, model) => Promise.resolve({ + provider, + id: model, + name: model, + reasoning: { efforts: [{ id: ReasoningEffortId("off"), name: "Off" }] }, + }), + prepareCall: (config) => { + attempt++; + // First (and only) attempt fails fast, leaving the loop in its + // 250 ms backoff wait when the caller cancels. + return Promise.resolve(preparedCall(config, () => (async function* () { + yield { type: "finish", reason: { kind: "error", failure: gateway502 } } as StreamChunk; + })())); + }, + }; + const routes = new DeepSeekHarnessLlmRouteContext(); + const bridge = createDeepSeekHarnessHostLlmBridge({ llm, routes }); + + const abortTimer = setTimeout( + () => controller.abort(new Error("caller cancelled during backoff")), + 50, + ); + try { + await expect(routes.run(ROUTE, () => bridge.complete({ + messages: [{ role: "user", content: "hello" }], + signal: controller.signal, + timeoutMs: 30_000, + }))).rejects.toThrow("caller cancelled during backoff"); + } finally { + clearTimeout(abortTimer); + } + expect(attempt).toBe(1); + }); + }); });