From 78184ea7e42f2f0509d57de5ecc5ccbd4982873c Mon Sep 17 00:00:00 2001 From: Alix-007 Date: Tue, 16 Jun 2026 02:08:47 +0800 Subject: [PATCH] fix(memory): abort orphaned qmd search subprocess when memory_search times out PR #91742 wired memory_search's 15s deadline AbortSignal through the builtin memory manager but missed the QMD backend behind the same MemorySearchManager.search interface. With QMD, the tool returns "timed out after 15s" to the agent while the spawned qmd query/search subprocess keeps running for the full qmd command timeout (memory.qmd.limits.timeoutMs, whose embed-heavy default was raised to 600s in #87572), leaving orphaned embedding/search work running after the agent already moved on. Add optional AbortSignal support to runCliCommand: an aborting signal kills the spawned child immediately and rejects with the abort reason, funneled through a single settle() guard so abort/timeout/error/close cannot double-settle. Thread the search signal through QmdMemoryManager.search -> runQmdSearch -> runQmd -> runCliCommand for the default direct-qmd subprocess path (including the query fallback), and fast-fail search() when the signal is already aborted. --- .../src/memory/qmd-manager.test.ts | 88 +++++++++++++++++ .../memory-core/src/memory/qmd-manager.ts | 38 +++++++- .../src/host/qmd-process.test.ts | 47 ++++++++- .../memory-host-sdk/src/host/qmd-process.ts | 97 ++++++++++++++----- 4 files changed, 241 insertions(+), 29 deletions(-) diff --git a/extensions/memory-core/src/memory/qmd-manager.test.ts b/extensions/memory-core/src/memory/qmd-manager.test.ts index 0022a1898818..cc98fa1b348b 100644 --- a/extensions/memory-core/src/memory/qmd-manager.test.ts +++ b/extensions/memory-core/src/memory/qmd-manager.test.ts @@ -2721,6 +2721,94 @@ describe("QmdMemoryManager", () => { await manager.close(); }); + it("aborts the in-flight qmd search subprocess when the caller signal aborts", async () => { + cfg = { + ...cfg, + memory: { + backend: "qmd", + qmd: { + includeDefaultMemory: false, + searchMode: "query", + update: { interval: "0s", debounceMs: 60_000, onBoot: false }, + paths: [{ path: workspaceDir, pattern: "**/*.md", name: "workspace" }], + }, + }, + } as OpenClawConfig; + + // The query child never closes on its own so the only way the search can + // settle is the caller-owned abort signal killing the subprocess. + let queryChildKill: ReturnType | undefined; + let queryChild: MockChild | undefined; + spawnMock.mockImplementation((_cmd: string, args: string[]) => { + if (args[0] === "query") { + const child = createMockChild({ autoClose: false }); + const kill = vi.fn(() => { + // Mirror a real child exiting after SIGKILL so the close handler runs. + queueMicrotask(() => child.emit("close", null)); + }); + Object.assign(child, { kill }); + queryChildKill = kill; + queryChild = child; + return child; + } + return createMockChild(); + }); + + const { manager } = await createManager(); + const controller = new AbortController(); + + const searchPromise = manager.search("test", { + sessionKey: "agent:main:slack:dm:u123", + signal: controller.signal, + }); + searchPromise.catch(() => undefined); + + await waitUntil(() => queryChildKill !== undefined); + expect(queryChild).toBeDefined(); + + controller.abort(new Error("memory_search timed out after 15s")); + + await expect(searchPromise).rejects.toThrow("memory_search timed out after 15s"); + expect(queryChildKill).toHaveBeenCalledWith("SIGKILL"); + await manager.close(); + }); + + it("rejects the qmd search before spawning when the caller signal is already aborted", async () => { + cfg = { + ...cfg, + memory: { + backend: "qmd", + qmd: { + includeDefaultMemory: false, + searchMode: "query", + update: { interval: "0s", debounceMs: 60_000, onBoot: false }, + paths: [{ path: workspaceDir, pattern: "**/*.md", name: "workspace" }], + }, + }, + } as OpenClawConfig; + + const { manager } = await createManager(); + const controller = new AbortController(); + controller.abort(new Error("memory_search timed out after 15s")); + + const callsBefore = spawnMock.mock.calls.filter( + (call: unknown[]) => (call[1] as string[])?.[0] === "query", + ).length; + + await expect( + manager.search("test", { + sessionKey: "agent:main:slack:dm:u123", + signal: controller.signal, + }), + ).rejects.toThrow("memory_search timed out after 15s"); + + const callsAfter = spawnMock.mock.calls.filter( + (call: unknown[]) => (call[1] as string[])?.[0] === "query", + ).length; + expect(callsAfter).toBe(callsBefore); + await manager.close(); + }); + it("does not pass --no-rerank to direct query fallback from search mode", async () => { cfg = { ...cfg, diff --git a/extensions/memory-core/src/memory/qmd-manager.ts b/extensions/memory-core/src/memory/qmd-manager.ts index 4da93d2b1d75..afbc5704b75d 100644 --- a/extensions/memory-core/src/memory/qmd-manager.ts +++ b/extensions/memory-core/src/memory/qmd-manager.ts @@ -90,6 +90,23 @@ type SqliteDatabase = import("node:sqlite").DatabaseSync; const log = createSubsystemLogger("memory"); +/** + * Normalize an already-aborted search signal into the error thrown before any + * qmd work starts. Prefers the caller-supplied abort reason (so a deadline + * message such as "memory_search timed out after 15s" survives) and falls back + * to a stable abort error. + */ +function asAbortError(signal: AbortSignal): Error { + const reason = signal.reason; + if (reason instanceof Error) { + return reason; + } + if (typeof reason === "string" && reason.length > 0) { + return new Error(reason); + } + return new Error("qmd search aborted"); +} + const SNIPPET_HEADER_RE = /@@\s*-([0-9]+),([0-9]+)/; const SEARCH_PENDING_UPDATE_WAIT_MS = 500; const MAX_QMD_OUTPUT_CHARS = 200_000; @@ -1280,12 +1297,23 @@ export class QmdMemoryManager implements MemorySearchManager { qmdSearchModeOverride?: "query" | "search" | "vsearch"; onDebug?: (debug: MemorySearchRuntimeDebug) => void; sources?: MemorySource[]; + /** + * Caller-owned cancellation. When the caller stops waiting (e.g. the + * memory_search tool deadline fires), abort kills the in-flight qmd + * subprocess instead of leaving it running orphaned for the full qmd + * timeout. + */ + signal?: AbortSignal; }, ): Promise { if (!this.isScopeAllowed(opts?.sessionKey)) { this.logScopeDenied(opts?.sessionKey); return []; } + const searchSignal = opts?.signal; + if (searchSignal?.aborted) { + throw asAbortError(searchSignal); + } const trimmed = query.trim(); if (!trimmed) { return []; @@ -1374,7 +1402,7 @@ export class QmdMemoryManager implements MemorySearchManager { } const args = this.buildSearchArgs(qmdSearchCommand, trimmed, limit); args.push(...this.buildCollectionFilterArgs(collectionGroups[0] ?? collectionNames)); - return await this.runQmdSearch(args, qmdSearchCommand); + return await this.runQmdSearch(args, qmdSearchCommand, searchSignal); } catch (err) { if (allowMissingCollectionRepair && this.isMissingCollectionSearchError(err)) { throw err; @@ -1403,7 +1431,7 @@ export class QmdMemoryManager implements MemorySearchManager { fallbackArgs.push( ...this.buildCollectionFilterArgs(collectionGroups[0] ?? collectionNames), ); - return await this.runQmdSearch(fallbackArgs, "query"); + return await this.runQmdSearch(fallbackArgs, "query", searchSignal); } catch (fallbackErr) { log.warn(`qmd query fallback failed: ${String(fallbackErr)}`); throw fallbackErr instanceof Error ? fallbackErr : new Error(String(fallbackErr)); @@ -2137,7 +2165,7 @@ export class QmdMemoryManager implements MemorySearchManager { private async runQmd( args: string[], - opts?: { timeoutMs?: number; discardOutput?: boolean }, + opts?: { timeoutMs?: number; discardOutput?: boolean; signal?: AbortSignal }, ): Promise<{ stdout: string; stderr: string }> { return await runCliCommand({ commandSummary: `qmd ${args.join(" ")}`, @@ -2153,15 +2181,17 @@ export class QmdMemoryManager implements MemorySearchManager { maxOutputChars: this.maxQmdOutputChars, // Large `qmd update` runs can easily exceed the output cap; keep only stderr. discardStdout: opts?.discardOutput, + signal: opts?.signal, }); } private async runQmdSearch( args: string[], command: "query" | "search" | "vsearch", + signal?: AbortSignal, ): Promise { try { - const result = await this.runQmd(args, { timeoutMs: this.qmd.limits.timeoutMs }); + const result = await this.runQmd(args, { timeoutMs: this.qmd.limits.timeoutMs, signal }); return parseQmdQueryJson(result.stdout, result.stderr); } catch (err) { const recovered = this.parseFailedQmdSearchJson(err, command); diff --git a/packages/memory-host-sdk/src/host/qmd-process.test.ts b/packages/memory-host-sdk/src/host/qmd-process.test.ts index f067204980d3..26c1d4175c2e 100644 --- a/packages/memory-host-sdk/src/host/qmd-process.test.ts +++ b/packages/memory-host-sdk/src/host/qmd-process.test.ts @@ -277,7 +277,6 @@ describe("checkQmdBinaryAvailability", () => { expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); }); - it("kills timed-out availability probes by process group on POSIX", async () => { platformSpy?.mockReturnValue("linux"); const killProcess = vi.spyOn(process, "kill").mockImplementation(() => true); @@ -429,6 +428,52 @@ describe("runCliCommand", () => { expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); }); + it("kills aborted cli command process groups on POSIX and rejects with the abort reason", async () => { + platformSpy?.mockReturnValue("linux"); + const killProcess = vi.spyOn(process, "kill").mockImplementation(() => true); + const child = createMockChild({ pid: 7654 }); + spawnMock.mockReturnValueOnce(child); + const controller = new AbortController(); + + try { + const pending = runCliCommand({ + commandSummary: "qmd query slow", + spawnInvocation: { command: "qmd", argv: ["query", "slow", "--json"] }, + env: process.env, + cwd: tempDir, + maxOutputChars: 10_000, + timeoutMs: 60_000, + signal: controller.signal, + }); + + controller.abort(new Error("memory_search timed out after 15s")); + + await expect(pending).rejects.toThrow("memory_search timed out after 15s"); + expect(spawnMock.mock.calls[0]?.[2]).toMatchObject({ detached: true }); + expect(killProcess).toHaveBeenCalledWith(-7654, "SIGKILL"); + expect(child.kill).not.toHaveBeenCalledWith("SIGKILL"); + } finally { + killProcess.mockRestore(); + } + }); + + it("rejects immediately without spawning when the signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(new Error("memory_search timed out after 15s")); + + await expect( + runCliCommand({ + commandSummary: "qmd query test", + spawnInvocation: { command: "qmd", argv: ["query", "test", "--json"] }, + env: process.env, + cwd: tempDir, + maxOutputChars: 10_000, + signal: controller.signal, + }), + ).rejects.toThrow("memory_search timed out after 15s"); + expect(spawnMock).not.toHaveBeenCalled(); + }); + it("kills timed-out cli command process groups on POSIX", async () => { platformSpy?.mockReturnValue("linux"); const killProcess = vi.spyOn(process, "kill").mockImplementation(() => true); diff --git a/packages/memory-host-sdk/src/host/qmd-process.ts b/packages/memory-host-sdk/src/host/qmd-process.ts index 94349e57f225..798da9937e06 100644 --- a/packages/memory-host-sdk/src/host/qmd-process.ts +++ b/packages/memory-host-sdk/src/host/qmd-process.ts @@ -156,6 +156,22 @@ function validateQmdProbeCwd(cwd: string): QmdBinaryAvailability | null { } } +/** + * Normalize an aborted signal into the error used to reject a killed command. + * Prefers the caller-supplied abort reason (so a deadline message survives) and + * falls back to a stable per-command abort error. + */ +function abortReason(signal: AbortSignal | undefined, commandSummary: string): Error { + const reason = signal?.reason; + if (reason instanceof Error) { + return reason; + } + if (typeof reason === "string" && reason.length > 0) { + return new Error(reason); + } + return new Error(`${commandSummary} aborted`); +} + export async function runCliCommand(params: { commandSummary: string; spawnInvocation: CliSpawnInvocation; @@ -164,8 +180,20 @@ export async function runCliCommand(params: { timeoutMs?: number; maxOutputChars: number; discardStdout?: boolean; + /** + * Caller-owned cancellation. When the signal aborts, the spawned child is + * killed immediately and the call rejects, so a caller that already stopped + * waiting (for example after its own deadline) does not leave an orphaned + * process running for the full command timeout. + */ + signal?: AbortSignal; }): Promise<{ stdout: string; stderr: string }> { return await new Promise((resolve, reject) => { + const { signal } = params; + if (signal?.aborted) { + reject(abortReason(signal, params.commandSummary)); + return; + } const child = spawn(params.spawnInvocation.command, params.spawnInvocation.argv, { env: params.env, cwd: params.cwd, @@ -177,15 +205,34 @@ export async function runCliCommand(params: { let stderr = ""; let stdoutTruncated = false; let stderrTruncated = false; + let settled = false; const discardStdout = params.discardStdout === true; const timeoutMs = params.timeoutMs === undefined ? undefined : resolveSafeTimeoutDelayMs(params.timeoutMs); const timer = timeoutMs ? setTimeout(() => { signalQmdProcessTree(child, "SIGKILL"); - reject(new Error(`${params.commandSummary} timed out after ${timeoutMs}ms`)); + settle(() => + reject(new Error(`${params.commandSummary} timed out after ${timeoutMs}ms`)), + ); }, timeoutMs) : null; + const onAbort = () => { + signalQmdProcessTree(child, "SIGKILL"); + settle(() => reject(abortReason(signal, params.commandSummary))); + }; + function settle(run: () => void): void { + if (settled) { + return; + } + settled = true; + if (timer) { + clearTimeout(timer); + } + signal?.removeEventListener("abort", onAbort); + run(); + } + signal?.addEventListener("abort", onAbort, { once: true }); child.stdout.on("data", (data) => { if (discardStdout) { return; @@ -203,33 +250,35 @@ export async function runCliCommand(params: { if (timer) { clearTimeout(timer); } - reject(err); + settle(() => reject(err)); }); - child.on("close", (code, signal) => { + child.on("close", (code, closeSignal) => { if (timer) { clearTimeout(timer); } - if (!discardStdout && (stdoutTruncated || stderrTruncated)) { - reject( - new Error( - `${params.commandSummary} produced too much output (limit ${params.maxOutputChars} chars)`, - ), - ); - return; - } - if (code === 0) { - resolve({ stdout, stderr }); - } else { - reject( - new CliCommandError({ - commandSummary: params.commandSummary, - code, - signal: signal ?? null, - stdout, - stderr, - }), - ); - } + settle(() => { + if (!discardStdout && (stdoutTruncated || stderrTruncated)) { + reject( + new Error( + `${params.commandSummary} produced too much output (limit ${params.maxOutputChars} chars)`, + ), + ); + return; + } + if (code === 0) { + resolve({ stdout, stderr }); + } else { + reject( + new CliCommandError({ + commandSummary: params.commandSummary, + code, + signal: closeSignal ?? null, + stdout, + stderr, + }), + ); + } + }); }); }); }