fix: index --full FK crash (#16) + cross-process contextfit ingest lock (#17)

src/adapters/change-feed/obsidian-fs/watcher.ts

@@ -146,6 +146,9 @@ export class VaultWatcher {
       const r = await indexVaultWithContextFit(this.opts.vault.config, {});
       if (r.status === "completed") {
         this.opts.log(`ContextFit KB refreshed (${r.durationMs}ms)`);
+      } else if (r.status === "skipped") {
+        // Issue #17: another process is mid-ingest; it will do a trailing pass.
+        this.opts.log(`ContextFit KB refresh skipped (another ingest in progress; flagged)`);
       } else {
         this.opts.log(`ContextFit KB refresh failed: ${r.error}`);
       }

src/adapters/retrieval/contextfit/index.ts

@@ -25,6 +25,13 @@ import {
   type ContextFitCliConfig,
   type ContextFitChunk,
 } from "./cli.js";
+import {
+  tryAcquireIngestLock,
+  releaseIngestLock,
+  markIngestDirty,
+  isIngestDirty,
+  clearIngestDirty,
+} from "./ingest-lock.js";
 
 const DEFAULT_COMMAND = "contextfit";
 
@@ -44,7 +51,13 @@ export function cliConfigForVault(vault: VaultConfig): ContextFitCliConfig {
 }
 
 export interface ContextFitIndexResult {
-  status: "completed" | "failed";
+  /**
+   * "skipped" (Issue #17): another ingest for this vault was already in flight,
+   * so this call marked the vault dirty and returned WITHOUT ingesting. The
+   * in-flight holder does a trailing re-ingest, so the change is not lost.
+   * Callers treat "skipped" as success (no error), not failure.
+   */
+  status: "completed" | "failed" | "skipped";
   /** Human-readable stats line from ContextFit's ingest output. */
   stats: string;
   durationMs: number;
@@ -59,14 +72,35 @@ export interface ContextFitIndexResult {
  */
 export async function indexVaultWithContextFit(
   vault: VaultConfig,
-  opts: { onProgress?: (msg: string) => void } = {},
+  opts: {
+    onProgress?: (msg: string) => void;
+    /** Test-only: `~/.vault-memory` root override for the ingest lock. */
+    lockRootOverride?: string;
+    /**
+     * Test-only dependency injection. Production omits these and the real
+     * probe/ingest (which spawn the `contextfit` CLI) are used. Tests pass
+     * fakes to exercise the lock/dirty/trailing-pass orchestration without the
+     * binary.
+     */
+    _deps?: {
+      probe?: (cfg: ContextFitCliConfig) => Promise<boolean>;
+      ingest?: (cfg: ContextFitCliConfig, source: string) => Promise<string>;
+      clearKb?: (kbPath: string) => Promise<void>;
+    };
+  } = {},
 ): Promise<ContextFitIndexResult> {
   const log = opts.onProgress ?? (() => {});
   const cfg = cliConfigForVault(vault);
   const start = Date.now();
+  const lockOpts =
+    opts.lockRootOverride !== undefined ? { rootOverride: opts.lockRootOverride } : {};
+  const probe =
+    opts._deps?.probe ?? ((c: ContextFitCliConfig) => contextFitProbe({ command: c.command }));
+  const ingest = opts._deps?.ingest ?? contextFitIngest;
+  const clearKb = opts._deps?.clearKb ?? ((p: string) => rm(p, { recursive: true, force: true }));
 
   log(`ContextFit: ingesting ${vault.path} → ${cfg.kbPath}`);
-  const available = await contextFitProbe({ command: cfg.command });
+  const available = await probe(cfg);
   if (!available) {
     return {
       status: "failed",
@@ -78,18 +112,44 @@ export async function indexVaultWithContextFit(
     };
   }
 
+  // Issue #17: serialize ingests cross-process. Second-comer marks the vault
+  // dirty and skips (no wait, no wasted double-rebuild); the in-flight holder
+  // does a trailing re-ingest so the latest change is captured.
+  const lock = await tryAcquireIngestLock(vault.name, lockOpts);
+  if (!lock.acquired) {
+    await markIngestDirty(vault.name, lockOpts);
+    log(`ContextFit: re-ingest already in progress (pid ${lock.ownerPid}); flagged for retry`);
+    return { status: "skipped", stats: "", durationMs: Date.now() - start };
+  }
+
   try {
-    // ContextFit refuses to ingest into an existing KB (it finds the manifest
-    // and exits non-zero, demanding --resume or a clean dir). Our index
-    // semantics are always a FULL rebuild, so clear the KB dir first — this
-    // makes re-index / live-reindex / write-refresh / catchup idempotent.
-    await rm(cfg.kbPath, { recursive: true, force: true });
-    const stats = await contextFitIngest(cfg, vault.path);
+    // Loop so a change that lands DURING our ingest triggers exactly one more
+    // pass. Bounded to avoid an unbounded churn loop under constant writes; the
+    // watcher's debounce already coalesces bursts, so 1 trailing pass suffices
+    // in practice and MAX_PASSES is a safety backstop.
+    const MAX_PASSES = 8;
+    let stats = "";
+    let passes = 0;
+    do {
+      // Clear the flag BEFORE ingesting: any write that arrives after this
+      // point re-sets it and earns another pass; writes before it are already
+      // captured by the rebuild we are about to do.
+      await clearIngestDirty(vault.name, lockOpts);
+      // ContextFit refuses to ingest into an existing KB (it finds the manifest
+      // and exits non-zero, demanding --resume or a clean dir). Our index
+      // semantics are always a FULL rebuild, so clear the KB dir first — this
+      // makes re-index / live-reindex / write-refresh / catchup idempotent.
+      await clearKb(cfg.kbPath);
+      stats = await ingest(cfg, vault.path);
+      passes += 1;
+    } while (passes < MAX_PASSES && (await isIngestDirty(vault.name, lockOpts)));
     log(stats.trim().split("\n").slice(-3).join(" · "));
     return { status: "completed", stats, durationMs: Date.now() - start };
   } catch (err) {
     const message = err instanceof Error ? err.message : String(err);
     return { status: "failed", stats: "", durationMs: Date.now() - start, error: message };
+  } finally {
+    await releaseIngestLock(vault.name, lockOpts);
   }
 }
 

src/adapters/retrieval/contextfit/ingest-lock.test.ts

@@ -0,0 +1,207 @@
+/**
+ * Ingest-lock tests (Issue #17). Exercise the cross-process mutex + dirty-flag
+ * primitive directly against a temp `~/.vault-memory` root (rootOverride), so
+ * they run without the contextfit binary. Concurrent-ingest serialization at
+ * the `indexVaultWithContextFit` level is covered by the live index test
+ * (gated on contextfit being installed) in ./index.test.ts.
+ */
+
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { promises as fs } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import {
+  tryAcquireIngestLock,
+  releaseIngestLock,
+  markIngestDirty,
+  isIngestDirty,
+  clearIngestDirty,
+} from "./ingest-lock.js";
+import { indexVaultWithContextFit } from "./index.js";
+import type { VaultConfig } from "../../../types.js";
+
+describe("ContextFit ingest lock (Issue #17)", () => {
+  let root = "";
+  const V = "vaultA";
+
+  beforeEach(async () => {
+    root = await fs.mkdtemp(join(tmpdir(), "vm-ingest-lock-"));
+  });
+  afterEach(async () => {
+    await fs.rm(root, { recursive: true, force: true });
+  });
+
+  it("acquires an uncontended lock and writes our pid", async () => {
+    const r = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(r.acquired).toBe(true);
+    if (r.acquired) {
+      const pid = parseInt(await fs.readFile(r.path, "utf8"), 10);
+      expect(pid).toBe(process.pid);
+    }
+  });
+
+  it("second acquire is contended while the first is held", async () => {
+    const first = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(first.acquired).toBe(true);
+    const second = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(second.acquired).toBe(false);
+    if (!second.acquired) expect(second.ownerPid).toBe(process.pid);
+  });
+
+  it("re-acquires after release", async () => {
+    await tryAcquireIngestLock(V, { rootOverride: root });
+    await releaseIngestLock(V, { rootOverride: root });
+    const again = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(again.acquired).toBe(true);
+  });
+
+  it("steals a lock whose recorded pid is dead", async () => {
+    // Hand-write a lock file owned by a pid that cannot be alive. POSIX pids
+    // are positive; a huge value is guaranteed dead (ESRCH), so it is stolen.
+    const dir = join(root, "locks");
+    await fs.mkdir(dir, { recursive: true });
+    await fs.writeFile(join(dir, `${V}.ingest.lock`), "2147483646\n");
+    const r = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(r.acquired).toBe(true);
+  });
+
+  it("steals a lock with malformed (non-numeric) contents", async () => {
+    const dir = join(root, "locks");
+    await fs.mkdir(dir, { recursive: true });
+    await fs.writeFile(join(dir, `${V}.ingest.lock`), "not-a-pid\n");
+    const r = await tryAcquireIngestLock(V, { rootOverride: root });
+    expect(r.acquired).toBe(true);
+  });
+
+  it("release is safe when we do not hold the lock", async () => {
+    await expect(releaseIngestLock(V, { rootOverride: root })).resolves.toBeUndefined();
+  });
+
+  it("dirty flag round-trips: mark → is → clear", async () => {
+    expect(await isIngestDirty(V, { rootOverride: root })).toBe(false);
+    await markIngestDirty(V, { rootOverride: root });
+    expect(await isIngestDirty(V, { rootOverride: root })).toBe(true);
+    await clearIngestDirty(V, { rootOverride: root });
+    expect(await isIngestDirty(V, { rootOverride: root })).toBe(false);
+  });
+
+  it("clear is idempotent when no flag exists", async () => {
+    await expect(clearIngestDirty(V, { rootOverride: root })).resolves.toBeUndefined();
+    expect(await isIngestDirty(V, { rootOverride: root })).toBe(false);
+  });
+
+  it("lock and dirty flag are independent per vault", async () => {
+    await tryAcquireIngestLock("vaultA", { rootOverride: root });
+    await markIngestDirty("vaultA", { rootOverride: root });
+    // A different vault is unaffected.
+    const b = await tryAcquireIngestLock("vaultB", { rootOverride: root });
+    expect(b.acquired).toBe(true);
+    expect(await isIngestDirty("vaultB", { rootOverride: root })).toBe(false);
+  });
+});
+
+describe("indexVaultWithContextFit lock orchestration (Issue #17)", () => {
+  let root = "";
+  const vault: VaultConfig = { name: "orch", path: "/tmp/orch-vault", backend: "contextfit" };
+
+  beforeEach(async () => {
+    root = await fs.mkdtemp(join(tmpdir(), "vm-ingest-orch-"));
+  });
+  afterEach(async () => {
+    await fs.rm(root, { recursive: true, force: true });
+  });
+
+  const okDeps = (ingest: (c: unknown, s: string) => Promise<string>) => ({
+    probe: async () => true,
+    ingest: ingest as never,
+    clearKb: async () => {},
+  });
+
+  it("skips (and flags dirty) when the lock is already held by another process", async () => {
+    // Simulate an in-flight ingest in another process: hand-hold the lock with
+    // a live pid (our own — isProcessAlive(process.pid) is true, so not stolen).
+    await tryAcquireIngestLock(vault.name, { rootOverride: root });
+
+    let ingestCalls = 0;
+    const r = await indexVaultWithContextFit(vault, {
+      lockRootOverride: root,
+      _deps: okDeps(async () => {
+        ingestCalls += 1;
+        return "stats";
+      }),
+    });
+
+    expect(r.status).toBe("skipped");
+    expect(ingestCalls).toBe(0); // did not ingest
+    expect(await isIngestDirty(vault.name, { rootOverride: root })).toBe(true); // flagged
+  });
+
+  it("ingests once, clears the dirty flag, and releases the lock", async () => {
+    let ingestCalls = 0;
+    const r = await indexVaultWithContextFit(vault, {
+      lockRootOverride: root,
+      _deps: okDeps(async () => {
+        ingestCalls += 1;
+        return "stats";
+      }),
+    });
+
+    expect(r.status).toBe("completed");
+    expect(ingestCalls).toBe(1);
+    expect(await isIngestDirty(vault.name, { rootOverride: root })).toBe(false);
+    // Lock released → a subsequent acquire succeeds.
+    const again = await tryAcquireIngestLock(vault.name, { rootOverride: root });
+    expect(again.acquired).toBe(true);
+  });
+
+  it("does a trailing re-ingest when a change lands DURING the ingest", async () => {
+    // First ingest sets the flag mid-run (simulating a write arriving during
+    // the rebuild); the holder must loop exactly once more, then stop.
+    let ingestCalls = 0;
+    const r = await indexVaultWithContextFit(vault, {
+      lockRootOverride: root,
+      _deps: okDeps(async () => {
+        ingestCalls += 1;
+        if (ingestCalls === 1) {
+          // A concurrent write arrives while pass 1 is running.
+          await markIngestDirty(vault.name, { rootOverride: root });
+        }
+        return "stats";
+      }),
+    });
+
+    expect(r.status).toBe("completed");
+    expect(ingestCalls).toBe(2); // one trailing pass captured the mid-run change
+    expect(await isIngestDirty(vault.name, { rootOverride: root })).toBe(false);
+  });
+
+  it("releases the lock even when the ingest throws", async () => {
+    const r = await indexVaultWithContextFit(vault, {
+      lockRootOverride: root,
+      _deps: okDeps(async () => {
+        throw new Error("boom");
+      }),
+    });
+    expect(r.status).toBe("failed");
+    expect(r.error).toMatch(/boom/);
+    // finally released the lock despite the throw.
+    const again = await tryAcquireIngestLock(vault.name, { rootOverride: root });
+    expect(again.acquired).toBe(true);
+  });
+
+  it("bounds trailing passes so constant writes cannot loop forever", async () => {
+    // The flag is re-set on every pass → the loop must stop at MAX_PASSES (8),
+    // not spin indefinitely.
+    let ingestCalls = 0;
+    const r = await indexVaultWithContextFit(vault, {
+      lockRootOverride: root,
+      _deps: okDeps(async () => {
+        ingestCalls += 1;
+        await markIngestDirty(vault.name, { rootOverride: root });
+        return "stats";
+      }),
+    });
+    expect(r.status).toBe("completed");
+    expect(ingestCalls).toBe(8); // MAX_PASSES backstop
+  });
+});