Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/memos-local-plugin/core/config/defaults.ts
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@ export const DEFAULT_CONFIG: ResolvedConfig = {
// hits before injection.
llmFilterMinCandidates: 2,
llmFilterCandidateBodyChars: 500,
llmFilterTimeoutMs: 2_000,
// Default 0 — no time-window bound, keeping the legacy
// brute-force scan behaviour for fresh installs that haven't
// grown past the threshold where the bound starts paying off.
Expand Down
11 changes: 11 additions & 0 deletions apps/memos-local-plugin/core/config/schema.ts
Original file line number Diff line number Diff line change
Expand Up @@ -586,6 +586,17 @@ const AlgorithmSchema = Type.Object({
* slightly larger window pays for itself).
*/
llmFilterCandidateBodyChars: NumberInRange(500, 120, 2000),
/**
* Wall-clock budget (ms) for one LLM relevance-filter call. The
* filter always runs inside the retrieval deadline, so this value
* is additionally clamped by the remaining deadline. Default 2000
* (2 s) fits hosted models; self-hosted OpenAI-compatible
* endpoints on consumer hardware regularly need more — with the
* old hard-coded cap the local call was aborted every turn and
* replayed on the host model
* (https://github.com/MemTensor/MemOS/issues/2499).
*/
llmFilterTimeoutMs: NumberInRange(2_000, 100, 30_000),
/**
* Tier-2 vector scan time-window bound (ms). When > 0, the
* vector scan path (`scanAndTopK` in `core/storage/vector.ts`)
Expand Down
28 changes: 23 additions & 5 deletions apps/memos-local-plugin/core/llm/fetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -338,7 +338,15 @@ function isTransientError(err: unknown): boolean {
function isTimeout(err: unknown): boolean {
if (err instanceof Error) {
if (err.name === "TimeoutError") return true;
if ((err as { code?: string }).code === "ABORT_ERR") return true;
// An AbortError whose *reason* is a TimeoutError comes from
// AbortSignal.timeout(); a plain AbortError without that reason comes
// from a caller ctrl.abort() and must NOT be classified as a timeout.
if (err.name === "AbortError") {
const reason: unknown = (err as { reason?: unknown }).reason;
if (reason instanceof Error && reason.name === "TimeoutError") return true;
}
const code: unknown = (err as { code?: unknown }).code;
if (code === "ABORT_ERR" || code === 20) return true;
if (/timeout|ETIMEDOUT/i.test(err.message ?? "")) return true;
}
return false;
Expand Down Expand Up @@ -374,10 +382,20 @@ function remainingDeadlineMs(deadlineAt?: number, nowMs: number = Date.now()): n
function mergeSignals(a: AbortSignal | undefined, b: AbortSignal): AbortSignal {
if (!a) return b;
const ctrl = new AbortController();
const forward = () => ctrl.abort();
if (a.aborted || b.aborted) ctrl.abort();
a.addEventListener("abort", forward, { once: true });
b.addEventListener("abort", forward, { once: true });
// Forward the abort *reason*: without it, `AbortSignal.timeout()`'s
// `TimeoutError` is flattened into a bare `AbortError` and downstream
// timeout classification can no longer tell a budget timeout from a
// caller cancellation (issue #2499).
const forwardFrom = (from: AbortSignal) => () => ctrl.abort(from.reason);
// Prefer the TimeoutError reason so downstream classification stays accurate
// even when both signals fire simultaneously.
const immediateReason =
(a.aborted && a.reason instanceof Error && a.reason.name === "TimeoutError")
? a.reason
: a.aborted ? a.reason : b.reason;
if (a.aborted || b.aborted) ctrl.abort(immediateReason);
a.addEventListener("abort", forwardFrom(a), { once: true });
b.addEventListener("abort", forwardFrom(b), { once: true });
return ctrl.signal;
}

Expand Down
1 change: 1 addition & 0 deletions apps/memos-local-plugin/core/pipeline/deps.ts
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,7 @@ export function extractAlgorithmConfig(
llmFilterMaxKeep: alg.retrieval.llmFilterMaxKeep,
llmFilterMinCandidates: alg.lightweightMemory.enabled ? 1 : alg.retrieval.llmFilterMinCandidates,
llmFilterCandidateBodyChars: alg.retrieval.llmFilterCandidateBodyChars,
llmFilterTimeoutMs: alg.retrieval.llmFilterTimeoutMs,
lightweightMemory: alg.lightweightMemory.enabled,
vectorScanMaxAgeMs: alg.retrieval.vectorScanMaxAgeMs,
},
Expand Down
12 changes: 8 additions & 4 deletions apps/memos-local-plugin/core/retrieval/retrieve.ts
Original file line number Diff line number Diff line change
Expand Up @@ -429,7 +429,7 @@ async function runAll(
config: deps.config,
signal: opts.signal,
deadlineAt: opts.deadlineAt,
timeoutMs: filterTimeoutMs(opts.deadlineAt),
timeoutMs: filterTimeoutMs(opts.deadlineAt, deps.config.llmFilterTimeoutMs),
malformedRetries: opts.llmFilterMalformedRetries,
},
);
Expand Down Expand Up @@ -490,7 +490,7 @@ async function runAll(
config: deps.config,
signal: opts.signal,
deadlineAt: opts.deadlineAt,
timeoutMs: filterTimeoutMs(opts.deadlineAt),
timeoutMs: filterTimeoutMs(opts.deadlineAt, deps.config.llmFilterTimeoutMs),
malformedRetries: opts.llmFilterMalformedRetries,
},
);
Expand Down Expand Up @@ -688,9 +688,13 @@ async function runAll(
}
}

function filterTimeoutMs(deadlineAt?: number): number | undefined {
/** Default cap for a single LLM filter call when the operator has not
* configured `llmFilterTimeoutMs` (kept from the legacy hard-coded value). */
const DEFAULT_FILTER_TIMEOUT_MS = 2_000;

function filterTimeoutMs(deadlineAt: number | undefined, capMs?: number): number | undefined {
if (deadlineAt === undefined) return undefined;
return Math.max(1, Math.min(2_000, deadlineAt - Date.now()));
return Math.max(1, Math.min(capMs ?? DEFAULT_FILTER_TIMEOUT_MS, deadlineAt - Date.now()));
}

function emptyResult(
Expand Down
9 changes: 9 additions & 0 deletions apps/memos-local-plugin/core/retrieval/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,15 @@ export interface RetrievalConfig {
* window pays for itself).
*/
llmFilterCandidateBodyChars?: number;
/**
* Wall-clock budget (ms) for a single LLM filter call. Always clamped
* by the retrieval deadline when one is set. Default 2000 (2 s) fits
* hosted models; self-hosted OpenAI-compatible endpoints on consumer
* hardware regularly need more, and with the old hard-coded cap every
* turn_start aborted the local call and fell back to the host model
* (issue #2499).
*/
llmFilterTimeoutMs?: number;
/** Low-cost mode: retrieve raw trace memories only. */
lightweightMemory?: boolean;
/**
Expand Down
57 changes: 56 additions & 1 deletion apps/memos-local-plugin/tests/unit/llm/fetcher.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";

import { MemosError } from "../../../agent-contract/errors.js";
import { MemosError, ERROR_CODES } from "../../../agent-contract/errors.js";
import { decodeSse, httpPostJson, httpPostStream } from "../../../core/llm/fetcher.js";
import { initTestLogger } from "../../../core/logger/index.js";
import type { LlmProviderLogger } from "../../../core/llm/types.js";
Expand Down Expand Up @@ -82,6 +82,61 @@ describe("llm/fetcher", () => {
vi.useRealTimers();
});

/** Fetch that hangs forever and rejects with the merged signal's
* reason when it fires — mirrors undici's rejection semantics. */
function hangingFetch() {
const f = vi.fn((_url: unknown, init?: RequestInit) =>
new Promise<Response>((_resolve, reject) => {
init?.signal?.addEventListener(
"abort",
() => reject(init.signal?.reason),
{ once: true },
);
}),
);
vi.stubGlobal("fetch", f);
return f;
}

it("classifies an internal per-attempt timeout as LLM_TIMEOUT, not LLM_UNAVAILABLE (#2499)", async () => {
hangingFetch();
// A caller signal must be present for `mergeSignals` to run — without
// one, the raw `AbortSignal.timeout()` reaches fetch and its
// `TimeoutError` reason survives. The retrieval filter always passes
// one, which is how the reason got flattened into a bare `AbortError`
// and surfaced as `llm_unavailable` (issue #2499).
const ctrl = new AbortController();

await expect(
httpPostJson({
url: "https://x",
body: {},
timeoutMs: 50,
maxRetries: 0,
signal: ctrl.signal,
provider: "openai_compatible",
log: nullLog(),
}),
).rejects.toMatchObject({ code: ERROR_CODES.LLM_TIMEOUT });
});

it("still reports caller cancellation distinctly from a budget timeout (#2499)", async () => {
const ctrl = new AbortController();
hangingFetch();

const pending = httpPostJson({
url: "https://x",
body: {},
timeoutMs: 5_000,
maxRetries: 0,
signal: ctrl.signal,
provider: "openai_compatible",
log: nullLog(),
});
ctrl.abort();
await expect(pending).rejects.toMatchObject({ message: /was cancelled/ });
});

it("defers a long Retry-After from a 503 without retrying early", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-08-04T00:00:00.000Z"));
Expand Down
102 changes: 102 additions & 0 deletions apps/memos-local-plugin/tests/unit/retrieval/integration.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -456,6 +456,108 @@ describe("retrieval/integration", () => {
);
});

it("caps the LLM filter budget at algorithm.retrieval.llmFilterTimeoutMs (#2499)", async () => {
const captured: Array<{ timeoutMs?: number }> = [];
const llm: any = {
completeJson: async (_messages: unknown, opts: { timeoutMs?: number }) => {
captured.push(opts);
return { value: { selected: [1], sufficient: true }, servedBy: "fake" };
},
};
const res = await turnStartRetrieve(
{
...makeDeps(handle),
llm,
config: {
...makeDeps(handle).config,
llmFilterEnabled: true,
llmFilterMinCandidates: 1,
llmFilterTimeoutMs: 4_321,
},
},
{
reason: "turn_start",
agent: "openclaw",
sessionId: "s1" as SessionId,
userText: "run docker compose",
ts: NOW as never,
},
{ deadlineAt: Date.now() + 60_000 },
);

expect(res.stats.llmFilterOutcome).toBe("llm_filtered");
expect(captured.length).toBe(1);
expect(captured[0]!.timeoutMs).toBe(4_321);
});

it("keeps the legacy 2 s default filter budget when llmFilterTimeoutMs is unset (#2499)", async () => {
const captured: Array<{ timeoutMs?: number }> = [];
const llm: any = {
completeJson: async (_messages: unknown, opts: { timeoutMs?: number }) => {
captured.push(opts);
return { value: { selected: [1], sufficient: true }, servedBy: "fake" };
},
};
const res = await turnStartRetrieve(
{
...makeDeps(handle),
llm,
config: {
...makeDeps(handle).config,
llmFilterEnabled: true,
llmFilterMinCandidates: 1,
},
},
{
reason: "turn_start",
agent: "openclaw",
sessionId: "s1" as SessionId,
userText: "run docker compose",
ts: NOW as never,
},
{ deadlineAt: Date.now() + 60_000 },
);

expect(res.stats.llmFilterOutcome).toBe("llm_filtered");
expect(captured.length).toBe(1);
expect(captured[0]!.timeoutMs).toBe(2_000);
});

it("still clamps the filter budget to the remaining retrieval deadline (#2499)", async () => {
const captured: Array<{ timeoutMs?: number }> = [];
const llm: any = {
completeJson: async (_messages: unknown, opts: { timeoutMs?: number }) => {
captured.push(opts);
return { value: { selected: [1], sufficient: true }, servedBy: "fake" };
},
};
const deadlineAt = Date.now() + 250;
await turnStartRetrieve(
{
...makeDeps(handle),
llm,
config: {
...makeDeps(handle).config,
llmFilterEnabled: true,
llmFilterMinCandidates: 1,
llmFilterTimeoutMs: 4_321,
},
},
{
reason: "turn_start",
agent: "openclaw",
sessionId: "s1" as SessionId,
userText: "run docker compose",
ts: NOW as never,
},
{ deadlineAt },
);

expect(captured.length).toBe(1);
expect(captured[0]!.timeoutMs!).toBeGreaterThan(0);
expect(captured[0]!.timeoutMs!).toBeLessThanOrEqual(250);
});

it("can defer the local LLM pass for one final merged filter", async () => {
let filterCalls = 0;
const llm: any = {
Expand Down
Loading