mason: wire git commit backlog drain into dream-timer sweep

After the indexed git sweep releases its lease, invoke
drainCommitBacklogForProject (ignoreCooldown + releaseGitSweepLease) so
pre-existing unembedded commits embed when git log finds no new commits.

Author: ualtinok

packages/plugin/src/features/magic-context/project-embedding-registry.test.ts

@@ -10,12 +10,17 @@ import {
     replaceCompartmentChunkEmbeddings,
 } from "./compartment-chunk-embedding";
 import { appendCompartments, getCompartments } from "./compartment-storage";
+import type { GitCommit } from "./git-commits/git-log-reader";
+import { countEmbeddedCommits } from "./git-commits/storage-git-commit-embeddings";
+import { upsertCommits } from "./git-commits/storage-git-commits";
+import { acquireGitSweepLease, releaseGitSweepLease } from "./git-commits/sweep-coordinator";
 import type { EmbeddingProvider, EmbeddingPurpose } from "./memory/embedding-provider";
 import { insertMemory } from "./memory/storage-memory";
 import { getStoredModelId, saveEmbedding } from "./memory/storage-memory-embeddings";
 import {
     _resetProjectEmbeddingRegistryForTests,
     _setTestProviderFactoryForProject,
+    drainCommitBacklogForProject,
     embedSessionCompartmentChunks,
     embedTextForProject,
     embedUnembeddedCompartmentChunksForProject,
@@ -72,6 +77,17 @@ function localConfig(model: string, maxInputTokens?: number): EmbeddingConfig {
     };
 }
 
