fix(offload): resolve session manager in assemble/afterTurn without sessionKey

MemoryCore/src/offload/hooks/after-tool-call.ts

@@ -34,11 +34,8 @@ import type { PluginConfig, PluginLogger, ToolPair } from "../types.js";
 import type { BackendClient } from "../backend-client.js";
 import {
   buildL3TriggerReport,
-  classifyPatchEffectiveness,
   reportL3Trigger,
   recordToolCall,
-  REPORT_TYPE_L3,
-  L3_FIXED_PATCH_COST_TOKENS,
 } from "../state-reporter.js";
 
 function isHeartbeatToolCall(event: any, cachedParams: any): boolean {
@@ -97,6 +94,9 @@ export function createAfterToolCallHandler(
   getContextWindow: (() => number) | undefined,
   pluginConfig: Partial<PluginConfig> | undefined,
   backendClient?: BackendClient | null,
+  /** Official OpenClaw API to fetch session messages (issue #851) — used when
+   *  the hook event does not carry messages (no dist-file patch). */
+  getSessionMessages?: (opts: { sessionKey: string; limit?: number }) => Promise<unknown[]>,
 ) {
   return async (event: any, ctx: any) => {
     // Skip internal memory-pipeline sessions
@@ -114,37 +114,18 @@ export function createAfterToolCallHandler(
     const hasMsgs = msgsValue && Array.isArray(msgsValue);
     logger.debug?.(`[context-offload] after_tool_call event keys=[${eventKeys.join(",")}], hasMsgsKey=${hasMsgsKey}, msgsType=${typeof msgsValue}, isArray=${Array.isArray(msgsValue)}, len=${hasMsgs ? msgsValue.length : "N/A"}`);
 
-    // ── Patch-effectiveness detection ──
-    // The upstream runtime patch is expected to populate event.messages with
-    // the current conversation. If it is missing/empty the patch is NOT in
-    // effect and L3 compression cannot run from this hook. Report that
-    // explicitly so operators can detect misconfigurations.
-    const _patchStatus = classifyPatchEffectiveness(event, "after_tool_call");
-    if (_patchStatus.status !== "effective") {
-      logger.warn(
-        `[context-offload] after_tool_call patch check: NOT EFFECTIVE (status=${_patchStatus.status}). ` +
-        `event.messages is ${Array.isArray(msgsValue) ? "empty array" : typeof msgsValue}. ` +
-        `L3 compression will be skipped this turn.`,
-      );
-      if (backendClient) {
-        try {
-          backendClient
-            .storeState({
-              reportType: REPORT_TYPE_L3,
-              reportedAt: new Date().toISOString(),
-              sessionKey: _sk ?? null,
-              stage: "after_tool_call",
-              triggerReason: "patch_not_effective",
-              patch: _patchStatus,
-              pluginState: {
-                l15Settled: stateManager.l15Settled === true,
-                pendingCount: stateManager.getPendingCount(),
-                activeMmdFile: stateManager.getActiveMmdFile?.() ?? null,
-              },
-              fixedPatchCostTokens: L3_FIXED_PATCH_COST_TOKENS,
-            })
-            .catch((err) => logger.warn(`[context-offload] patch-miss report failed: ${err}`));
-        } catch { /* ignore */ }
+    // Issue #851: when the hook event has no messages (OpenClaw no longer
+    // injects them via a dist-file patch), fetch them through the official
+    // getSessionMessages API so L3 compression / MMD logic still has context.
+    if (!hasMsgs && _sk && getSessionMessages) {
+      try {
+        const fetched = await getSessionMessages({ sessionKey: _sk, limit: 50 });
+        if (Array.isArray(fetched) && fetched.length > 0) {
+          event.messages = fetched;
+          logger.debug?.(`[context-offload] after_tool_call: fetched ${fetched.length} messages via getSessionMessages`);
+        }
+      } catch (err) {
+        logger.debug?.(`[context-offload] after_tool_call: getSessionMessages failed: ${err instanceof Error ? err.message : String(err)}`);
       }
     }
 

MemoryCore/src/offload/index.ts

@@ -1039,7 +1039,12 @@ export function registerOffload(api: any, offloadConfig: OffloadConfig): void {
         logger.debug?.(`[context-offload] <<< after_tool_call SKIP: no session manager (${Date.now() - _atcStart}ms)`);
         return;
       }
-      const afterToolCallHandler = createAfterToolCallHandler(_mgr, logger, getContextWindow, pCfg, backendClient as any);
+      const afterToolCallHandler = createAfterToolCallHandler(
+        _mgr, logger, getContextWindow, pCfg, backendClient as any,
+        // Official OpenClaw API replaces the dist-file patch for fetching
+        // session messages in after_tool_call (issue #851).
+        (api.runtime?.subagent?.getSessionMessages) as any,
+      );
       await afterToolCallHandler(event, ctx);
       const _handlerDone = Date.now();
       logger.debug?.(`[context-offload] after_tool_call handler done: ${_handlerDone - _atcStart}ms`);
@@ -1397,16 +1402,22 @@ class OffloadContextEngine {
     // Resolve stateManager: prefer params._offloadManager (set by bootstrap),
     // then fall back to SessionRegistry resolve (framework may pass different params objects).
     let stateManager: OffloadStateManager | undefined = params._offloadManager;
-    if (!stateManager && params.sessionKey) {
-      try {
-        const entry = await this._sessions.resolveIfAllowed(params.sessionKey, params.sessionId);
-        if (entry) {
-          stateManager = entry.manager;
-          params._offloadManager = entry.manager; // cache for compact/afterTurn
-          logger.debug?.(`[context-offload] assemble: resolved manager from SessionRegistry for ${params.sessionKey}`);
+    if (!stateManager) {
+      // OpenClaw's afterTurn/assemble call passes sessionId but not always
+      // sessionKey — fall back to those so L1.5 / L3 / MMD still resolve their
+      // session manager (issue #878). Mirrors the compact() resolution path.
+      const sessionKey = params.sessionKey ?? params.sessionId ?? params.sessionTarget?.sessionKey;
+      if (sessionKey) {
+        try {
+          const entry = await this._sessions.resolveIfAllowed(sessionKey, params.sessionId);
+          if (entry) {
+            stateManager = entry.manager;
+            params._offloadManager = entry.manager; // cache for compact/afterTurn
+            logger.debug?.(`[context-offload] assemble: resolved manager from SessionRegistry for ${sessionKey}`);
+          }
+        } catch (err) {
+          logger.warn(`[context-offload] assemble: failed to resolve session ${sessionKey}: ${err}`);
         }
-      } catch (err) {
-        logger.warn(`[context-offload] assemble: failed to resolve session ${params.sessionKey}: ${err}`);
       }
     }
     const pCfg = this._pCfg;
@@ -2254,11 +2265,18 @@ class OffloadContextEngine {
     const logger = this._logger;
     logger.debug?.(`[context-offload] >>> CE.afterTurn CALLED: sessionKey=${_params?.sessionKey ?? "?"}`);
     let stateManager: OffloadStateManager | undefined = _params?._offloadManager;
-    if (!stateManager && _params?.sessionKey && !isInternalMemorySession(_params.sessionKey)) {
-      try {
-        const entry = this._sessions.get(_params.sessionKey);
-        stateManager = entry?.manager;
-      } catch { /* ignore */ }
+    if (!stateManager) {
+      // OpenClaw's afterTurn hook may pass sessionId (or sessionTarget.sessionKey)
+      // without a top-level sessionKey — fall back so per-turn L1 flush still runs
+      // (issue #878). Mirrors the assemble() resolution path.
+      const sessionKey = _params?.sessionKey ?? _params?.sessionId ?? _params?.sessionTarget?.sessionKey;
+      if (sessionKey && !isInternalMemorySession(sessionKey)) {
+        try {
+          const entry = this._sessions.get(sessionKey);
+          stateManager = entry?.manager;
+          if (stateManager) _params!._offloadManager = stateManager; // cache
+        } catch { /* ignore */ }
+      }
     }
     if (!stateManager) return;
     try {

MemoryCore/src/offload/session-resolve.test.ts

@@ -0,0 +1,107 @@
+/**
+ * Tests for #878 — OffloadContextEngine.assemble()/afterTurn() must resolve
+ * their session manager when OpenClaw passes only sessionId (or
+ * sessionTarget.sessionKey) without a top-level sessionKey.
+ *
+ * Previously the fallback guarded on `params.sessionKey`, so on OpenClaw's
+ * afterTurn/assemble path (which does not include sessionKey) stateManager was
+ * never resolved → assemble returned early → L1.5 never settled → MMD
+ * injection / L2 / L3 never ran (#880).
+ */
+
+import { describe, expect, it } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { SessionRegistry } from "./session-registry.js";
+import { _testExports } from "./index.js";
+
+const { OffloadContextEngine } = _testExports;
+
+function makeLogger() {
+  return { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} } as never;
+}
+
+/** Build an engine with a real SessionRegistry rooted at a temp dir. */
+function makeEngine(opts: { dataRoot?: string; backendClient?: unknown } = {}) {
+  const dataRoot = opts.dataRoot ?? mkdtempSync(join(tmpdir(), "offload-878-"));
+  const sessions = new SessionRegistry(dataRoot);
+  const judged: Array<{ sessionKey: string | undefined }> = [];
+
+  const engine = new OffloadContextEngine({
+    sessions,
+    logger: makeLogger(),
+    pCfg: {},
+    getContextWindow: () => 1_000_000,
+    notifyL2NewNullEntries: () => {},
+    clearL2Timeout: () => {},
+    l4State: { pendingResult: null },
+    flushL1: async () => {},
+    backendClient: opts.backendClient ?? null,
+    judgeL15: async (mgr: any, _event: any, ctx: any) => {
+      judged.push({ sessionKey: ctx?.sessionKey });
+      mgr.l15Settled = true;
+      mgr.setMmdInjectionReady?.(true);
+    },
+    disposeL15: () => {},
+  });
+
+  return { engine, sessions, judged, dataRoot };
+}
+
+describe("OffloadContextEngine session resolution (#878)", () => {
+  it("assemble resolves manager from sessionId when sessionKey is absent", async () => {
+    const { engine, judged, dataRoot } = makeEngine({
+      backendClient: { compact: async () => null, summarize: async () => null },
+    });
+    try {
+      // Pre-register a session by its real sessionKey so resolution succeeds.
+      const sessionKey = "agent:test-agent:878-session-1";
+      await engine.bootstrap({ sessionKey, sessionId: "878-session-1" });
+
+      // OpenClaw's afterTurn/assemble path: no top-level sessionKey, only sessionId.
+      const res = await engine.assemble({
+        sessionId: "878-session-1",
+        messages: [{ role: "user", content: "hello" }],
+        tokenBudget: 1_000_000,
+        prompt: "hello world",
+      });
+
+      expect(res).toBeTruthy();
+      // L1.5 judgment fired (assemble reached the trigger point).
+      expect(judged.length).toBeGreaterThan(0);
+    } finally {
+      rmSync(dataRoot, { recursive: true, force: true });
+    }
+  });
+
+  it("afterTurn resolves manager from sessionId and flushes", async () => {
+    const { engine, dataRoot } = makeEngine();
+    try {
+      const sessionKey = "agent:test-agent:878-session-2";
+      await engine.bootstrap({ sessionKey, sessionId: "878-session-2" });
+
+      // No top-level sessionKey — should resolve from sessionId, no throw.
+      await expect(
+        engine.afterTurn({ sessionId: "878-session-2" }),
+      ).resolves.toBeUndefined();
+    } finally {
+      rmSync(dataRoot, { recursive: true, force: true });
+    }
+  });
+
+  it("assemble still skips cleanly when no session can be resolved", async () => {
+    const { engine, judged, dataRoot } = makeEngine();
+    try {
+      const res = await engine.assemble({
+        // Unknown session — nothing registered, no sessionId, no sessionKey.
+        messages: [{ role: "user", content: "hello" }],
+        prompt: "hello",
+      });
+      expect(res.messages).toEqual([{ role: "user", content: "hello" }]);
+      expect(judged.length).toBe(0);
+    } finally {
+      rmSync(dataRoot, { recursive: true, force: true });
+    }
+  });
+});

MemoryCore/src/utils/serial-queue.test.ts

@@ -0,0 +1,98 @@
+/**
+ * Tests for #518 — SerialQueue must not deadlock when a queued task throws
+ * synchronously. Previously drain() called entry.task() directly; a sync throw
+ * escaped before .finally() was registered, leaving running=true forever so
+ * every later add() hung and onIdle() never resolved.
+ */
+
+import { describe, expect, it } from "vitest";
+import { SerialQueue } from "./serial-queue.js";
+
+function timeout(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
+  return Promise.race([
+    promise,
+    timeout(ms).then(() => {
+      throw new Error(`${label}: timed out after ${ms}ms`);
+    }),
+  ]);
+}
+
+describe("SerialQueue sync-throw resilience (#518)", () => {
+  it("recovers and runs subsequent tasks after a synchronous throw", async () => {
+    const queue = new SerialQueue("repro");
+
+    await queue
+      .add(() => {
+        throw new Error("sync failure");
+      })
+      .catch(() => undefined);
+
+    const next = queue.add(async () => "ran");
+    const outcome = await withTimeout(next, 500, "second task");
+
+    expect(outcome).toBe("ran");
+    expect(queue.size).toBe(0);
+    expect(queue.pending).toBe(false);
+    expect(queue.idle).toBe(true);
+  });
+
+  it("keeps FIFO order across a mix of sync-throw and async tasks", async () => {
+    const queue = new SerialQueue("fifo");
+    const order: string[] = [];
+
+    const p1 = queue.add(() => {
+      order.push("t1");
+      return Promise.resolve("ok1");
+    });
+    const p2 = queue.add(() => {
+      throw new Error("t2 boom");
+    }).catch(() => "caught2");
+    const p3 = queue.add(async () => {
+      order.push("t3");
+      return "ok3";
+    });
+
+    const [r1, r2, r3] = await withTimeout(Promise.all([p1, p2, p3]), 500, "all tasks");
+
+    expect(r1).toBe("ok1");
+    expect(r2).toBe("caught2");
+    expect(r3).toBe("ok3");
+    expect(order).toEqual(["t1", "t3"]);
+    expect(queue.idle).toBe(true);
+  });
+
+  it("onIdle() resolves after a sync-throw task is drained", async () => {
+    const queue = new SerialQueue("idle");
+
+    const idleP = queue.onIdle();
+    queue.add(() => {
+      throw new Error("sync");
+    }).catch(() => undefined);
+    await queue.add(async () => "done");
+
+    await withTimeout(idleP, 500, "onIdle");
+    expect(queue.idle).toBe(true);
+  });
+
+  it("does not double-run a task after a sync throw", async () => {
+    const queue = new SerialQueue("no-double");
+    let calls = 0;
+
+    queue.add(() => {
+      calls++;
+      throw new Error("sync boom");
+    }).catch(() => undefined);
+
+    await queue.add(async () => "after");
+    // entry.resolve runs inside .then(), while running=false lands in the
+    // following .finally() — wait on onIdle() for the bookkeeping to finish.
+    await withTimeout(queue.onIdle(), 500, "onIdle");
+
+    expect(calls).toBe(1);
+    expect(queue.idle).toBe(true);
+  });
+});

MemoryCore/src/utils/serial-queue.ts

@@ -105,8 +105,12 @@ export class SerialQueue {
 
     this.debugFn?.(`[queue:${this.name}] dequeued, starting execution (remaining=${this.queue.length})`);
 
-    entry
-      .task()
+    // Invoke the task from a resolved promise instead of calling it directly:
+    // if the task throws synchronously, the call escapes before the .finally()
+    // handler below is registered, leaving `running = true` forever and
+    // deadlocking the queue (issue #518).
+    Promise.resolve()
+      .then(() => entry.task())
       .then((result) => entry.resolve(result))
       .catch((err) => entry.reject(err))
       .finally(() => {