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, + }), + ); + } + }); }); }); }