fix: harden session memory and drain safety

Tighten session-memory rewriting signals, improve warning fallbacks, dedupe repeated event text, and clamp unsafe drain heartbeat settings against the claim TTL.

Author: vicary

README.md

@@ -73,7 +73,8 @@ This plugin uses a two-layer memory architecture:
 - Saves compaction summaries as episodes so knowledge survives across session
   boundaries
 
-No Graphiti call ever blocks a hook return.
+Graphiti stays off the steady-state hook path for cached recall, but fresh
+long-term fact lookups can still run during session memory preparation.
 
 ## Prerequisites
 
@@ -255,8 +256,9 @@ sources:
 - Cached long-term facts from Graphiti
 
 These are composed into a `<session_memory>` envelope and staged for the
-transform hook. All reads are local (sub-ms); no Graphiti call is made at this
-stage.
+transform hook. The baseline reads are local (sub-ms), and Graphiti is only
+consulted here when the plugin asks for fresh long-term facts for the current
+request.
 
 ### User Message Injection (`experimental.chat.messages.transform`)
 
@@ -275,8 +277,8 @@ that produced the cached memory. When Jaccard similarity on cached fact UUIDs
 drops below `driftThreshold` (default 0.5), a background cache refresh is
 scheduled via Graphiti. The current cached context is still injected
 immediately; the refreshed cache becomes available on the next message. This
-trades one message of staleness for keeping Graphiti entirely off the
-response-time path.
+trades one message of staleness for keeping most long-term memory refresh work
+off the response-time path.
 
 ### Event Extraction and Buffering (`event`)
 

src/handlers/messages.test.ts

@@ -1,5 +1,7 @@
 import { assertEquals, assertStringIncludes } from "jsr:@std/assert@^1.0.0";
 import { describe, it } from "jsr:@std/testing@^1.0.0/bdd";
