feat(dreamer): gate the memory-pool tasks on session message activity

packages/plugin/src/features/magic-context/dreamer/message-activity.test.ts

@@ -0,0 +1,137 @@
+/// <reference types="bun-types" />
+
+import { afterEach, describe, expect, test } from "bun:test";
+import { Database } from "../../../shared/sqlite";
+import { closeQuietly } from "../../../shared/sqlite-helpers";
+import { runMigrations } from "../migrations";
+import { initializeDatabase } from "../storage-db";
+import { createMessageActivityProvider } from "./message-activity";
+
+const PROJECT_IDENTITY = "/repo/project";
+
+const dbs: Database[] = [];
+
+afterEach(() => {
+    for (const db of dbs.splice(0)) closeQuietly(db);
+});
+
+function track(db: Database): Database {
+    dbs.push(db);
+    return db;
+}
+
+function freshContextDb(): Database {
+    const db = track(new Database(":memory:"));
+    initializeDatabase(db);
+    runMigrations(db);
+    return db;
+}
+
+function freshOpenCodeDb(): Database {
+    const db = track(new Database(":memory:"));
+    db.exec(`
+        CREATE TABLE message (
+            id TEXT PRIMARY KEY,
+            session_id TEXT NOT NULL,
+            time_created INTEGER NOT NULL,
+            time_updated INTEGER NOT NULL,
+            data TEXT
+        );
+    `);
+    return db;
+}
+
+function registerSession(
+    contextDb: Database,
+    sessionId: string,
+    updatedAt: number,
+    isSubagent = 0,
+): void {
+    contextDb
+        .prepare(
+            "INSERT INTO session_projects (session_id, harness, project_path, updated_at) VALUES (?, ?, ?, ?)",
+        )
+        .run(sessionId, "opencode", PROJECT_IDENTITY, updatedAt);
+    contextDb
+        .prepare("INSERT INTO session_meta (session_id, is_subagent) VALUES (?, ?)")
+        .run(sessionId, isSubagent);
+}
+
+function addMessage(openDb: Database, sessionId: string, ts: number): void {
+    openDb
+        .prepare(
+            "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
+        )
+        .run(`${sessionId}-${ts}`, sessionId, ts, ts, "{}");
+}
+
+describe("MessageActivityProvider", () => {
+    test("sinceMs=null counts all ROOT sessions, subagents excluded", () => {
+        const contextDb = freshContextDb();
+        registerSession(contextDb, "root1", 100);
+        registerSession(contextDb, "root2", 200);
+        registerSession(contextDb, "sub1", 300, 1);
+        const provider = createMessageActivityProvider({
+            contextDb,
+            openOpenCodeDb: () => freshOpenCodeDb(),
+        });
+
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, null)).toBe(2);
+        provider.dispose();
+    });
+
+    test("counts root sessions with ≥1 message newer than sinceMs; subagent activity does not count", () => {
+        const contextDb = freshContextDb();
+        const openDb = freshOpenCodeDb();
+        registerSession(contextDb, "root1", 100);
+        registerSession(contextDb, "root2", 200);
+        registerSession(contextDb, "root3", 300);
+        registerSession(contextDb, "sub1", 400, 1);
+        addMessage(openDb, "root1", 100); // stale (≤ sinceMs)
+        addMessage(openDb, "root2", 200); // fresh
+        addMessage(openDb, "root3", 300); // fresh
+        addMessage(openDb, "sub1", 999); // newest, but a subagent → ignored
+        const provider = createMessageActivityProvider({ contextDb, openOpenCodeDb: () => openDb });
+
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, 150)).toBe(2);
+        provider.dispose();
+    });
+
+    test("returns null when opencode.db is unavailable", () => {
+        const contextDb = freshContextDb();
+        registerSession(contextDb, "root1", 100);
+        const provider = createMessageActivityProvider({
+            contextDb,
+            openOpenCodeDb: () => null,
+        });
+
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, null)).toBeNull();
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, 150)).toBeNull();
+        provider.dispose();
+    });
+
+    test("no project sessions → 0 when opencode.db is present", () => {
+        const contextDb = freshContextDb();
+        const openDb = freshOpenCodeDb();
+        // Messages exist but no session_projects row binds them to the project.
+        addMessage(openDb, "orphan", 100);
+        const provider = createMessageActivityProvider({ contextDb, openOpenCodeDb: () => openDb });
+
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, null)).toBe(0);
+        expect(provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, 50)).toBe(0);
+        provider.dispose();
+    });
+
+    test("dispose() closes the open handle without throwing", () => {
+        const contextDb = freshContextDb();
+        const openDb = freshOpenCodeDb();
+        registerSession(contextDb, "root1", 100);
+        addMessage(openDb, "root1", 100);
+        const provider = createMessageActivityProvider({ contextDb, openOpenCodeDb: () => openDb });
+
+        provider.countRootSessionsWithMessagesSince(PROJECT_IDENTITY, 0);
+        expect(() => provider.dispose()).not.toThrow();
+        // Idempotent: a second dispose is a no-op, not an error.
+        expect(() => provider.dispose()).not.toThrow();
+    });
+});

