diff --git a/extensions/active-memory/index.test.ts b/extensions/active-memory/index.test.ts index 731794606df6..f5eacd842328 100644 --- a/extensions/active-memory/index.test.ts +++ b/extensions/active-memory/index.test.ts @@ -15,6 +15,7 @@ import { appendSessionTranscriptMessageByIdentity } from "openclaw/plugin-sdk/se import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { applyCliRuntimeRecallTimeoutDefault } from "./config.js"; import plugin, { testing } from "./index.js"; +import { resolveActiveRecallForRun } from "./recall-state.js"; import { hasRememberAcrossConversationsAgent } from "./session-policy.js"; // Match only lone surrogates so valid supplementary-plane characters remain allowed. @@ -715,12 +716,13 @@ describe("active-memory plugin", () => { stateDir = ""; }); - it("registers a before_prompt_build hook", () => { + it("registers prompt-build and run-cleanup hooks", () => { const [hookName, handler, options] = firstHookRegistration(); expect(hookName).toBe("before_prompt_build"); expect(typeof handler).toBe("function"); expect(options).toEqual({ timeoutMs: 153_000 }); expect(hookOptions.before_prompt_build?.timeoutMs).toBe(153_000); + expect(typeof hooks.agent_end).toBe("function"); }); it("does not synthesize a main agent when every configured agent opts out", () => { @@ -847,6 +849,178 @@ describe("active-memory plugin", () => { expect(secondSessionKey).not.toBe(firstSessionKey); }); + it("shares recall results across changed-prompt retries in one run", async () => { + const context = { + runId: "run-changed-prompt-retry", + sessionKey: "agent:main:changed-prompt-retry", + }; + + const first = await runPromptBuild({ prompt: "what wings should i order?" }, context); + const second = await runPromptBuild( + { prompt: "actually, what did I order last time?" }, + context, + ); + + expectPrependContextContains(first, "lemon pepper wings"); + expectPrependContextContains(second, "lemon pepper wings"); + expect(runEmbeddedAgent).toHaveBeenCalledTimes(1); + }); + + it("joins concurrent recall attempts in one run", async () => { + let releaseRecall: () => void = () => { + throw new Error("recall gate was not initialized"); + }; + const recallGate = new Promise((resolve) => { + releaseRecall = resolve; + }); + runEmbeddedAgent.mockImplementationOnce(async (params: { sessionFile: string }) => { + await recallGate; + await writeUsableMemoryTranscript(params.sessionFile, "lemon pepper wings"); + return { payloads: [{ text: "- lemon pepper wings" }] }; + }); + const context = { + runId: "run-concurrent-attempts", + sessionKey: "agent:main:concurrent-attempts", + }; + + const first = runPromptBuild({ prompt: "what wings should i order?" }, context); + await vi.waitFor(() => { + expect(runEmbeddedAgent).toHaveBeenCalledTimes(1); + }); + const second = runPromptBuild({ prompt: "what wings did I order last time?" }, context); + await new Promise((resolve) => { + setImmediate(resolve); + }); + + expect(runEmbeddedAgent).toHaveBeenCalledTimes(1); + releaseRecall(); + const [firstResult, secondResult] = await Promise.all([first, second]); + expectPrependContextContains(firstResult, "lemon pepper wings"); + expectPrependContextContains(secondResult, "lemon pepper wings"); + }); + + it("waits for timeout cleanup before replacing a recall in the same run", async () => { + testing.setMinimumTimeoutMsForTests(1); + testing.setSetupGraceTimeoutMsForTests(0); + registerPluginConfig({ timeoutMs: 100, logging: true }); + let releaseCleanup: () => void = () => { + throw new Error("cleanup gate was not initialized"); + }; + const cleanupGate = new Promise((resolve) => { + releaseCleanup = resolve; + }); + hoisted.closeActiveMemorySearchManager.mockImplementationOnce(async () => { + await cleanupGate; + }); + runEmbeddedAgent.mockImplementationOnce( + async (params: { abortSignal?: AbortSignal }) => await waitForAbort(params.abortSignal), + ); + const context = { + runId: "run-timeout-retry", + sessionKey: "agent:main:timeout-retry", + }; + + await expect( + runPromptBuild({ prompt: "what wings should i order before timeout?" }, context), + ).resolves.toBeUndefined(); + await vi.waitFor(() => { + expect(hoisted.closeActiveMemorySearchManager).toHaveBeenCalledTimes(1); + }); + + const retry = runPromptBuild({ prompt: "what wings should i order after timeout?" }, context); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(runEmbeddedAgent).toHaveBeenCalledTimes(1); + + releaseCleanup(); + await expect(retry).resolves.toEqual( + expect.objectContaining({ prependContext: expect.stringContaining("lemon pepper wings") }), + ); + expect(runEmbeddedAgent).toHaveBeenCalledTimes(2); + }); + + it("evicts a rejected replacement after timeout cleanup settles", async () => { + let releaseCleanup: () => void = () => { + throw new Error("cleanup gate was not initialized"); + }; + const cleanupGate = new Promise((resolve) => { + releaseCleanup = resolve; + }); + const initialResult = { status: "timeout" as const, elapsedMs: 1, summary: null }; + await expect( + resolveActiveRecallForRun("run-rejected-replacement", async (onTimeoutCleanup) => { + onTimeoutCleanup(cleanupGate); + return initialResult; + }), + ).resolves.toEqual(initialResult); + + let rejectedReplacementStarts = 0; + const rejectedReplacement = resolveActiveRecallForRun("run-rejected-replacement", async () => { + rejectedReplacementStarts++; + throw new Error("retry deadline expired"); + }); + releaseCleanup(); + await expect(rejectedReplacement).rejects.toThrow("retry deadline expired"); + + const freshResult = { + status: "ok" as const, + elapsedMs: 2, + rawReply: "recovered", + summary: "recovered", + }; + let freshStarts = 0; + await expect( + resolveActiveRecallForRun("run-rejected-replacement", async () => { + freshStarts++; + return freshResult; + }), + ).resolves.toEqual(freshResult); + expect({ freshStarts, rejectedReplacementStarts }).toEqual({ + freshStarts: 1, + rejectedReplacementStarts: 1, + }); + }); + + it("deduplicates cache-disabled private recall until the run ends", async () => { + setMemorySlot("memory-core"); + configFile = { + ...configFile, + agents: { + defaults: { + model: { primary: "github-copilot/gpt-5.4-mini" }, + }, + list: [ + { + id: "main", + memory: { search: { rememberAcrossConversations: true } }, + }, + ], + }, + }; + const context = { + runId: "run-private-recall", + sessionKey: "agent:main:telegram:direct:owner", + messageProvider: "telegram", + channelId: "owner", + }; + seedSession(context.sessionKey, "s-private-recall", 0); + + await runPromptBuild({ prompt: "what wings should i order?" }, context); + await runPromptBuild({ prompt: "what wings should i order?" }, context); + + expect(lastEmbeddedRunParams().conversationRecall).toEqual({ + anchorSessionKey: context.sessionKey, + scope: "same-agent-private", + corpus: "configured", + }); + expect(runEmbeddedAgent).toHaveBeenCalledTimes(1); + + await requireHook("agent_end")({ runId: context.runId, messages: [], success: true }, context); + await runPromptBuild({ prompt: "what wings should i order?" }, context); + expect(runEmbeddedAgent).toHaveBeenCalledTimes(2); + }); + it("retries transient SQLite recall cleanup failures", async () => { hoisted.cleanupSessionLifecycleArtifacts.mockRejectedValueOnce( new Error("session store is busy"), diff --git a/extensions/active-memory/index.ts b/extensions/active-memory/index.ts index 51a4b47e6768..b0d06a1b7179 100644 --- a/extensions/active-memory/index.ts +++ b/extensions/active-memory/index.ts @@ -23,6 +23,7 @@ import { buildQuery, buildSearchQuery, extractRecentTurns, getModelRef } from ". import { buildCacheKey, buildCircuitBreakerKey, + forgetActiveRecallRun, getCachedResult, getCircuitBreakerEntry, isCircuitBreakerOpen, @@ -459,6 +460,7 @@ export default definePluginEntry({ currentModelId: ctx.modelId, conversationRecall, abortSignal: deadlineController.signal, + runId: ctx.runId, }); deadlineController.signal.throwIfAborted(); if (!result.summary) { @@ -493,6 +495,9 @@ export default definePluginEntry({ }, { timeoutMs: beforePromptBuildTimeoutMs }, ); + api.on("agent_end", (event, ctx) => { + forgetActiveRecallRun(event.runId ?? ctx.runId); + }); }, }); diff --git a/extensions/active-memory/recall-state.ts b/extensions/active-memory/recall-state.ts index 02e1d4829252..76af3383410a 100644 --- a/extensions/active-memory/recall-state.ts +++ b/extensions/active-memory/recall-state.ts @@ -18,6 +18,11 @@ import { let lastActiveRecallCacheSweepAt = 0; const activeRecallCache = new Map(); +type ActiveRecallRunEntry = { + promise: Promise; + timeoutCleanup?: Promise; +}; +const activeRecallRuns = new Map(); const timeoutCircuitBreaker = new Map(); function buildCircuitBreakerKey(agentId: string, provider?: string, model?: string): string { @@ -55,22 +60,74 @@ function scheduleMemorySearchCleanupAfterTimeout( api: OpenClawPluginApi, logPrefix: string, agentId: string, -): void { - const cfg = resolveActiveMemoryCleanupConfig(api); - setTimeout(() => { - void closeActiveMemorySearchManager({ cfg: cfg ?? api.config, agentId }) - .then(() => { - api.logger.debug?.(`${logPrefix} released memory search managers after timeout`); - }) - .catch((error: unknown) => { - const message = toSingleLineLogValue( - error instanceof Error ? error.message : String(error), - ); - api.logger.warn?.( - `${logPrefix} failed to release memory search managers after timeout: ${message}`, - ); - }); - }, 0); +): Promise { + return new Promise((resolve) => { + const cfg = resolveActiveMemoryCleanupConfig(api); + setTimeout(() => { + void closeActiveMemorySearchManager({ cfg: cfg ?? api.config, agentId }) + .then(() => { + api.logger.debug?.(`${logPrefix} released memory search managers after timeout`); + }) + .catch((error: unknown) => { + const message = toSingleLineLogValue( + error instanceof Error ? error.message : String(error), + ); + api.logger.warn?.( + `${logPrefix} failed to release memory search managers after timeout: ${message}`, + ); + }) + .finally(resolve); + }, 0); + }); +} + +async function resolveActiveRecallForRun( + runId: string, + start: (onTimeoutCleanup: (cleanup: Promise) => void) => Promise, +): Promise { + const existing = activeRecallRuns.get(runId); + if (existing?.timeoutCleanup) { + // A replacement must not reuse managers while the timed-out recall or its + // cleanup is still settling; concurrent callers then join the replacement. + await Promise.allSettled([existing.promise, existing.timeoutCleanup]); + if (activeRecallRuns.get(runId) === existing) { + activeRecallRuns.delete(runId); + } + return await resolveActiveRecallForRun(runId, start); + } + if (existing) { + return await existing.promise; + } + + const entry: ActiveRecallRunEntry = { + promise: Promise.resolve().then(() => + start((cleanup) => { + entry.timeoutCleanup = cleanup; + void Promise.allSettled([entry.promise, cleanup]).then(() => { + if (activeRecallRuns.get(runId) === entry) { + activeRecallRuns.delete(runId); + } + }); + }), + ), + }; + activeRecallRuns.set(runId, entry); + void entry.promise.catch(() => { + // Failures before timeout cleanup starts must not poison this run; + // timeout-backed entries stay registered until manager cleanup settles. + if (!entry.timeoutCleanup && activeRecallRuns.get(runId) === entry) { + activeRecallRuns.delete(runId); + } + }); + // Fulfilled results remain stable through agent_end, including `failed`; + // rerunning them would recreate the redundant same-turn recalls this registry prevents. + return await entry.promise; +} + +function forgetActiveRecallRun(runId: string | undefined): void { + if (runId) { + activeRecallRuns.delete(runId); + } } function buildCacheKey(params: { @@ -172,6 +229,7 @@ function shouldCacheResult(result: ActiveRecallResult): boolean { function resetActiveRecallStateForTests(): void { activeRecallCache.clear(); + activeRecallRuns.clear(); timeoutCircuitBreaker.clear(); lastActiveRecallCacheSweepAt = 0; } @@ -186,9 +244,11 @@ export { getCachedResult, getCircuitBreakerEntry, isCircuitBreakerOpen, + forgetActiveRecallRun, recordCircuitBreakerTimeout, resetActiveRecallStateForTests, resetCircuitBreaker, + resolveActiveRecallForRun, scheduleMemorySearchCleanupAfterTimeout, setCachedResult, shouldCacheResult, diff --git a/extensions/active-memory/recall.ts b/extensions/active-memory/recall.ts index 67a059624626..4ff1ed59c55b 100644 --- a/extensions/active-memory/recall.ts +++ b/extensions/active-memory/recall.ts @@ -11,6 +11,7 @@ import { isCircuitBreakerOpen, recordCircuitBreakerTimeout, resetCircuitBreaker, + resolveActiveRecallForRun, scheduleMemorySearchCleanupAfterTimeout, setCachedResult, shouldCacheResult, @@ -90,7 +91,7 @@ function prepareRecallRunContext(params: { return { parentSessionKey, storePath, fastMode }; } -async function maybeResolveActiveRecall(params: { +type ActiveRecallParams = { api: OpenClawPluginApi; runtimeConfig: OpenClawConfig; config: ResolvedActiveRecallPluginConfig; @@ -105,7 +106,14 @@ async function maybeResolveActiveRecall(params: { currentModelId?: string; conversationRecall?: ConversationRecallContext; abortSignal?: AbortSignal; -}): Promise { + runId?: string; +}; + +async function resolveActiveRecall( + params: Omit & { + onTimeoutCleanup?: (cleanup: Promise) => void; + }, +): Promise { params.abortSignal?.throwIfAborted(); const startedAt = Date.now(); // Memory Core re-authorizes every conversation-recall request against live @@ -169,7 +177,8 @@ async function maybeResolveActiveRecall(params: { return; } timeoutCleanupScheduled = true; - scheduleMemorySearchCleanupAfterTimeout(params.api, logPrefix, params.agentId); + const cleanup = scheduleMemorySearchCleanupAfterTimeout(params.api, logPrefix, params.agentId); + params.onTimeoutCleanup?.(cleanup); }; let circuitBreakerTimeoutRecorded = false; const recordRecallTimeout = () => { @@ -457,4 +466,14 @@ async function maybeResolveActiveRecall(params: { } } +async function maybeResolveActiveRecall(params: ActiveRecallParams): Promise { + const { runId, ...recallParams } = params; + if (!runId) { + return await resolveActiveRecall(recallParams); + } + return await resolveActiveRecallForRun(runId, (onTimeoutCleanup) => + resolveActiveRecall({ ...recallParams, onTimeoutCleanup }), + ); +} + export { maybeResolveActiveRecall }; diff --git a/src/plugins/contracts/boundary-invariants.test.ts b/src/plugins/contracts/boundary-invariants.test.ts index 8faa13821bd8..65829f9a1753 100644 --- a/src/plugins/contracts/boundary-invariants.test.ts +++ b/src/plugins/contracts/boundary-invariants.test.ts @@ -30,7 +30,7 @@ const BUNDLED_TYPED_HOOK_REGISTRATION_FILES = [ ] as const; const BUNDLED_TYPED_HOOK_REGISTRATION_GUARDS = { "extensions/acpx/index.ts": ["reply_dispatch"], - "extensions/active-memory/index.ts": ["before_prompt_build"], + "extensions/active-memory/index.ts": ["agent_end", "before_prompt_build"], "extensions/clickclack/src/discussions/register.ts": ["before_tool_call"], "extensions/codex/index.ts": ["after_compaction", "inbound_claim", "session_end"], "extensions/diffs/src/plugin.ts": ["before_prompt_build"],