fix: make recall resilient to transient database contention

dsh.ts

@@ -369,6 +369,13 @@ export function apply(ctx: DshContext, input: Config = {}): void {
         assembly.contexts.push({ name: "graph-memory:recall", text });
       }
     } catch (error) {
+      // A failed recall must not stay pinned in recallCache: the next
+      // system-prompt/assemble would re-await the same rejected Promise and
+      // log "[graph-memory] DSH recall failed" on every round until the next
+      // agent/inbox/claimed clears it. Evict the entry so the next assembly
+      // performs a fresh recall attempt.
+      const current = recallCache.get(key);
+      if (current && current.query === query) recallCache.delete(key);
       ctx.logger.warn(`[graph-memory] DSH recall failed: ${String(error)}`);
     }
     return next();

src/store/db.ts

@@ -46,6 +46,12 @@ export function openDb(dbPath: string): DatabaseSyncInstance {
   const db = new DatabaseSync(resolved);
   db.exec("PRAGMA journal_mode = WAL");
   db.exec("PRAGMA foreign_keys = ON");
+  // Multiple plugin instances (DSH fibers/profiles, tests) may open the same
+  // database file. Without a busy timeout, a write that collides with another
+  // connection's transaction fails immediately with SQLITE_BUSY ("database is
+  // locked") and the error surfaces as a spurious recall/extraction failure.
+  // Wait up to 5s for the lock instead so transient contention is retried.
+  db.exec("PRAGMA busy_timeout = 5000");
   migrate(db);
   return db;
 }

test/db.test.ts

