afterTurn: role-preserving message storage + heartbeat filter (volcengine#1340 hand-merged)

Cherry-picked upstream 68e4d89 with manual conflict resolution against
our prior plugin commit acbb7f9 (pinned context injection, recall refactor).

Three changes from volcengine#1340:

1. Role-preserving message storage (the main feature):
   Instead of collapsing entire turns into a single 'user' addSessionMessage
   call, iterate turnMessages and group adjacent same-role messages by
   actual role (user/assistant). Sends one addSessionMessage per group,
   preserving conversation structure for OpenViking. Replaces the previous
   join-everything-as-user logic.

2. Heartbeat filters (defense in depth, not replacement):
   - Session-level: 'if (afterTurnParams.isHeartbeat) return;' early-return
     when the calling code flags the whole turn as a heartbeat.
   - Per-message: HEARTBEAT_RE = /\bHEARTBEAT(?:\.md|_OK)\b/ drops individual
     heartbeat messages within a turn.
   These are narrow (uppercase HEARTBEAT only) and complement — not replace —
   our broader server-side trivial filter at openviking/session/session.py
   _is_trivial_session, which catches lowercase heartbeat/health_check/ping
   for ANY ingest path (CLI, eval, future clients), not just the plugin.

3. extractSingleMessageText helper in text-utils.ts (additive).

Hand-merge decisions vs upstream:

- Imports collision: kept both extractLastAssistantText (ours) and
  extractSingleMessageText (volcengine#1340).
- Backoff guard order: heartbeat early-return placed BEFORE the consecutive-
  failure backoff guard so heartbeats never increment failure counters.
- addSessionMessage call arity fix: volcengine#1340's loop calls 5-arg form
  (id, role, content, agentId, createdAt) which would put agentId in our
  parts? slot. Updated to 6-arg form (id, role, content, undefined, agentId,
  createdAt) matching client.ts signature.
- Inline commit block at the end of volcengine#1340: dropped. Our architecture routes
  commits through doCommitOVSession (separate function with alignment +
  drift detection hooks). Reintroducing the inline commit would conflict
  with that routing.

Co-Authored-By: claude-flow <ruv@ruv.net>

Author: 2 people

examples/openclaw-plugin/context-engine.ts

@@ -10,6 +10,7 @@ import {
   getCaptureDecision,
   extractNewTurnTexts,
   extractLastAssistantText,
+  extractSingleMessageText,
   shouldBypassSession,
 } from "./text-utils.js";
 import {
@@ -812,6 +813,11 @@ export function createMemoryOpenVikingContextEngine(params: {
         return;
       }
 
+      // Heartbeat session — skip entirely (from upstream #1340)
+      if (afterTurnParams.isHeartbeat) {
+        return;
+      }
+
       // Exponential backoff after consecutive failures
       if (_consecutiveCaptureFailures >= MAX_CONSECUTIVE_FAILURES) {
         const backoffMs = BACKOFF_BASE_MS * Math.pow(2, _consecutiveCaptureFailures - MAX_CONSECUTIVE_FAILURES);
@@ -932,19 +938,42 @@ export function createMemoryOpenVikingContextEngine(params: {
           return;
         }
         const client = await getClient();
-        const turnText = newTexts.join("\n");
-        const sanitized = turnText.replace(/<relevant-memories>[\s\S]*?<\/relevant-memories>/gi, " ").replace(/\s+/g, " ").trim();
         const createdAt = pickLatestCreatedAt(turnMessages);
 
-        if (sanitized) {
-          await client.addSessionMessage(OVSessionId, "user", sanitized, undefined, agentId, createdAt);
-        } else {
-          diag("afterTurn_skip", OVSessionId, {
-            reason: "sanitized_empty",
-          });
+        // Group by OV role (user|assistant), merge adjacent same-role.
+        // Drops heartbeat messages and strips <relevant-memories> tags from
+        // user content. From upstream #1340 — preserves conversation
+        // structure when sending to OpenViking instead of collapsing the
+        // turn into a single "user" message.
+        const HEARTBEAT_RE = /\bHEARTBEAT(?:\.md|_OK)\b/;
+        const groups: Array<{ role: "user" | "assistant"; texts: string[] }> = [];
+        for (const msg of turnMessages) {
+          const text = extractSingleMessageText(msg);
+          if (!text) continue;
+          if (HEARTBEAT_RE.test(text)) continue;
+          const role = (msg as Record<string, unknown>).role as string;
+          const ovRole: "user" | "assistant" = role === "assistant" ? "assistant" : "user";
+          const content = ovRole === "user"
+            ? text.replace(/<relevant-memories>[\s\S]*?<\/relevant-memories>/gi, " ").replace(/\s+/g, " ").trim()
+            : text;
+          if (!content) continue;
+          const last = groups[groups.length - 1];
+          if (last && last.role === ovRole) {
+            last.texts.push(content);
+          } else {
+            groups.push({ role: ovRole, texts: [content] });
+          }
+        }
+
+        if (groups.length === 0) {
+          diag("afterTurn_skip", OVSessionId, { reason: "sanitized_empty" });
           return;
         }
 
+        for (const group of groups) {
+          await client.addSessionMessage(OVSessionId, group.role, group.texts.join("\n"), undefined, agentId, createdAt);
+        }
+
         const session = await client.getSession(OVSessionId, agentId);
         const pendingTokens = session.pending_tokens ?? 0;
 

examples/openclaw-plugin/tests/ut/context-engine-afterTurn.test.ts

@@ -182,7 +182,7 @@ describe("context-engine afterTurn()", () => {
     );
   });
 
-  it("stores new messages via addSessionMessage", async () => {
+  it("stores new messages via addSessionMessage with proper roles", async () => {
     const { engine, client } = makeEngine();
 
     const messages = [
@@ -198,10 +198,13 @@ describe("context-engine afterTurn()", () => {
       prePromptMessageCount: 1,
     });
 
-    expect(client.addSessionMessage).toHaveBeenCalledTimes(1);
-    const storedContent = client.addSessionMessage.mock.calls[0][2] as string;
-    expect(storedContent).toContain("hello world");
-    expect(storedContent).toContain("hi there");
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(2);
+    // First call: user message
+    expect(client.addSessionMessage.mock.calls[0][1]).toBe("user");
+    expect(client.addSessionMessage.mock.calls[0][2]).toContain("hello world");
+    // Second call: assistant message
+    expect(client.addSessionMessage.mock.calls[1][1]).toBe("assistant");
+    expect(client.addSessionMessage.mock.calls[1][2]).toContain("hi there");
   });
 
   it("passes the latest non-system message timestamp to addSessionMessage as ISO string", async () => {
@@ -220,12 +223,14 @@ describe("context-engine afterTurn()", () => {
       prePromptMessageCount: 1,
     });
 
-    expect(client.addSessionMessage).toHaveBeenCalledTimes(1);
-    const createdAt = client.addSessionMessage.mock.calls[0][4] as string;
+    // user + assistant + toolResult(→user) = 3 calls (toolResult merges with no adjacent user)
+    expect(client.addSessionMessage).toHaveBeenCalled();
+    const lastCallIdx = client.addSessionMessage.mock.calls.length - 1;
+    const createdAt = client.addSessionMessage.mock.calls[lastCallIdx][4] as string;
     expect(createdAt).toBe("2026-04-01T10:03:00.000Z");
   });
 
-  it("sanitizes <relevant-memories> from stored content", async () => {
+  it("sanitizes <relevant-memories> from user content but not from assistant", async () => {
     const { engine, client } = makeEngine();
 
     const messages = [
@@ -243,6 +248,7 @@ describe("context-engine afterTurn()", () => {
     });
 
     expect(client.addSessionMessage).toHaveBeenCalledTimes(1);
+    expect(client.addSessionMessage.mock.calls[0][1]).toBe("user");
     const storedContent = client.addSessionMessage.mock.calls[0][2] as string;
     expect(storedContent).not.toContain("relevant-memories");
     expect(storedContent).not.toContain("injected memory data");
@@ -391,10 +397,12 @@ describe("context-engine afterTurn()", () => {
       prePromptMessageCount: 0,
     });
 
-    const storedContent = client.addSessionMessage.mock.calls[0][2] as string;
-    expect(storedContent).toContain("src/app.ts");
-    expect(storedContent).toContain("npm install");
-    expect(storedContent).toContain("export const x = 1");
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(2);
+    const userContent = client.addSessionMessage.mock.calls[0][2] as string;
+    const assistantContent = client.addSessionMessage.mock.calls[1][2] as string;
+    expect(userContent).toContain("src/app.ts");
+    expect(userContent).toContain("npm install");
+    expect(assistantContent).toContain("export const x = 1");
   });
 
   it("passes agentId to addSessionMessage", async () => {
@@ -428,6 +436,143 @@ describe("context-engine afterTurn()", () => {
     expect(client.getSession).toHaveBeenCalled();
   });
 
+  it("maps toolResult to user role", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "assistant", content: [
+        { type: "text", text: "running tool" },
+        { type: "toolUse", name: "bash", input: { cmd: "ls" } },
+      ] },
+      { role: "toolResult", toolName: "bash", content: "file1.txt\nfile2.txt" },
+      { role: "assistant", content: "done" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+    });
+
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(3);
+    // assistant → user(toolResult) → assistant
+    expect(client.addSessionMessage.mock.calls[0][1]).toBe("assistant");
+    expect(client.addSessionMessage.mock.calls[1][1]).toBe("user");
+    expect(client.addSessionMessage.mock.calls[1][2]).toContain("[bash result]:");
+    expect(client.addSessionMessage.mock.calls[1][2]).toContain("file1.txt");
+    expect(client.addSessionMessage.mock.calls[2][1]).toBe("assistant");
+  });
+
+  it("merges adjacent same-role messages", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "user", content: "first question" },
+      { role: "user", content: "second question" },
+      { role: "assistant", content: "answer" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+    });
+
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(2);
+    expect(client.addSessionMessage.mock.calls[0][1]).toBe("user");
+    expect(client.addSessionMessage.mock.calls[0][2]).toContain("first question");
+    expect(client.addSessionMessage.mock.calls[0][2]).toContain("second question");
+    expect(client.addSessionMessage.mock.calls[1][1]).toBe("assistant");
+  });
+
+  it("merges adjacent toolResults into one user group", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "assistant", content: [
+        { type: "text", text: "calling tools" },
+        { type: "toolUse", name: "read", input: { path: "a.txt" } },
+      ] },
+      { role: "toolResult", toolName: "read", content: "content of a" },
+      { role: "toolResult", toolName: "write", content: "ok" },
+      { role: "assistant", content: "all done" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+    });
+
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(3);
+    expect(client.addSessionMessage.mock.calls[0][1]).toBe("assistant");
+    // Two toolResults merged into one user call
+    expect(client.addSessionMessage.mock.calls[1][1]).toBe("user");
+    expect(client.addSessionMessage.mock.calls[1][2]).toContain("[read result]:");
+    expect(client.addSessionMessage.mock.calls[1][2]).toContain("[write result]:");
+    expect(client.addSessionMessage.mock.calls[2][1]).toBe("assistant");
+  });
+
+  it("does not sanitize <relevant-memories> from assistant content", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "user", content: "question" },
+      { role: "assistant", content: "Here is context <relevant-memories>data</relevant-memories> end" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+    });
+
+    expect(client.addSessionMessage).toHaveBeenCalledTimes(2);
+    const assistantContent = client.addSessionMessage.mock.calls[1][2] as string;
+    expect(assistantContent).toContain("relevant-memories");
+  });
+
+  it("skips heartbeat messages from being stored", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "user", content: "Read HEARTBEAT.md if it exists (workspace context). Follow it strictly. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK." },
+      { role: "assistant", content: "HEARTBEAT_OK" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+    });
+
+    expect(client.addSessionMessage).not.toHaveBeenCalled();
+  });
+
+  it("skips heartbeat via isHeartbeat flag", async () => {
+    const { engine, client } = makeEngine();
+
+    const messages = [
+      { role: "user", content: "regular message" },
+      { role: "assistant", content: "reply" },
+    ];
+
+    await engine.afterTurn!({
+      sessionId: "s1",
+      sessionFile: "",
+      messages,
+      prePromptMessageCount: 0,
+      isHeartbeat: true,
+    });
+
+    expect(client.addSessionMessage).not.toHaveBeenCalled();
+  });
+
   it("skips store when all new messages are system only", async () => {
     const { engine, client } = makeEngine();
 

examples/openclaw-plugin/text-utils.ts

@@ -422,6 +422,39 @@ function formatToolResultContent(content: unknown): string {
   return "";
 }
 
+/**
+ * Extract text from a single message without a `[role]:` prefix.
+ * Used by afterTurn to send messages with their actual role.
+ */
+export function extractSingleMessageText(msg: unknown): string {
+  if (!msg || typeof msg !== "object") return "";
+  const m = msg as Record<string, unknown>;
+  const role = m.role as string;
+  if (!role || role === "system") return "";
+
+  if (role === "toolResult") {
+    const toolName = typeof m.toolName === "string" ? m.toolName : "tool";
+    const resultText = formatToolResultContent(m.content);
+    return resultText ? `[${toolName} result]: ${resultText}` : "";
+  }
+
+  const content = m.content;
+  if (typeof content === "string") return content.trim();
+  if (Array.isArray(content)) {
+    const parts: string[] = [];
+    for (const block of content) {
+      const b = block as Record<string, unknown>;
+      if (b?.type === "text" && typeof b.text === "string") {
+        parts.push((b.text as string).trim());
+      } else if (b?.type === "toolUse") {
+        parts.push(formatToolUseBlock(b));
+      }
+    }
+    return parts.join("\n");
+  }
+  return "";
+}
+
 /**
  * 提取从 startIndex 开始的新消息（user + assistant + toolResult），返回格式化的文本。
  * 保留 toolUse 完整内容（tool name + input）和 toolResult 完整内容，