fix(pi): dedupe active_memory injection across turns (#1007)

src/adapters/pi/extension.ts

@@ -164,6 +164,24 @@ let _buildAutoInjection:
 // which breaks prefix prompt cache on DeepSeek/Anthropic/OpenAI).
 // See: https://github.com/mksglu/context-mode/issues/598
 let _pendingContext = "";
+
+// High-water mark (max session_events.id already shown as active_memory) for
+// the CURRENT session. session_events.id is an AUTOINCREMENT primary key, so
+// it is a stable, monotonically increasing "newness" cursor per event.
+//
+// Issue #1007 — before this, every before_agent_start turn where
+// db.getEvents(..., { minPriority: 3, limit: 50 }) returned ANY rows
+// unconditionally rebuilt _pendingContext from that full set, even when it
+// was identical to what was already injected on a prior turn. The 'context'
+// hook pushes _pendingContext as a transient message that is never persisted
+// into context.messages, so a repeated (but never-before-seen-in-this-exact-
+// form) message lands right where Anthropic's cache_control breakpoint sits —
+// a full cache-miss rewrite instead of a cache-read on every such turn. This
+// mirrors the existing `resume.consumed` / db.markResumeConsumed() pattern
+// (below), which already solves the identical problem for resume snapshots
+// by showing them exactly once. Module-level state matches this file's
+// existing single-session-per-process assumption (_sessionId, _pendingContext).
+let _lastShownActiveMemoryEventId = 0;
 async function getAutoInjection(
   pluginRoot: string,
 ): Promise<((events: Array<{ category: string; data: string }>) => string) | null> {
@@ -467,6 +485,7 @@ export default function piExtension(pi: any): void {
   pi.on("session_start", (_event: any, ctx: any) => {
     try {
       _sessionId = deriveSessionId(ctx ?? {});
+      _lastShownActiveMemoryEventId = 0; // fresh session — no active_memory shown yet
       db.ensureSession(_sessionId, projectDir);
       db.cleanupOldSessions(7);
     } catch {
@@ -681,12 +700,21 @@ export default function piExtension(pi: any): void {
           limit: 50,
         })
         .filter((e: any) => String(e.category ?? "") !== "role");
-      if (activeEvents.length > 0) {
+
+      // Issue #1007 — only consider events that are NEW since the last time
+      // active_memory was injected in this session. If nothing qualifies,
+      // leave _pendingContext untouched by this block (no re-injection, no
+      // cache-miss cost) instead of unconditionally rebuilding the same
+      // content every turn a priority>=3 event happens to still exist.
+      const newActiveEvents = activeEvents.filter(
+          (e: any) => Number(e.id ?? 0) > _lastShownActiveMemoryEventId,
+        );
+      if (newActiveEvents.length > 0) {
         const buildAuto = await getAutoInjection(pluginRoot);
         let memoryContext = "";
         if (buildAuto) {
           memoryContext = buildAuto(
-            activeEvents.map((e: any) => ({
+            newActiveEvents.map((e: any) => ({
               category: String(e.category ?? ""),
               data: String(e.data ?? ""),
             })),
@@ -696,7 +724,7 @@ export default function piExtension(pi: any): void {
         if (!memoryContext) {
           const memoryLines: string[] = ["<active_memory>"];
           let budget = 2000; // ~500 tokens at 4 chars/token
-          for (const ev of activeEvents) {
+          for (const ev of newActiveEvents) {
             const line = `  <event type="${ev.type}" category="${ev.category}">${ev.data}</event>`;
             if (line.length > budget) break;
             memoryLines.push(line);
@@ -705,7 +733,17 @@ export default function piExtension(pi: any): void {
           memoryLines.push("</active_memory>");
           if (memoryLines.length > 2) memoryContext = memoryLines.join("\n");
         }
-        if (memoryContext) parts.push(memoryContext);
+        if (memoryContext) {
+          parts.push(memoryContext);
+          // Advance the high-water mark past every new event we just considered
+          // (whether or not each individual one made it into the capped output —
+          // same one-shot-then-move-on semantics as db.markResumeConsumed below),
+          // so this exact content is never rebuilt/re-injected on a later turn.
+          for (const ev of newActiveEvents) {
+            const id = Number(ev.id ?? 0);
+            if (id > _lastShownActiveMemoryEventId) _lastShownActiveMemoryEventId = id;
+          }
+        }
       }
 
       // Resume snapshot (only when present and unconsumed).
@@ -863,6 +901,7 @@ export default function piExtension(pi: any): void {
       _db = null;
       _dbPath = "";
       _sessionId = "";
+      _lastShownActiveMemoryEventId = 0;
     } catch {
       // best effort — never throw during shutdown
     }

tests/pi-extension.test.ts

@@ -1046,6 +1046,111 @@ describe("Pi Extension", () => {
     });
   });
 
+  // ═══════════════════════════════════════════════════════════
+  // Issue #1007: active_memory injection must not repeat unchanged
+  // content on every turn (prompt-cache regression). The 'context' hook
+  // pushes _pendingContext as a transient, never-persisted message, so
+  // rebuilding it from the SAME session events on every turn creates a
+  // brand-new never-before-cached prefix each time — a cache-miss rewrite
+  // instead of a cache-read. Only genuinely NEW events (since the last time
+  // active_memory was shown) should trigger a re-injection.
+  // ═══════════════════════════════════════════════════════════
+
+  describe("Issue #1007: active_memory dedupe across turns", () => {
+    // Helper: derive the same session id the extension computes internally
+    // (sha256(sessionFile).slice(0,16)) and open the SAME SessionDB file the
+    // extension writes to, so the test can seed priority>=3 events directly
+    // without depending on the user-prompt text extractors (which also emit
+    // an incidental 'intent' event on nearly every non-question prompt).
+    async function openExtensionDb(sessionFile: string, projectDir: string) {
+      const { resolveSessionDbPath } = await import("../src/session/db.js");
+      const sessionsDir = join(process.env.HOME!, ".pi", "context-mode", "sessions");
+      const dbPath = resolveSessionDbPath({ projectDir, sessionsDir });
+      const sessionId = createHash("sha256").update(sessionFile).digest("hex").slice(0, 16);
+      return { db: new SessionDB({ dbPath }), sessionId };
+    }
+
+    it("does NOT re-inject the same active_memory content on a turn with no new qualifying events", async () => {
+      await registerPiExtension(api);
+      const sessionFile = `dedupe-1007-${Date.now()}-${Math.random()}`;
+      await api._trigger(
+        "session_start",
+        {},
+        { sessionManager: { getSessionFile: () => sessionFile } },
+      );
+
+      const { db, sessionId } = await openExtensionDb(sessionFile, tempDir);
+      db.insertEvent(sessionId, {
+        type: "external_ref",
+        category: "external-ref",
+        data: "https://example.com/ticket-1",
+        priority: 3,
+      });
+
+      // First turn: the event above qualifies (priority >= 3, category != role)
+      // and is injected as active_memory.
+      await api._trigger("before_agent_start", { systemPrompt: "Base." });
+      const firstCtx = await api._trigger("context", { messages: [] });
+      const firstContent = String(firstCtx?.messages?.[0]?.content ?? "");
+      expect(firstContent).toContain("https://example.com/ticket-1");
+
+      // Second turn: no new event was recorded since the first injection.
+      // Pre-#1007-fix, db.getEvents(...) would return the SAME event and
+      // _pendingContext would be unconditionally rebuilt from it — a
+      // never-before-seen-in-this-exact-form message that breaks prompt
+      // cache every such turn.
+      await api._trigger("before_agent_start", { systemPrompt: "Base 2." });
+      const secondCtx = await api._trigger("context", { messages: [] });
+      const secondContent = String(secondCtx?.messages?.[0]?.content ?? "");
+
+      // The always-on routing anchor (Pi-1) is out of scope for #1007 and
+      // still fires every turn — but the active_memory block built from the
+      // SAME already-shown event must NOT be rebuilt/re-injected again.
+      expect(secondContent).not.toContain("https://example.com/ticket-1");
+    });
+
+    it("injects only the NEW qualifying event(s) when the active-memory set grows between turns", async () => {
+      await registerPiExtension(api);
+      const sessionFile = `dedupe-1007-grow-${Date.now()}-${Math.random()}`;
+      await api._trigger(
+        "session_start",
+        {},
+        { sessionManager: { getSessionFile: () => sessionFile } },
+      );
+
+      const { db, sessionId } = await openExtensionDb(sessionFile, tempDir);
+      db.insertEvent(sessionId, {
+        type: "external_ref",
+        category: "external-ref",
+        data: "https://example.com/ticket-1",
+        priority: 3,
+      });
+
+      await api._trigger("before_agent_start", { systemPrompt: "Base." });
+      const firstCtx = await api._trigger("context", { messages: [] });
+      expect(String(firstCtx?.messages?.[0]?.content ?? "")).toContain(
+        "https://example.com/ticket-1",
+      );
+
+      // A genuinely NEW priority>=3 event lands before the next turn.
+      db.insertEvent(sessionId, {
+        type: "external_ref",
+        category: "external-ref",
+        data: "https://example.com/ticket-2",
+        priority: 3,
+      });
+
+      await api._trigger("before_agent_start", { systemPrompt: "Base 2." });
+      const secondCtx = await api._trigger("context", { messages: [] });
+      const secondContent = String(secondCtx?.messages?.[0]?.content ?? "");
+
+      // The new event is present...
+      expect(secondContent).toContain("https://example.com/ticket-2");
+      // ...but the already-shown event is NOT repeated.
+      expect(secondContent).not.toContain("https://example.com/ticket-1");
+    });
+  });
+
   // ═══════════════════════════════════════════════════════════
   // Slice 9b (#856): role is NOT re-injected as a standing
   // behavioral_directive every turn (do-nothing-loop fix)