fix(active-memory): share one recall per run across retries (#115627)

* fix(active-memory): share one recall per run across retries

Register in-flight recalls per ctx.runId so overlapping or changed-prompt
attempts join a single execution, and gate replacement recalls on settled
timeout cleanup. Entries clear on agent_end. Fixes #106957.

* fix(active-memory): evict rejected recall entries

* test(active-memory): clarify run result retention

* test(active-memory): complete recall result fixture

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<void>((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<void>((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<void>((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<void>((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<void>((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"),

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

extensions/active-memory/recall-state.ts

@@ -18,6 +18,11 @@ import {
 
 let lastActiveRecallCacheSweepAt = 0;
 const activeRecallCache = new Map<string, CachedActiveRecallResult>();
+type ActiveRecallRunEntry = {
+  promise: Promise<ActiveRecallResult>;
+  timeoutCleanup?: Promise<void>;
+};
+const activeRecallRuns = new Map<string, ActiveRecallRunEntry>();
 const timeoutCircuitBreaker = new Map<string, CircuitBreakerEntry>();
 
 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<void> {
+  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>) => void) => Promise<ActiveRecallResult>,
+): Promise<ActiveRecallResult> {
+  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,

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<ActiveRecallResult> {
+  runId?: string;
+};
+
+async function resolveActiveRecall(
+  params: Omit<ActiveRecallParams, "runId"> & {
+    onTimeoutCleanup?: (cleanup: Promise<void>) => void;
+  },
+): Promise<ActiveRecallResult> {
   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<ActiveRecallResult> {
+  const { runId, ...recallParams } = params;
+  if (!runId) {
+    return await resolveActiveRecall(recallParams);
+  }
+  return await resolveActiveRecallForRun(runId, (onTimeoutCleanup) =>
+    resolveActiveRecall({ ...recallParams, onTimeoutCleanup }),
+  );
+}
+
 export { maybeResolveActiveRecall };

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"],