fix(openclaw-plugin): run afterTurn auto-capture in background queue

examples/openclaw-plugin/context-engine.ts

@@ -7,6 +7,7 @@ import {
 import {
   trimForLog,
   toJsonLog,
+  summarizeExtractedMemories,
 } from "./memory-ranking.js";
 
 type AgentMessage = {
@@ -51,6 +52,7 @@ type ContextEngine = {
   }) => Promise<IngestBatchResult>;
   afterTurn?: (params: {
     sessionId: string;
+    sessionKey?: string;
     sessionFile: string;
     messages: AgentMessage[];
     prePromptMessageCount: number;
@@ -175,6 +177,109 @@ export function createMemoryOpenVikingContextEngine(params: {
     return typeof key === "string" && key.trim() ? key.trim() : undefined;
   }
 
+  type AfterTurnCaptureParams = {
+    sessionId: string;
+    sessionKey?: string;
+    messages: AgentMessage[];
+    prePromptMessageCount?: number;
+    runtimeContext?: Record<string, unknown>;
+  };
+
+  const captureQueueBySession = new Map<string, Promise<void>>();
+
+  const runAutoCapture = async (afterTurnParams: AfterTurnCaptureParams): Promise<void> => {
+    try {
+      const runtimeSessionKey = extractSessionKey(afterTurnParams.runtimeContext);
+      const OVSessionId = afterTurnParams.sessionKey ?? runtimeSessionKey ?? afterTurnParams.sessionId;
+      const agentId = resolveAgentId(OVSessionId);
+
+      const messages = afterTurnParams.messages ?? [];
+      if (messages.length === 0) {
+        logger.info("openviking: afterTurn skipped (messages=0)");
+        return;
+      }
+
+      const start =
+        typeof afterTurnParams.prePromptMessageCount === "number" &&
+        afterTurnParams.prePromptMessageCount >= 0
+          ? afterTurnParams.prePromptMessageCount
+          : 0;
+
+      const { texts: newTexts, newCount } = extractNewTurnTexts(messages, start);
+
+      if (newTexts.length === 0) {
+        logger.info("openviking: afterTurn skipped (no new user/assistant messages)");
+        return;
+      }
+
+      // Always store messages into OV session so assemble can retrieve them.
+      // Capture decision only controls whether we trigger commit (archive+extract).
+      const client = await getClient();
+      const turnText = newTexts.join("\n");
+      const sanitized = turnText.replace(/<relevant-memories>[\s\S]*?<\/relevant-memories>/gi, " ").replace(/\s+/g, " ").trim();
+
+      if (sanitized) {
+        await client.addSessionMessage(OVSessionId, "user", sanitized, agentId);
+        logger.info(
+          `openviking: afterTurn stored ${newCount} msgs in session=${OVSessionId} (${sanitized.length} chars)`,
+        );
+      } else {
+        logger.info("openviking: afterTurn skipped store (sanitized text empty)");
+        return;
+      }
+
+      // Capture decision: controls commit (archive + memory extraction)
+      const decision = getCaptureDecision(turnText, cfg.captureMode, cfg.captureMaxLength);
+      logger.info(
+        `openviking: capture-check shouldCapture=${String(decision.shouldCapture)} reason=${decision.reason}`,
+      );
+
+      if (!decision.shouldCapture) {
+        logger.info("openviking: afterTurn skipped commit (capture decision rejected)");
+        return;
+      }
+
+      const session = await client.getSession(OVSessionId, agentId);
+      const pendingTokens = session.pending_tokens ?? 0;
+
+      if (pendingTokens < cfg.commitTokenThreshold) {
+        logger.info(
+          `openviking: pending_tokens=${pendingTokens}/${cfg.commitTokenThreshold} in session=${OVSessionId}, deferring commit`,
+        );
+        return;
+      }
+
+      logger.info(
+        `openviking: committing session=${OVSessionId} (wait=false), pendingTokens=${pendingTokens}, threshold=${cfg.commitTokenThreshold}`,
+      );
+      const commitResult = await client.commitSession(OVSessionId, { wait: false, agentId });
+      logger.info(
+        `openviking: committed session=${OVSessionId}, ` +
+          `status=${commitResult.status}, archived=${commitResult.archived ?? false}, ` +
+          `task_id=${commitResult.task_id ?? "none"} ${toJsonLog({ captured: [trimForLog(turnText, 260)] })}`,
+      );
+    } catch (err) {
+      warnOrInfo(logger, `openviking: auto-capture failed: ${String(err)}`);
+    }
+  };
+
+  const enqueueAutoCapture = (afterTurnParams: AfterTurnCaptureParams): void => {
+    const queueKey = afterTurnParams.sessionKey || afterTurnParams.sessionId;
+    const previous = captureQueueBySession.get(queueKey) ?? Promise.resolve();
+    const next = previous
+      .catch(() => {})
+      .then(() => runAutoCapture(afterTurnParams))
+      .catch((err) => {
+        warnOrInfo(logger, `openviking: queued auto-capture failed: ${String(err)}`);
+      })
+      .finally(() => {
+        if (captureQueueBySession.get(queueKey) === next) {
+          captureQueueBySession.delete(queueKey);
+        }
+      });
+    captureQueueBySession.set(queueKey, next);
+  };
+
   return {
     info: {
       id,
@@ -214,55 +319,13 @@ export function createMemoryOpenVikingContextEngine(params: {
         return;
       }
 
-      try {
-        const sessionKey = extractSessionKey(afterTurnParams.runtimeContext);
-        const agentId = resolveAgentId(sessionKey ?? afterTurnParams.sessionId);
-
-        const messages = afterTurnParams.messages ?? [];
-        if (messages.length === 0) {
-          logger.info("openviking: auto-capture skipped (messages=0)");
-          return;
-        }
-
-        const start =
-          typeof afterTurnParams.prePromptMessageCount === "number" &&
-          afterTurnParams.prePromptMessageCount >= 0
-            ? afterTurnParams.prePromptMessageCount
-            : 0;
-
-        const { texts: newTexts, newCount } = extractNewTurnTexts(messages, start);
-
-        if (newTexts.length === 0) {
-          logger.info("openviking: auto-capture skipped (no new user/assistant messages)");
-          return;
-        }
-
-        const turnText = newTexts.join("\n");
-        const decision = getCaptureDecision(turnText, cfg.captureMode, cfg.captureMaxLength);
-        const preview = turnText.length > 80 ? `${turnText.slice(0, 80)}...` : turnText;
-        logger.info(
-          "openviking: capture-check " +
-            `shouldCapture=${String(decision.shouldCapture)} ` +
-            `reason=${decision.reason} newMsgCount=${newCount} text=\"${preview}\"`,
-        );
-
-        if (!decision.shouldCapture) {
-          logger.info("openviking: auto-capture skipped (capture decision rejected)");
-          return;
-        }
-
-        const client = await getClient();
-        const OVSessionId = sessionKey ?? afterTurnParams.sessionId;
-        await client.addSessionMessage(OVSessionId, "user", decision.normalizedText, agentId);
-        const commitResult = await client.commitSession(OVSessionId, { wait: true, agentId });
-        logger.info(
-          `openviking: committed ${newCount} messages in session=${OVSessionId}, ` +
-            `archived=${commitResult.archived ?? false}, memories=${commitResult.memories_extracted ?? 0}, ` +
-            `task_id=${commitResult.task_id ?? "none"} ${toJsonLog({ captured: [trimForLog(turnText, 260)] })}`,
-        );
-      } catch (err) {
-        warnOrInfo(logger, `openviking: auto-capture failed: ${String(err)}`);
-      }
+      enqueueAutoCapture({
+        sessionId: afterTurnParams.sessionId,
+        sessionKey: afterTurnParams.sessionKey,
+        messages: afterTurnParams.messages ?? [],
+        prePromptMessageCount: afterTurnParams.prePromptMessageCount,
+        runtimeContext: afterTurnParams.runtimeContext,
+      });
     },
 
     async compact(compactParams): Promise<CompactResult> {