+import { spy } from "jsr:@std/testing@^1.0.0/mock";
+import { logger } from "../services/logger.ts";
 import { createMessagesHandler } from "./messages.ts";
 
 describe("messages handler", () => {
@@ -322,6 +324,63 @@ describe("messages handler", () => {
     );
   });
 
+  it("reports rewroteExistingMemory when canonical or legacy blocks were scrubbed", async () => {
+    const state = {
+      isMain: true,
+      visibleFactUuids: new Set<string>(),
+      pendingInjection: {
+        envelope:
+          '<session_memory version="1"><last_request>next</last_request></session_memory>',
+        factUuids: [],
+        nodeRefs: [],
+        refreshDecision: {
+          classification: "aligned",
+          shouldRefresh: false,
+          similarity: 1,
+          threshold: 0.5,
+          cachedQuery: "next",
+        },
+      },
+    };
+    const infoSpy = spy(logger, "info");
+    try {
+      const handler = createMessagesHandler({
+        sessionManager: {
+          getState() {
+            return state;
+          },
+          prepareInjection() {
+            return state.pendingInjection;
+          },
+        } as never,
+      });
+
+      const output = {
+        messages: [{
+          info: { role: "user", sessionID: "session-1" },
+          parts: [{
+            type: "text",
+            text: '<memory data-uuids="fact-legacy-1"></memory>\n\nnext',
+          }],
+        }],
+      };
+
+      await handler({}, output as never);
+
+      const call = infoSpy.calls.find((entry) =>
+        entry.args[0] === "Injected canonical session_memory block"
+      );
+      assertEquals(Boolean(call), true);
+      assertEquals(
+        (call?.args[1] as { rewroteExistingMemory: boolean })
+          .rewroteExistingMemory,
+        true,
+      );
+    } finally {
+      infoSpy.restore();
+    }
+  });
+
   it("passes current-turn visible fact uuids into prepareInjection", async () => {
     const state = {
       isMain: true,

src/handlers/messages.ts

@@ -79,9 +79,17 @@ export function createMessagesHandler(deps: MessagesHandlerDeps) {
     if (!state?.isMain) return;
 
     const allVisibleUuids = new Set<string>();
+    let rewroteExistingMemory = false;
     for (const entry of output.messages) {
       for (const part of entry.parts) {
         if (isTextPart(part)) {
+          if (
+            !rewroteExistingMemory &&
+            (part.text.includes("<session_memory") ||
+              part.text.includes("<memory"))
+          ) {
+            rewroteExistingMemory = true;
+          }
           for (const uuid of extractVisibleUuids(part.text)) {
             allVisibleUuids.add(uuid);
           }
@@ -106,8 +114,6 @@ export function createMessagesHandler(deps: MessagesHandlerDeps) {
 
     const textPart = lastUserEntry.parts.find(isTextPart);
     if (!textPart) return;
-    const hadCanonical = textPart.text.includes("<session_memory");
-    const hadLegacy = textPart.text.includes("<memory");
     const effectiveUserText = sanitizeMemoryInput(
       stripInjectedMemoryBlocks(textPart.text),
     );
@@ -121,7 +127,7 @@ export function createMessagesHandler(deps: MessagesHandlerDeps) {
     logger.info("Injected canonical session_memory block", {
       sessionID,
       factCount: prepared.factUuids.length,
-      rewroteExistingMemory: hadCanonical || hadLegacy,
+      rewroteExistingMemory,
     });
     if (state.pendingInjection === prepared) {
       state.pendingInjection = undefined;

src/services/batch-drain.test.ts

@@ -1,5 +1,7 @@
 import { assertEquals } from "jsr:@std/assert@^1.0.0";
 import { describe, it } from "jsr:@std/testing@^1.0.0/bdd";
+import { spy } from "jsr:@std/testing@^1.0.0/mock";
+import { logger } from "./logger.ts";
 import { BatchDrainService } from "./batch-drain.ts";
 import { createSessionEvent } from "./event-extractor.ts";
 import { RedisClient } from "./redis-client.ts";
@@ -13,22 +15,69 @@ import {
   RedisEventsService,
 } from "./redis-events.ts";
 
-const createDeps = () => {
+const createDeps = (options?: {
+  events?: { claimLockTtlSeconds?: number };
+  drain?: { claimHeartbeatIntervalMs?: number | null };
+}) => {
   const redis = new RedisClient({ endpoint: "redis://unused" });
   const events = new RedisEventsService(redis, {
     sessionTtlSeconds: 60,
-    claimLockTtlSeconds: 1,
+    claimLockTtlSeconds: options?.events?.claimLockTtlSeconds ?? 1,
   });
-  const drain = new BatchDrainService(redis, events, {
+  const drainOptions = {
     batchSize: 2,
     batchMaxBytes: 20_000,
     drainRetryMax: 2,
-    claimHeartbeatIntervalMs: 100,
-  });
+  };
+  const heartbeatIntervalMs = options?.drain?.claimHeartbeatIntervalMs;
+  const drain = new BatchDrainService(
+    redis,
+    events,
+    heartbeatIntervalMs === null ? drainOptions : {
+      ...drainOptions,
+      claimHeartbeatIntervalMs: heartbeatIntervalMs ?? 100,
+    },
+  );
   return { redis, events, drain };
 };
 
 describe("batch drain", () => {
+  it("uses a sub-TTL default heartbeat when the claim TTL is small", () => {
+    const { drain } = createDeps({
+      events: { claimLockTtlSeconds: 0.2 },
+      drain: { claimHeartbeatIntervalMs: null },
+    });
+
+    const heartbeatIntervalMs = (drain as unknown as {
+      getClaimHeartbeatIntervalMs: (ttl: number) => number;
+    }).getClaimHeartbeatIntervalMs(1);
+
+    assertEquals(heartbeatIntervalMs, 333);
+  });
+
+  it("warns and clamps an explicit heartbeat interval that exceeds the claim TTL budget", () => {
+    const warnSpy = spy(logger, "warn");
+    try {
+      const { drain } = createDeps({
+        events: { claimLockTtlSeconds: 1 },
+        drain: { claimHeartbeatIntervalMs: 1_500 },
+      });
+
+      const heartbeatIntervalMs = (drain as unknown as {
+        getClaimHeartbeatIntervalMs: (ttl: number) => number;
+      }).getClaimHeartbeatIntervalMs(1);
+
+      assertEquals(heartbeatIntervalMs, 500);
+      assertEquals(warnSpy.calls.length, 1);
+      assertEquals(
+        warnSpy.calls[0].args[0],
+        "Clamped drain heartbeat interval to stay below claim TTL",
+      );
+    } finally {
+      warnSpy.restore();
+    }
+  });
+
   it("claims oldest events, drains them FIFO, and leaves newer items pending", async () => {
     const { redis, events, drain } = createDeps();
     const added: string[] = [];

src/services/batch-drain.ts

@@ -82,8 +82,23 @@ export class BatchDrainService {
   ) {}
 
   private getClaimHeartbeatIntervalMs(lockTtlSeconds: number): number {
-    return this.options.claimHeartbeatIntervalMs ??
-      Math.max(1_000, Math.floor((lockTtlSeconds * 1000) / 3));
+    const ttlMs = Math.max(1_000, Math.floor(lockTtlSeconds * 1000));
+    const defaultIntervalMs = Math.max(250, Math.floor(ttlMs / 3));
+    const configuredIntervalMs = this.options.claimHeartbeatIntervalMs;
+    const requestedIntervalMs = configuredIntervalMs ?? defaultIntervalMs;
+    const maxSafeIntervalMs = Math.max(250, Math.floor(ttlMs / 2));
+
+    if (requestedIntervalMs <= maxSafeIntervalMs) {
+      return requestedIntervalMs;
+    }
+
+    logger.warn("Clamped drain heartbeat interval to stay below claim TTL", {
+      claimLockTtlSeconds: lockTtlSeconds,
+      requestedHeartbeatIntervalMs: requestedIntervalMs,
+      effectiveHeartbeatIntervalMs: maxSafeIntervalMs,
+      configuredHeartbeatIntervalMs: configuredIntervalMs,
+    });
+    return maxSafeIntervalMs;
   }
 
   private async getRetryState(

src/services/event-extractor.test.ts

@@ -115,6 +115,24 @@ describe("event-extractor", () => {
     );
   });
 
+  it("dedupes repeated detail fragments in compactParts-backed task updates", () => {
+    const [event] = extractStructuredEvents({
+      eventType: "task.updated",
+      properties: {
+        task: {
+          id: "t1",
+          summary:
+            "yes, keep the review-refine loop until no more issues are found.",
+        },
+      },
+    });
+
+    assertEquals(
+      event.detail,
+      "Task update — yes, keep the review-refine loop until no more issues are found.",
+    );
+  });
+
   it("extracts rules, environment, and subagent signals while filtering assistant operational blocker chatter", () => {
     const rules = extractStructuredEvents({
       eventType: "rules.loaded",

src/services/event-extractor.ts

@@ -286,10 +286,17 @@ const hasKeyword = (
 const compactParts = (
   ...parts: Array<string | undefined>
 ): string | undefined => {
-  const compact = parts
-    .map((part) => part ? normalizeWhitespace(part) : "")
-    .filter(Boolean)
-    .join(" — ");
+  const fragments: string[] = [];
+  for (const part of parts) {
+    const value = part ? normalizeWhitespace(part) : "";
+    if (!value) continue;
+    const normalized = value.toLowerCase();
+    if (fragments.some((fragment) => fragment.toLowerCase() === normalized)) {
+      continue;
+    }
+    fragments.push(value);
+  }
+  const compact = fragments.join(" — ");
   return compact || undefined;
 };
 

src/services/logger.test.ts

@@ -139,6 +139,32 @@ describe("logger", () => {
       }]);
     });
 
+    it("falls back to console.warn when structured warn logging rejects later", async () => {
+      const scheduledTasks: Array<() => void> = [];
+      setWarningTaskScheduler((callback) => {
+        scheduledTasks.push(callback);
+      });
+      setOpenCodeClient({
+        app: {
+          log: () => Promise.reject(new Error("structured warn failed")),
+        },
+      });
+
+      const { logger } = await import("./logger.ts");
+      logger.warn("warning", { code: 42 });
+
+      assertEquals(consoleWarnSpy.calls.length, 0);
+      assertEquals(scheduledTasks.length, 1);
+      for (const task of scheduledTasks) task();
+      await Promise.resolve();
+      assertEquals(consoleWarnSpy.calls.length, 1);
+      assertEquals(consoleWarnSpy.calls[0].args[0], "[graphiti]");
+      assertEquals(consoleWarnSpy.calls[0].args[1], "warning");
+      assertEquals(consoleWarnSpy.calls[0].args[2], {
+        data: [{ code: 42 }],
+      });
+    });
+
     it("should forward multiple arguments to error", async () => {
       const { logger } = await import("./logger.ts");
       const error = new Error("test");

src/services/opencode-warning.ts

@@ -78,15 +78,23 @@ export const logStructuredWarning = (
   const client = getClient();
   if (!client?.app?.log) return false;
 
-  runSafely(() =>
-    client.app!.log({
-      body: {
-        service: "graphiti",
-        level: "warn",
-        message,
-        ...(extra === undefined ? {} : { extra: asRecord(extra) }),
-      },
-    })
+  runSafely(
+    () =>
+      client.app!.log({
+        body: {
+          service: "graphiti",
+          level: "warn",
+          message,
+          ...(extra === undefined ? {} : { extra: asRecord(extra) }),
+        },
+      }),
+    (error) => {
+      if (extra === undefined) {
+        console.warn(PREFIX, message, error);
+        return;
+      }
+      console.warn(PREFIX, message, extra, error);
+    },
   );
   return true;
 };