packages/plugin/src/features/magic-context/dreamer/message-activity.ts

@@ -0,0 +1,62 @@
+import type { Database } from "../../../shared/sqlite";
+import { closeQuietly } from "../../../shared/sqlite-helpers";
+import {
+    readOpenCodeOldestMessageTimesSince,
+    selectProjectSessions,
+} from "./retrospective-raw-provider";
+
+/**
+ * The "did any session actually change" signal the dreamer gates on. Scopes a
+ * project's ROOT sessions via context.db (session_projects ⋈ session_meta) and
+ * counts message activity from the authoritative opencode.db — never from a
+ * denormalized copy. session_projects.updated_at records when a session was
+ * first bound to a project, not its last activity, so it is not used here.
+ */
+export interface MessageActivityProvider {
+    /**
+     * Root sessions of the project with at least one message newer than `sinceMs`
+     * (null → all root sessions, for the never-run case). Returns null when
+     * opencode.db is unavailable — callers must fall back to conservative
+     * behavior ("unknown" is not "no work").
+     */
+    countRootSessionsWithMessagesSince(
+        projectIdentity: string,
+        sinceMs: number | null,
+    ): number | null;
+}
+
+export function createMessageActivityProvider(deps: {
+    contextDb: Database;
+    openOpenCodeDb: () => Database | null;
+}): MessageActivityProvider & { dispose(): void } {
+    let sharedDb: Database | null | undefined;
+    let sharedDbOpened = false;
+    // The declared sharedDb type includes `undefined` (the closed state), so the
+    // return type is inferred rather than narrowed — the caller's `if (!db)`
+    // guard treats both missing states identically.
+    const resolveDb = () => {
+        if (!sharedDbOpened) {
+            sharedDbOpened = true;
+            sharedDb = deps.openOpenCodeDb();
+        }
+        return sharedDb;
+    };
+    return {
+        countRootSessionsWithMessagesSince(projectIdentity, sinceMs) {
+            const db = resolveDb();
+            if (!db) return null;
+            const sessions = selectProjectSessions(deps.contextDb, projectIdentity);
+            if (sinceMs === null) return sessions.length;
+            return readOpenCodeOldestMessageTimesSince(
+                db,
+                sessions.map((s) => s.session_id),
+                sinceMs,
+            ).size;
+        },
+        dispose() {
+            if (sharedDb) closeQuietly(sharedDb);
+            sharedDb = undefined;
+            sharedDbOpened = false;
+        },
+    };
+}

packages/plugin/src/features/magic-context/dreamer/retrospective-gate.test.ts

