From 892ab4cccdff680d1877a5dbd536830ddb3df2be Mon Sep 17 00:00:00 2001 From: autodev Date: Wed, 14 Oct 2026 15:04:24 +0800 Subject: [PATCH 1/3] fix: retry transient gateway 502 on the DSH host-LLM path (#2494) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit complete() dispatched a single prepared.stream() call with no retry, so a transient 5xx from the gateway in front of the host Messages API failed the auxiliary call outright. Recall filtering and capture summarization then degraded silently (safeCutoff / heuristic) — 25 of 96 host calls failed on the reporter's installation. Add a bounded retry in the bridge: at most 3 attempts, 250ms x attempt backoff, bounded by the existing per-call deadline. Retry only LLM_UNAVAILABLE with no status or status >= 500; never retry 401/402/403, rate limits, timeouts, malformed output, or deadline/caller aborts. Each attempt re-prepares via the cached capability path because a prepared DSH call is single-use. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../adapters/deepseek-harness/host-llm.ts | 175 +++++++++++---- .../deepseek-harness-host-llm.test.ts | 206 ++++++++++++++++++ 2 files changed, 340 insertions(+), 41 deletions(-) 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..b2a9e1c63 100644 --- a/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts +++ b/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts @@ -31,6 +31,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 +161,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 +191,124 @@ 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 }> { + 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 waitForRetryBackoff(HOST_LLM_RETRY_BASE_DELAY_MS * attempt, signal); + } + // Every iteration returns or throws; this only satisfies the type checker. + 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), 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; + } + const status = error.details?.status; + if (status === undefined) return true; + return typeof status === "number" && status >= 500; +} + +/** Cancellable backoff; rejects with the caller's abort reason if cancelled mid-wait. */ +function waitForRetryBackoff(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) { + return Promise.reject(signal.reason ?? new DOMException("DSH operation aborted", "AbortError")); + } + return new Promise((resolve, reject) => { + const onAbort = (): void => { + clearTimeout(timer); + reject(signal.reason ?? new DOMException("DSH operation aborted", "AbortError")); + }; + const timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, Math.max(0, delayMs)); + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +/** 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..23e7a6dae 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,209 @@ 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 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); + }); + }); }); From 7bcf73a3b81b18b3c8fe33dee60e3a8c8e5c04ee Mon Sep 17 00:00:00 2001 From: autodev Date: Wed, 14 Oct 2026 15:23:53 +0800 Subject: [PATCH 2/3] fix: never retry a provider-side abort on the DSH host-LLM path (#2494) A provider `aborted` finish maps to LLM_UNAVAILABLE with no status, which the status gate would treat as a retryable transport failure. A cancellation is not transient, so exclude dshCode "ABORTED" explicitly and cover it with a regression test. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../adapters/deepseek-harness/host-llm.ts | 7 +++++-- .../adapters/deepseek-harness-host-llm.test.ts | 15 +++++++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) 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 b2a9e1c63..7bbf3b4ec 100644 --- a/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts +++ b/apps/memos-local-plugin/adapters/deepseek-harness/host-llm.ts @@ -242,13 +242,16 @@ async function completeAuxiliaryWithRetry( * 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), rate limits and timeouts - * (their own codes), malformed output, and deadline aborts. + * 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; 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 23e7a6dae..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 @@ -811,6 +811,21 @@ describe("DeepSeek Harness host LLM bridge", () => { 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 }, From be5840da5d0d07389b752c5bac643928fdaea190 Mon Sep 17 00:00:00 2001 From: autodev Date: Wed, 14 Oct 2026 16:04:01 +0800 Subject: [PATCH 3/3] refactor(plugin): reuse shared waitForRetry and guard the retry bound Replace the adapter-local waitForRetryBackoff with the shared core/util/retry-after.js waitForRetry so the abort-listener contract has a single implementation, and fail fast when HOST_LLM_MAX_ATTEMPTS is misconfigured to a non-positive value instead of throwing `undefined`. Refs #2494 --- .../adapters/deepseek-harness/host-llm.ts | 27 +++++-------------- 1 file changed, 7 insertions(+), 20 deletions(-) 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 7bbf3b4ec..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"; @@ -209,6 +210,9 @@ async function completeAuxiliaryWithRetry( 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(); @@ -229,9 +233,10 @@ async function completeAuxiliaryWithRetry( } lastError = error; } - await waitForRetryBackoff(HOST_LLM_RETRY_BASE_DELAY_MS * attempt, signal); + await waitForRetry(HOST_LLM_RETRY_BASE_DELAY_MS * attempt, signal); } - // Every iteration returns or throws; this only satisfies the type checker. + // 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; } @@ -257,24 +262,6 @@ function isRetryableAuxiliaryError(error: unknown): boolean { return typeof status === "number" && status >= 500; } -/** Cancellable backoff; rejects with the caller's abort reason if cancelled mid-wait. */ -function waitForRetryBackoff(delayMs: number, signal: AbortSignal): Promise { - if (signal.aborted) { - return Promise.reject(signal.reason ?? new DOMException("DSH operation aborted", "AbortError")); - } - return new Promise((resolve, reject) => { - const onAbort = (): void => { - clearTimeout(timer); - reject(signal.reason ?? new DOMException("DSH operation aborted", "AbortError")); - }; - const timer = setTimeout(() => { - signal.removeEventListener("abort", onAbort); - resolve(); - }, Math.max(0, delayMs)); - signal.addEventListener("abort", onAbort, { once: true }); - }); -} - /** Validate one completed attempt's blocks and extract its text. */ function assembleAuxiliaryText( assembler: BlockAssembler,