@@ -0,0 +1,88 @@
+/**
+ * graph-memory — SQLite 连接回归测试
+ *
+ * By: adoresever
+ * Email: Wywelljob@gmail.com
+ */
+
+import { describe, expect, it } from "vitest";
+import { spawn } from "node:child_process";
+import { mkdtempSync, writeFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { openDb } from "../src/store/db.ts";
+
+describe("openDb", () => {
+  it("configures a busy timeout on every connection", () => {
+    const dir = mkdtempSync(join(tmpdir(), "gm-db-"));
+    const db = openDb(join(dir, "graph-memory.db"));
+    try {
+      const row = db.prepare("PRAGMA busy_timeout").get() as { timeout: number };
+      expect(Number(row.timeout)).toBe(5000);
+    } finally {
+      db.close();
+    }
+  });
+
+  it("waits for a write lock held by another process instead of throwing database is locked", async () => {
+    const dir = mkdtempSync(join(tmpdir(), "gm-db-"));
+    const dbPath = join(dir, "graph-memory.db");
+    const db = openDb(dbPath);
+
+    // Child process mirrors a second host connection (e.g. another DSH
+    // process sharing the same database file): it takes the write lock, holds
+    // it briefly, then commits. Its event loop is its own, so it can release
+    // the lock while this process' busy handler is waiting. The script lives
+    // in the project root so require("@photostructure/sqlite") resolves.
+    const childScript = `
+      const { DatabaseSync } = require("@photostructure/sqlite");
+      const file = process.argv[2];
+      const db = new DatabaseSync(file);
+      db.exec("PRAGMA journal_mode = WAL");
+      db.exec("PRAGMA busy_timeout = 5000");
+      db.exec("BEGIN IMMEDIATE");
+      console.log("LOCKED");
+      setTimeout(() => {
+        try { db.exec("COMMIT"); } catch (e) { console.error("child commit failed:", String(e)); process.exit(1); }
+        db.close();
+      }, 500);
+    `;
+    const scriptPath = join(process.cwd(), ".gm-lock-holder.cjs");
+    writeFileSync(scriptPath, childScript);
+
+    try {
+      const child = spawn(process.execPath, [scriptPath, dbPath], { stdio: ["ignore", "pipe", "pipe"] });
+      const locked = new Promise<void>((resolve, reject) => {
+        let settled = false;
+        child.stdout.on("data", (chunk: Buffer) => {
+          if (chunk.toString().includes("LOCKED") && !settled) {
+            settled = true;
+            resolve();
+          }
+        });
+        child.on("error", reject);
+        child.on("exit", (code) => {
+          if (!settled && code !== 0) reject(new Error(`child exited early with code ${code}`));
+        });
+      });
+
+      await locked;
+
+      // This write collides with the child's open transaction. Without a busy
+      // timeout it fails instantly with SQLITE_BUSY; with one it blocks until
+      // the child commits (~500ms) and then succeeds.
+      const t0 = Date.now();
+      db.prepare(`
+        INSERT INTO gm_nodes (id, type, name, description, content, status, validated_count, source_sessions, created_at, updated_at)
+        VALUES (?, 'SKILL', ?, ?, ?, 'active', 1, '[]', ?, ?)
+      `).run("n-busy-timeout", "busy-timeout-node", "desc", "content", Date.now(), Date.now());
+      const elapsed = Date.now() - t0;
+
+      expect(elapsed).toBeGreaterThanOrEqual(300);
+    } finally {
+      db.close();
+      rmSync(scriptPath, { force: true });
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+});

test/dsh-recall-cache.test.ts

@@ -0,0 +1,126 @@
+/**
+ * graph-memory — DSH recallCache 回归测试
+ *
+ * By: adoresever
+ * Email: Wywelljob@gmail.com
+ *
+ * 覆盖 dsh.ts 的 system-prompt/assemble 召回缓存：一次失败的召回必须被
+ * 从 recallCache 驱逐，否则同一 rejected Promise 会被钉在缓存里，之后每轮
+ * assemble 都 re-await 它，把单次瞬时错误变成每轮必现的重复报错。
+ */
+
+import { describe, expect, it, vi } from "vitest";
+
+const { recallMock } = vi.hoisted(() => ({ recallMock: vi.fn() }));
+
+vi.mock("../src/recaller/recall.ts", () => {
+  class MockRecaller {
+    static instances: MockRecaller[] = [];
+    recall = recallMock;
+    setEmbedFn = vi.fn();
+    syncEmbed = vi.fn();
+    constructor(_db: unknown, _cfg: unknown) {
+      MockRecaller.instances.push(this);
+    }
+  }
+  return { Recaller: MockRecaller };
+});
+
+import { apply } from "../dsh.ts";
+
+interface Handler {
+  (...args: any[]): any;
+}
+
+function mockCtx() {
+  const handlers = new Map<string, Handler>();
+  const disposers: Array<() => void | Promise<void>> = [];
+  return {
+    handlers,
+    disposers,
+    ctx: {
+      logger: { info() {}, warn() {}, error() {} },
+      llm: {
+        stream() {
+          throw new Error("llm.stream should not be used in this test");
+        },
+      },
+      tools: { register() { return () => {}; } },
+      credentials: {
+        async resolve() { return undefined; },
+      },
+      on(event: string, listener: Handler) {
+        handlers.set(event, listener);
+        return () => {};
+      },
+      effect(register: () => () => void | Promise<void>) {
+        const disposer = register();
+        disposers.push(disposer);
+        return () => {};
+      },
+    },
+  };
+}
+
+async function dispose(harness: { disposers: Array<() => void | Promise<void>> }): Promise<void> {
+  for (const disposer of harness.disposers) await disposer();
+}
+
+describe("DSH recall cache", () => {
+  beforeEach(() => {
+    recallMock.mockReset();
+  });
+
+  it("evicts a failed recall so the next assembly retries instead of re-awaiting the rejected promise", async () => {
+    recallMock
+      .mockRejectedValueOnce(new Error("transient: database is locked"))
+      .mockResolvedValue({ nodes: [], edges: [] });
+
+    const harness = mockCtx();
+    const { ctx, handlers } = harness;
+    apply(ctx, { dbPath: ":memory:", extractionEnabled: false });
+    try {
+      const claimed = handlers.get("agent/inbox/claimed")!;
+      const assemble = handlers.get("system-prompt/assemble")!;
+
+      claimed({ agent: { id: "agent-1" }, message: { source: { kind: "user" }, content: "remember anything about sqlite?" } });
+
+      const assembly = { contexts: [] as any[] };
+      await assemble(assembly, { agent: { id: "agent-1" } }, async () => {});
+      // First attempt failed (recall rejected).
+      expect(recallMock).toHaveBeenCalledTimes(1);
+
+      await assemble(assembly, { agent: { id: "agent-1" } }, async () => {});
+      // Regression: without eviction the second assembly re-awaits the pinned
+      // rejected Promise (still 1 call) and fails again. With eviction it must
+      // run a fresh recall.
+      expect(recallMock).toHaveBeenCalledTimes(2);
+    } finally {
+      await dispose(harness);
+    }
+  });
+
+  it("keeps caching a successful recall for the same agent and query", async () => {
+    recallMock.mockResolvedValue({ nodes: [], edges: [] });
+
+    const harness = mockCtx();
+    const { ctx, handlers } = harness;
+    apply(ctx, { dbPath: ":memory:", extractionEnabled: false });
+    try {
+      const claimed = handlers.get("agent/inbox/claimed")!;
+      const assemble = handlers.get("system-prompt/assemble")!;
+
+      claimed({ agent: { id: "agent-2" }, message: { source: { kind: "user" }, content: "what do I know about pagerank?" } });
+
+      const assembly = { contexts: [] as any[] };
+      await assemble(assembly, { agent: { id: "agent-2" } }, async () => {});
+      await assemble(assembly, { agent: { id: "agent-2" } }, async () => {});
+      await assemble(assembly, { agent: { id: "agent-2" } }, async () => {});
+
+      // Success must stay cached: repeated assemblies reuse the same recall.
+      expect(recallMock).toHaveBeenCalledTimes(1);
+    } finally {
+      await dispose(harness);
+    }
+  });
+});