+function makeGitCommit(shaSeed: string, committedAtMs: number): GitCommit {
+    const sha = shaSeed.padEnd(40, shaSeed);
+    return {
+        sha,
+        shortSha: sha.slice(0, 7),
+        message: `commit ${shaSeed}`,
+        author: "dev@example.com",
+        committedAtMs,
+    };
+}
+
 function seedCompartmentWithFts(
     db: NonNullable<ReturnType<typeof openDatabase>>,
     sessionId: string,
@@ -151,6 +167,92 @@ describe("project embedding registry", () => {
         tempDirs.length = 0;
     });
 
+    it("drainCommitBacklogForProject embeds pre-indexed commits with no new git log work", async () => {
+        _setTestProviderFactoryForProject(
+            (config) =>
+                new FakeEmbeddingProvider(config.provider === "local" ? config.model : "off"),
+        );
+        const db = useTempDb();
+        const projectIdentity = "git:commit-backlog";
+        upsertCommits(db, projectIdentity, [
+            makeGitCommit("backlog-a", 1000),
+            makeGitCommit("backlog-b", 2000),
+        ]);
+        registerProjectEmbeddingAndMaybeWipe(
+            db,
+            projectIdentity,
+            localConfig("model-commits"),
+            { memoryEnabled: true, gitCommitEnabled: true },
+            "/tmp/commits",
+        );
+
+        expect(countEmbeddedCommits(db, projectIdentity)).toBe(0);
+        const drained = await drainCommitBacklogForProject(
+            db,
+            projectIdentity,
+            Date.now() + 60_000,
+        );
+        expect(drained).toBe(2);
+        expect(countEmbeddedCommits(db, projectIdentity)).toBe(2);
+    });
+
+    it("drainCommitBacklogForProject skips when git commit indexing is disabled", async () => {
+        _setTestProviderFactoryForProject(
+            (config) =>
+                new FakeEmbeddingProvider(config.provider === "local" ? config.model : "off"),
+        );
+        const db = useTempDb();
+        const projectIdentity = "git:commits-off";
+        upsertCommits(db, projectIdentity, [makeGitCommit("off-a", 1000)]);
+        registerProjectEmbeddingAndMaybeWipe(
+            db,
+            projectIdentity,
+            localConfig("model-commits"),
+            { memoryEnabled: true, gitCommitEnabled: false },
+            "/tmp/commits-off",
+        );
+
+        const drained = await drainCommitBacklogForProject(
+            db,
+            projectIdentity,
+            Date.now() + 60_000,
+        );
+        expect(drained).toBe(0);
+        expect(countEmbeddedCommits(db, projectIdentity)).toBe(0);
+    });
+
+    it("drainCommitBacklogForProject short-circuits when the git sweep lease is held", async () => {
+        _setTestProviderFactoryForProject(
+            (config) =>
+                new FakeEmbeddingProvider(config.provider === "local" ? config.model : "off"),
+        );
+        const db = useTempDb();
+        const projectIdentity = "git:lease-block";
+        upsertCommits(db, projectIdentity, [makeGitCommit("lease-a", 1000)]);
+        registerProjectEmbeddingAndMaybeWipe(
+            db,
+            projectIdentity,
+            localConfig("model-commits"),
+            { memoryEnabled: true, gitCommitEnabled: true },
+            "/tmp/lease",
+        );
+
+        const holder = acquireGitSweepLease(db, projectIdentity, "other-holder");
+        expect(holder.acquired).toBe(true);
+
+        const drained = await drainCommitBacklogForProject(
+            db,
+            projectIdentity,
+            Date.now() + 60_000,
+        );
+        expect(drained).toBe(0);
+        expect(countEmbeddedCommits(db, projectIdentity)).toBe(0);
+
+        if (holder.acquired) {
+            releaseGitSweepLease(db, projectIdentity, holder.holderId);
+        }
+    });
+
     it("keeps independent snapshots and providers for two projects in one process", async () => {
         _setTestProviderFactoryForProject(
             (config) =>

packages/plugin/src/features/magic-context/project-embedding-registry.ts

@@ -635,17 +635,18 @@ async function embedCommitBatch(
 /**
  * Drain a project's unembedded-commit backlog, coordinated across processes.
  *
- * This is the ONLY path that drains pure backlogs (the dream-timer git-sweep
- * only embeds when `git log` finds NEW commits, so a repo indexed before
- * embeddings were enabled never drains there). Every plugin process runs this
+ * Drains pure backlogs (indexed commits with no embedding row). The dream-timer
+ * git-sweep embeds new commits from `git log` but skips backlog drain when
+ * `embedded=0`; this path runs after each sweep (ignoreCooldown lease) so
+ * pre-existing backlogs clear. Every plugin process runs this
  * on its dream-timer tick, so without coordination N processes hammer the
  * embedding provider with the same commits. We take the shared git-sweep lease
  * (mutual exclusion) per identity — but with `ignoreCooldown`, because a
  * backlog must keep draining every tick until empty and must not be blocked by
  * the cooldown the dream-timer sweep advances. We release without marking
  * success so the two paths' cooldown tracking stays independent.
  */
-async function drainCommitBacklogForProject(
+export async function drainCommitBacklogForProject(
     db: Database,
     projectIdentity: string,
     deadline: number,

packages/plugin/src/plugin/dream-timer.test.ts

@@ -71,3 +71,13 @@ describe("dream-timer null-DB guards (static)", () => {
         expect(source).not.toContain("db: Database = openDatabase()");
     });
 });
+
+describe("dream-timer git commit backlog drain (static)", () => {
+    const source = readFileSync(join(import.meta.dir, "dream-timer.ts"), "utf8");
+
+    test("sweepGitCommits invokes coordinated backlog drain after the index sweep", () => {
+        expect(source).toContain("drainCommitBacklogForProject");
+        expect(source).toContain("memorySnapshot?.gitCommitEnabled");
+        expect(source).toContain("backlogDrained");
+    });
+});

packages/plugin/src/plugin/dream-timer.ts

@@ -13,6 +13,7 @@ import {
     embedUnembeddedMemoriesForProject,
     getProjectEmbeddingSnapshot,
 } from "../features/magic-context/memory/embedding";
+import { drainCommitBacklogForProject } from "../features/magic-context/project-embedding-registry";
 import { openDatabase, runSqliteOptimize } from "../features/magic-context/storage";
 import { log } from "../shared/logger";
 import { resolveFallbackChain } from "../shared/resolve-fallbacks";
@@ -21,6 +22,8 @@ import type { PluginContext } from "./types";
 
 /** Check interval for dream schedule (15 minutes). */
 const DREAM_TIMER_INTERVAL_MS = 15 * 60 * 1000;
+/** Wall-clock budget for post-sweep commit backlog drain (matches indexer embed sweep). */
+const GIT_COMMIT_BACKLOG_DRAIN_MAX_MS = 5 * 60 * 1000;
 
 /**
  * Per-project work registered with the timer. The timer is a process-wide
@@ -313,7 +316,7 @@ async function sweepGitCommits(args: {
             sinceDays: gitCommitIndexing.since_days,
             maxCommits: gitCommitIndexing.max_commits,
         });
-        // Drain any remaining embedding backlog (indexer caps per run).
+        // Drain any remaining embedding backlog from this sweep (indexer caps per run).
         let drainedEmbeddings = 0;
         if (result.embedded > 0) {
             drainedEmbeddings = await embedUnembeddedCommits(db, projectIdentity);
@@ -325,9 +328,26 @@ async function sweepGitCommits(args: {
                 `[git-commits] sweep finished for ${projectIdentity}, but lease was no longer active; cooldown not advanced`,
             );
         }
+
+        const memorySnapshot = getProjectEmbeddingSnapshot(projectIdentity);
+        let backlogDrained = 0;
+        if (memorySnapshot?.gitCommitEnabled) {
+            try {
+                backlogDrained = await drainCommitBacklogForProject(
+                    db,
+                    projectIdentity,
+                    Date.now() + GIT_COMMIT_BACKLOG_DRAIN_MAX_MS,
+                );
+            } catch (error) {
+                log(
+                    `[git-commits] commit backlog drain failed for ${projectIdentity}: ${error instanceof Error ? error.message : String(error)}`,
+                );
+            }
+        }
+
         const elapsedMs = Date.now() - startedAt;
         log(
-            `[git-commits] sweep finished for ${projectIdentity} in ${elapsedMs}ms: scanned=${result.scanned} inserted=${result.inserted} updated=${result.updated} evicted=${result.evicted} embedded=${result.embedded} drained=${drainedEmbeddings}`,
+            `[git-commits] sweep finished for ${projectIdentity} in ${elapsedMs}ms: scanned=${result.scanned} inserted=${result.inserted} updated=${result.updated} evicted=${result.evicted} embedded=${result.embedded} drained=${drainedEmbeddings} backlogDrained=${backlogDrained}`,
         );
     } catch (error) {
         releaseGitSweepLease(db, projectIdentity, holderId);