@@ -105,14 +105,15 @@ class ScriptedProvider implements RetrospectiveRawProvider {
     constructor(
         private readonly sessions: string[],
         private readonly rowsBySession: Map<string, RetrospectiveRawMessage[]>,
+        /** Decouples updated_at (registration time) from the row timestamps. */
+        private readonly updatedAtOverride?: Map<string, number>,
     ) {}
     listProjectSessions(): RetrospectiveProjectSession[] {
         return this.sessions.map((sessionId) => ({
             sessionId,
-            updatedAt: Math.max(
-                0,
-                ...(this.rowsBySession.get(sessionId) ?? []).map((row) => row.ts),
-            ),
+            updatedAt:
+                this.updatedAtOverride?.get(sessionId) ??
+                Math.max(0, ...(this.rowsBySession.get(sessionId) ?? []).map((row) => row.ts)),
         }));
     }
     readUserMessagesSince(
@@ -184,6 +185,60 @@ describe("readRetrospectiveScanWindow", () => {
         expect(win.maxScannedTs).toBe(200);
     });
 
+    test("eligibility: a session registered BEFORE the watermark is still scanned (under-scan fix)", async () => {
+        // updated_at is REGISTRATION time (100), not activity. The old
+        // updatedAt > watermark filter excluded this session forever despite its
+        // real messages past the watermark; eligibility comes from the frontier.
+        const rows = new Map([
+            ["s1", [u("s1", 100, "old1"), u("s1", 250, "new1"), u("s1", 300, "new2")]],
+        ]);
+        const updatedAtOverride = new Map([["s1", 100]]);
+        const provider = new ScriptedProvider(["s1"], rows, updatedAtOverride);
+
+        const win = await readRetrospectiveScanWindow(provider, "proj", 250, 0);
+        expect(win.messages.map((m) => m.text)).toEqual(["new2"]);
+        expect(win.maxScannedTs).toBe(300);
+    });
+
+    test("eligibility: a backfilled session registered AFTER the watermark is NOT scanned (over-scan fix)", async () => {
+        // updated_at (300) is backfill/registration time, not activity: s2's real
+        // messages are all ≤ the watermark, so it must not re-enter the scan each
+        // run. s1 (registered long ago but with a new message) makes the spurious
+        // eligibility observable: the window must contain ONLY s1's new message.
+        const rows = new Map([
+            ["s1", [u("s1", 100, "old1"), u("s1", 200, "old2"), u("s1", 300, "new1")]],
+            ["s2", [u("s2", 100, "x"), u("s2", 150, "y"), u("s2", 200, "z")]],
+        ]);
+        const updatedAtOverride = new Map([
+            ["s1", 100],
+            ["s2", 300],
+        ]);
+        const provider = new ScriptedProvider(["s1", "s2"], rows, updatedAtOverride);
+
+        const win = await readRetrospectiveScanWindow(provider, "proj", 250, 0);
+        expect(win.messages.map((m) => m.text)).toEqual(["new1"]);
+        expect(win.messages.every((m) => m.sessionId === "s1")).toBe(true);
+    });
+
+    test("fallback: a provider without an indexed frontier keeps updatedAt-based eligibility", async () => {
+        // Non-indexed providers lack readOldestMessageTimesSince; the updated_at
+        // filter is their only eligibility signal and must still exclude stale
+        // sessions (registration time ≤ watermark).
+        const provider: RetrospectiveRawProvider = {
+            listProjectSessions: () => [
+                { sessionId: "active", updatedAt: 500 },
+                { sessionId: "stale", updatedAt: 100 },
+            ],
+            readUserMessagesSince: (sessionId) => ({
+                messages: sessionId === "active" ? [u("active", 400, "fresh")] : [],
+                truncated: false,
+            }),
+            readUserMessagesBefore: () => [],
+        };
+        const win = await readRetrospectiveScanWindow(provider, "proj", 250, 0);
+        expect(win.messages.map((m) => m.text)).toEqual(["fresh"]);
+    });
+
     test("backlog: keeps the OLDEST since-rows and never advances the watermark past a dropped row (global cap)", async () => {
         // 6 new post-watermark rows, global cap 3. Must keep the oldest 3 and
         // stop the watermark BELOW the first dropped row, so the dropped newer

packages/plugin/src/features/magic-context/dreamer/retrospective-raw-provider.ts

@@ -73,11 +73,39 @@ interface OpenCodeRetrospectiveRawProviderDeps {
     opencodeDb?: Database;
 }
 
-interface SessionProjectRow {
+export interface SessionProjectRow {
     session_id: string;
     updated_at?: number | null;
 }
 
+/**
+ * ROOT sessions of the project, oldest-first. Shared by the retrospective
+ * provider and the message-activity gate so both scope sessions with one
+ * canonical SQL. The retrospective learns from USER friction, but a subagent
+ * child (oracle / mason / historian / dreamer) has no user — its "user
+ * messages" are agent-authored task prompts whose audit/spec wording ("fail",
+ * "error", "wrong", "no padding") trips the frustration regex and whose tool
+ * fan-out trips repeated-tool-call. In a delegation-heavy period children also
+ * outnumber roots ~30:1, so a bounded session scan can be entirely consumed by
+ * them and the real user session is never scanned. is_subagent lives in
+ * session_meta (same DB); missing meta → treat as root.
+ */
+export function selectProjectSessions(
+    contextDb: Database,
+    projectIdentity: string,
+): SessionProjectRow[] {
+    return contextDb
+        .prepare<[string], SessionProjectRow>(
+            `SELECT sp.session_id, sp.updated_at
+               FROM session_projects sp
+               LEFT JOIN session_meta m ON m.session_id = sp.session_id
+              WHERE sp.project_path = ? AND sp.harness = 'opencode'
+                AND COALESCE(m.is_subagent, 0) = 0
+              ORDER BY sp.updated_at ASC, sp.session_id ASC`,
+        )
+        .all(projectIdentity);
+}
+
 interface OpenCodeMessageRow {
     id: string;
     data: string;
@@ -103,25 +131,7 @@ export class OpenCodeRetrospectiveRawProvider implements RetrospectiveRawProvide
     }
 
     listProjectSessions(projectIdentity: string): RetrospectiveProjectSession[] {
-        // ROOT sessions only. The retrospective learns from USER friction, but a
-        // subagent child (oracle / mason / historian / dreamer) has no user — its
-        // "user messages" are agent-authored task prompts whose audit/spec wording
-        // ("fail", "error", "wrong", "no padding") trips the frustration regex and
-        // whose tool fan-out trips repeated-tool-call. In a delegation-heavy period
-        // children also outnumber roots ~30:1, so a bounded session scan can be
-        // entirely consumed by them and the real user session is never scanned.
-        // is_subagent lives in session_meta (same DB); missing meta → treat as root.
-        const rows = this.deps.contextDb
-            .prepare<[string], SessionProjectRow>(
-                `SELECT sp.session_id, sp.updated_at
-                   FROM session_projects sp
-                   LEFT JOIN session_meta m ON m.session_id = sp.session_id
-                  WHERE sp.project_path = ? AND sp.harness = 'opencode'
-                    AND COALESCE(m.is_subagent, 0) = 0
-                  ORDER BY sp.updated_at ASC, sp.session_id ASC`,
-            )
-            .all(projectIdentity);
-        return rows.map((row) => ({
+        return selectProjectSessions(this.deps.contextDb, projectIdentity).map((row) => ({
             sessionId: row.session_id,
             updatedAt: typeof row.updated_at === "number" ? row.updated_at : undefined,
         }));
@@ -220,15 +230,23 @@ export async function readRetrospectiveScanWindow(
     );
     try {
         const allSessions = await provider.listProjectSessions(projectIdentity);
-        const eligibleSessions = allSessions
-            .map((session, index) => ({ session, index }))
-            .filter(({ session }) => (session.updatedAt ?? Number.POSITIVE_INFINITY) > watermarkMs);
+        // Eligibility is message activity, not the registration-time updated_at column.
+        // readOldestMessageTimesSince computes exactly "has a message newer than the
+        // watermark" over the message table; only providers without an indexed store
+        // fall back to the updated_at filter.
         const oldestBySession = provider.readOldestMessageTimesSince
             ? await provider.readOldestMessageTimesSince(
-                  eligibleSessions.map(({ session }) => session.sessionId),
+                  allSessions.map((session) => session.sessionId),
                   watermarkMs,
               )
             : null;
+        const eligibleSessions = (
+            oldestBySession
+                ? allSessions.filter((session) => oldestBySession.has(session.sessionId))
+                : allSessions.filter(
+                      (session) => (session.updatedAt ?? Number.POSITIVE_INFINITY) > watermarkMs,
+                  )
+        ).map((session, index) => ({ session, index }));
         const sessions = (
             oldestBySession
                 ? eligibleSessions.filter(({ session }) => oldestBySession.has(session.sessionId))
@@ -381,7 +399,7 @@ function readOpenCodeMessagesSince(
     return { messages: normalizeOpenCodeRows(db, sessionId, kept), truncated };
 }
 
-function readOpenCodeOldestMessageTimesSince(
+export function readOpenCodeOldestMessageTimesSince(
     db: Database,
     sessionIds: readonly string[],
     sinceMs: number,