From 785bcc39fdc7458244f9f0798eb72c029b858039 Mon Sep 17 00:00:00 2001 From: Jona Date: Fri, 18 Sep 2026 02:39:35 +0200 Subject: [PATCH 1/2] fix(dreamer): gate and scan the retrospective on real message activity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retrospective gate tested `session_projects.updated_at > watermark`, but updated_at is written only at first project binding (and by the backfill at scan time) — it is not an activity timestamp: - under-scan: once the watermark passed a session's registration time, new messages in that session could never re-trigger a run, and a session truncated at the per-session cap never had its tail read; - over-scan: a backfilled old session is stamped with the backfill time, newer than every message it contains, so it stayed eligible forever and re-ran nightly to no effect. Derive the activity signal from the message table at query time instead: - a new MessageActivityProvider counts root sessions with a message newer than the watermark (indexed message table, sub-ms); a missing provider or an unavailable store is treated conservatively as "run" (the executor bails before any child session); - the retrospective scanner makes message activity the eligibility driver, demoting the updated_at filter to the indexless-provider fallback. --- .../dreamer/message-activity.test.ts | 137 ++++++++++++++++++ .../magic-context/dreamer/message-activity.ts | 62 ++++++++ .../dreamer/retrospective-gate.test.ts | 63 +++++++- .../dreamer/retrospective-raw-provider.ts | 68 +++++---- .../magic-context/dreamer/task-gates.test.ts | 98 +++++++++++++ .../magic-context/dreamer/task-gates.ts | 45 ++++-- .../dreamer/task-scheduler.test.ts | 24 +++ .../magic-context/dreamer/task-scheduler.ts | 16 +- .../plugin/src/hooks/magic-context/hook.ts | 13 +- packages/plugin/src/plugin/dream-timer.ts | 23 ++- 10 files changed, 495 insertions(+), 54 deletions(-) create mode 100644 packages/plugin/src/features/magic-context/dreamer/message-activity.test.ts create mode 100644 packages/plugin/src/features/magic-context/dreamer/message-activity.ts diff --git a/packages/plugin/src/features/magic-context/dreamer/message-activity.test.ts b/packages/plugin/src/features/magic-context/dreamer/message-activity.test.ts new file mode 100644 index 0000000000..d5f161b540 --- /dev/null +++ b/packages/plugin/src/features/magic-context/dreamer/message-activity.test.ts @@ -0,0 +1,137 @@ +/// + +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 = "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/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(); + }); +}); diff --git a/packages/plugin/src/features/magic-context/dreamer/message-activity.ts b/packages/plugin/src/features/magic-context/dreamer/message-activity.ts new file mode 100644 index 0000000000..10d0f30783 --- /dev/null +++ b/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; + }, + }; +} diff --git a/packages/plugin/src/features/magic-context/dreamer/retrospective-gate.test.ts b/packages/plugin/src/features/magic-context/dreamer/retrospective-gate.test.ts index 95ce4710f7..bea902c798 100644 --- a/packages/plugin/src/features/magic-context/dreamer/retrospective-gate.test.ts +++ b/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, + /** Decouples updated_at (registration time) from the row timestamps. */ + private readonly updatedAtOverride?: Map, ) {} 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 diff --git a/packages/plugin/src/features/magic-context/dreamer/retrospective-raw-provider.ts b/packages/plugin/src/features/magic-context/dreamer/retrospective-raw-provider.ts index 3a17c9fd6f..15d4087dad 100644 --- a/packages/plugin/src/features/magic-context/dreamer/retrospective-raw-provider.ts +++ b/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, diff --git a/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts b/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts index f92ba4a1f2..c6f61b72f0 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts @@ -280,3 +280,101 @@ describe("evaluateTaskGate", () => { ).toBe(false); }); }); + +/** Stub message-activity provider returning a fixed count (null = store down). */ +function stubActivity(count: number | null): { + countRootSessionsWithMessagesSince: () => number | null; +} { + return { countRootSessionsWithMessagesSince: () => count }; +} + +describe("retrospective gate — message activity (session message store)", () => { + const projectIdentity = "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/repo/project"; + + test("gates on message activity, allowing when the store is down", () => { + db = freshDb(); + expect( + evaluateTaskGate("retrospective", { + db, + projectIdentity, + lastRunAt: null, + retrospectiveWatermarkMs: 100, + promotionThreshold: 3, + messageActivity: stubActivity(0), + }), + ).toBe(false); + expect( + evaluateTaskGate("retrospective", { + db, + projectIdentity, + lastRunAt: null, + retrospectiveWatermarkMs: 100, + promotionThreshold: 3, + messageActivity: stubActivity(1), + }), + ).toBe(true); + expect( + evaluateTaskGate("retrospective", { + db, + projectIdentity, + lastRunAt: null, + retrospectiveWatermarkMs: 100, + promotionThreshold: 3, + messageActivity: stubActivity(null), + }), + ).toBe(true); + }); + + test("forwards the CONTENT watermark to the provider", () => { + db = freshDb(); + const seen: (number | null)[] = []; + const capturing = { + countRootSessionsWithMessagesSince: (_project: string, sinceMs: number | null) => { + seen.push(sinceMs); + return 1; + }, + }; + // Unset watermark → provider sees null (never-run → any root session). + evaluateTaskGate("retrospective", { + db, + projectIdentity, + lastRunAt: null, + retrospectiveWatermarkMs: undefined, + promotionThreshold: 3, + messageActivity: capturing, + }); + // Set watermark → forwarded verbatim. + evaluateTaskGate("retrospective", { + db, + projectIdentity, + lastRunAt: null, + retrospectiveWatermarkMs: 500, + promotionThreshold: 3, + messageActivity: capturing, + }); + expect(seen).toEqual([null, 500]); + }); + + test("backlog uses the provider count when present", () => { + db = freshDb(); + expect( + getDreamTaskBacklog(db, projectIdentity, "retrospective", { + retrospectiveWatermarkMs: 100, + messageActivity: stubActivity(3), + }), + ).toEqual({ pending: 3, total: 3 }); + }); + + test("backlog falls back to the updated_at count when the provider is null", () => { + db = freshDb(); + db.prepare( + "INSERT INTO session_projects (session_id, harness, project_path, updated_at) VALUES (?, ?, ?, ?)", + ).run("s1", "opencode", projectIdentity, 200); + expect( + getDreamTaskBacklog(db, projectIdentity, "retrospective", { + retrospectiveWatermarkMs: 100, + messageActivity: stubActivity(null), + }), + ).toEqual({ pending: 1, total: 1 }); + }); +}); diff --git a/packages/plugin/src/features/magic-context/dreamer/task-gates.ts b/packages/plugin/src/features/magic-context/dreamer/task-gates.ts index 0a44005ff5..ce381db7c5 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-gates.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-gates.ts @@ -9,6 +9,7 @@ import { getPendingSmartNotes } from "../storage-notes"; import { countPrimerCandidatesForProject, getActivePrimers } from "../storage-primers"; import { getUserMemoryCandidates } from "../user-memory/storage-user-memory"; import { peekCurateCategoryScope } from "./curate-category-rotation"; +import type { MessageActivityProvider } from "./message-activity"; import { getTaskScheduleState } from "./storage-task-schedule"; import { CANONICAL_DREAM_TASKS, @@ -37,6 +38,9 @@ export interface TaskGateContext { retrospectiveWatermarkMs?: number | null; /** review-user-memories: min candidate observations before a review is worthwhile. */ promotionThreshold: number; + /** Optional message-activity signal (session message store). Absent → legacy + * gates; a provider returning null (store unavailable) → conservative allow. */ + messageActivity?: MessageActivityProvider; } /** Raw status count used only to let curate transition expired active rows. */ @@ -236,7 +240,11 @@ export function getDreamTaskBacklog( db: Database, projectPath: string, task: DreamTaskName, - options: { lastRunAt?: number | null; retrospectiveWatermarkMs?: number | null } = {}, + options: { + lastRunAt?: number | null; + retrospectiveWatermarkMs?: number | null; + messageActivity?: MessageActivityProvider; + } = {}, ): DreamTaskBacklog { switch (task) { case "map-memories": { @@ -292,11 +300,11 @@ export function getDreamTaskBacklog( }; } case "retrospective": { - const pending = countProjectSessionsSince( - db, - projectPath, - options.retrospectiveWatermarkMs ?? null, - ); + const since = options.retrospectiveWatermarkMs ?? null; + const pending = options.messageActivity + ? (options.messageActivity.countRootSessionsWithMessagesSince(projectPath, since) ?? + countProjectSessionsSince(db, projectPath, since)) + : countProjectSessionsSince(db, projectPath, since); return { pending, total: pending }; } case "maintain-docs": { @@ -332,7 +340,11 @@ export function getDreamTaskBacklogs( db: Database, projectPath: string, tasks: readonly DreamTaskName[] = CANONICAL_DREAM_TASKS, - options: { lastRunAt?: number | null; retrospectiveWatermarkMs?: number | null } = {}, + options: { + lastRunAt?: number | null; + retrospectiveWatermarkMs?: number | null; + messageActivity?: MessageActivityProvider; + } = {}, ): DreamTaskBacklogMap { const result: DreamTaskBacklogMap = {}; for (const task of tasks) result[task] = getDreamTaskBacklog(db, projectPath, task, options); @@ -384,11 +396,20 @@ export function evaluateTaskGate(task: DreamTaskName, ctx: TaskGateContext): boo return countLiveMemories(db, project) > 0; case "retrospective": - // Cheap pre-gate: any project session updated since the CONTENT - // watermark (max message ts actually scanned), not lastRunAt — a - // session updated mid-run would otherwise be skipped. The executor's - // raw provider does the precise typed-user-message scan and bails - // before any child session if empty. Never-run → "sessions exist". + // Cheap pre-gate: any project ROOT session with a message newer than the + // CONTENT watermark (max message ts actually scanned), not lastRunAt and + // not session_projects.updated_at — which records first-binding/backfill + // time, so it over-scans backfilled sessions and under-scans active ones. + // The executor's raw provider does the precise typed-user-message scan + // and bails before any child session if empty. Never-run → any root + // session; an unavailable message store → conservative allow. + if (ctx.messageActivity) { + const count = ctx.messageActivity.countRootSessionsWithMessagesSince( + project, + ctx.retrospectiveWatermarkMs ?? null, + ); + return count === null ? true : count > 0; + } return countProjectSessionsSince(db, project, ctx.retrospectiveWatermarkMs ?? null) > 0; case "maintain-docs": diff --git a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts index 2c302ddf50..3f67c3018f 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts @@ -690,3 +690,27 @@ describe("task-scheduler — runManualDream", () => { expect(result.ran).toEqual([]); }); }); + +/** Stub message-activity provider returning a fixed count (null = store down). */ +function stubActivity(count: number | null): { + countRootSessionsWithMessagesSince: () => number | null; +} { + return { countRootSessionsWithMessagesSince: () => count }; +} + +describe("task-scheduler — message-activity provider threading", () => { + it("runManualDream forwards the provider to retrospective backlog probes", async () => { + db = freshDb(); + const tasks = [cfg("retrospective", "0 3 * * *")]; + const executor = async (): Promise => ({ status: "completed" }); + const result = await runManualDream({ + db, + projectIdentity: PROJECT, + tasks, + executor, + task: "retrospective", + messageActivity: stubActivity(3), + }); + expect(result.backlogBefore.retrospective).toEqual({ pending: 3, total: 3 }); + }); +}); diff --git a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.ts b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.ts index a7bad2b05b..8a313520a8 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.ts @@ -9,6 +9,7 @@ import { leaseOwnershipMatches, releaseLease, } from "./lease"; +import type { MessageActivityProvider } from "./message-activity"; import { getDreamState } from "./storage-dream-state"; import { getTaskScheduleState, @@ -88,6 +89,9 @@ export interface RunDueTasksDeps { tasks: readonly DreamTaskRuntimeConfig[]; executor: TaskExecutor; now?: number; + /** Optional message-activity signal from the session message store; absent → + * legacy gates apply (see task-gates). */ + messageActivity?: MessageActivityProvider; } /** First-seed a task's schedule row if absent. next_due_at from cron(after now); @@ -361,6 +365,7 @@ async function runDomainGroup( due.config.task, ), promotionThreshold: due.config.promotionThreshold ?? 3, + messageActivity: deps.messageActivity, }); if (!gatePass) { advanceAfterRun(db, projectIdentity, due, Date.now(), "skipped", null); @@ -471,7 +476,9 @@ export async function runManualDream( if (selected.length === 0) return result; const selectedTaskNames = selected.map((config) => config.task); - result.backlogBefore = getDreamTaskBacklogs(deps.db, deps.projectIdentity, selectedTaskNames); + result.backlogBefore = getDreamTaskBacklogs(deps.db, deps.projectIdentity, selectedTaskNames, { + messageActivity: deps.messageActivity, + }); result.backlogAfter = { ...result.backlogBefore }; // Seed rows so completion advancement has a row to update. @@ -497,6 +504,7 @@ export async function runManualDream( d.config.task, ), promotionThreshold: d.config.promotionThreshold ?? 3, + messageActivity: deps.messageActivity, }); if (pass) gated.push(d); else result.skippedNoWork.push(d.config.task); @@ -506,6 +514,7 @@ export async function runManualDream( deps.db, deps.projectIdentity, selectedTaskNames, + { messageActivity: deps.messageActivity }, ); return result; } @@ -538,7 +547,9 @@ export async function runManualDream( ), ); result.backlogAfter = { - ...getDreamTaskBacklogs(deps.db, deps.projectIdentity, selectedTaskNames), + ...getDreamTaskBacklogs(deps.db, deps.projectIdentity, selectedTaskNames, { + messageActivity: deps.messageActivity, + }), ...runLocalBacklogs, }; return result; @@ -569,6 +580,7 @@ export async function runDueTasksForProject(deps: RunDueTasksDeps): Promise { - log("[dreamer] scheduled task run failed:", error); - }); + messageActivity, + }) + .catch((error: unknown) => { + log("[dreamer] scheduled task run failed:", error); + }) + .finally(() => { + messageActivity.dispose(); + }); }; const commandHandler = createMagicContextCommandHandler({ diff --git a/packages/plugin/src/plugin/dream-timer.ts b/packages/plugin/src/plugin/dream-timer.ts index 8b274efeb9..ba18e06f9e 100644 --- a/packages/plugin/src/plugin/dream-timer.ts +++ b/packages/plugin/src/plugin/dream-timer.ts @@ -3,6 +3,7 @@ import { statSync } from "node:fs"; import type { DreamerConfig } from "../config/schema/magic-context"; import type { ClassifyModuleClient } from "../features/magic-context/dreamer/classify"; import { acquireLease, releaseLease } from "../features/magic-context/dreamer/lease"; +import { createMessageActivityProvider } from "../features/magic-context/dreamer/message-activity"; import { openOpenCodeDb } from "../features/magic-context/dreamer/open-opencode-db"; import { historianOrphanStaleMs, @@ -537,14 +538,20 @@ async function sweepProject( onProgress: (progress, completedTask) => reg.onDreamerProgress?.(progress, completedTask), }); - const ran = await runDueTasksForProject({ - db, - projectIdentity: reg.projectIdentity, - tasks: runtimeConfigs, - executor, - }); - if (ran > 0) { - log(`[dreamer] timer tick (${origin}) ${reg.projectIdentity} — ran ${ran} task(s)`); + const messageActivity = createMessageActivityProvider({ contextDb: db, openOpenCodeDb }); + try { + const ran = await runDueTasksForProject({ + db, + projectIdentity: reg.projectIdentity, + tasks: runtimeConfigs, + executor, + messageActivity, + }); + if (ran > 0) { + log(`[dreamer] timer tick (${origin}) ${reg.projectIdentity} — ran ${ran} task(s)`); + } + } finally { + messageActivity.dispose(); } } catch (error) { log(`[dreamer] timer-triggered task scheduling failed for ${reg.projectIdentity}:`, error); From 59a923f83b3c8c768e185f6529145ae0b1af5186 Mon Sep 17 00:00:00 2001 From: Jona Date: Fri, 18 Sep 2026 03:02:40 +0200 Subject: [PATCH 2/2] feat(dreamer): gate the memory-pool tasks on session message activity verify, verify-broad, curate, compress-cues, and classify-memories gated only on pool existence, so an untouched project still took the memory lease and ran a whole-pool LLM pass on every cron slot with nothing new to process. Require session activity since the last successful run in addition to the pool half, reusing the MessageActivityProvider introduced for the retrospective: verify / compress-cues / classify-memories: live pool AND activity verify-broad: open cycle OR (live pool AND activity) curate: raw status pool AND activity (curate owns expiry hygiene) A missing provider or an unavailable message store is treated conservatively as "run", so hosts without an indexed message store keep their previous behavior. --- .../magic-context/dreamer/task-gates.test.ts | 120 ++++++++++++++++++ .../magic-context/dreamer/task-gates.ts | 46 +++++-- .../dreamer/task-scheduler.test.ts | 59 +++++++++ 3 files changed, 212 insertions(+), 13 deletions(-) diff --git a/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts b/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts index c6f61b72f0..8413883bc9 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-gates.test.ts @@ -12,6 +12,7 @@ import { } from "../memory"; import { runMigrations } from "../migrations"; import { initializeDatabase } from "../storage-db"; +import { writeTaskScheduleState } from "./storage-task-schedule"; import { evaluateTaskGate, getDreamTaskBacklog } from "./task-gates"; import { formatDreamTaskBacklogs, processedDreamTaskItems } from "./task-registry"; @@ -378,3 +379,122 @@ describe("retrospective gate — message activity (session message store)", () = ).toEqual({ pending: 1, total: 1 }); }); }); + +/** Give a project an active memory so memory-domain pool checks pass. */ +let memorySeq = 0; +function seedActiveMemory(d: Database, project = "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/repo/project"): void { + memorySeq += 1; + insertMemory(d, { + projectPath: project, + category: "PROJECT_RULES", + content: `mem-${memorySeq}`, + }); +} + +describe("evaluateTaskGate — memory tasks on session activity", () => { + const MEMORY_TASKS = ["verify", "curate", "compress-cues", "classify-memories"] as const; + const projectIdentity = "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/repo/project"; + + test("memory tasks need a pool AND session activity since the last run", () => { + db = freshDb(); + seedActiveMemory(db, projectIdentity); + for (const task of MEMORY_TASKS) { + expect( + evaluateTaskGate(task, { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(0), + }), + ).toBe(false); + expect( + evaluateTaskGate(task, { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(1), + }), + ).toBe(true); + } + }); + + test("memory tasks treat an unavailable message store as activity (conservative)", () => { + db = freshDb(); + seedActiveMemory(db, projectIdentity); + for (const task of MEMORY_TASKS) { + expect( + evaluateTaskGate(task, { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(null), + }), + ).toBe(true); + } + }); + + test("memory tasks still need a pool even when sessions changed", () => { + db = freshDb(); + for (const task of MEMORY_TASKS) { + expect( + evaluateTaskGate(task, { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(1), + }), + ).toBe(false); + } + }); + + test("verify-broad keeps an open cycle runnable with zero activity", () => { + db = freshDb(); + writeTaskScheduleState(db, { + projectPath: projectIdentity, + task: "verify-broad", + lastRunAt: null, + nextDueAt: Date.now() - 1000, + schedule: "0 3 * * 0", + lastStatus: null, + lastError: null, + retryCount: 0, + lastBroadRunAt: 123, + }); + expect( + evaluateTaskGate("verify-broad", { + db, + projectIdentity, + lastRunAt: null, + promotionThreshold: 3, + messageActivity: stubActivity(0), + }), + ).toBe(true); + }); + + test("verify-broad with a closed cycle requires pool AND activity", () => { + db = freshDb(); + seedActiveMemory(db, projectIdentity); + expect( + evaluateTaskGate("verify-broad", { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(0), + }), + ).toBe(false); + expect( + evaluateTaskGate("verify-broad", { + db, + projectIdentity, + lastRunAt: Date.now(), + promotionThreshold: 3, + messageActivity: stubActivity(1), + }), + ).toBe(true); + }); +}); diff --git a/packages/plugin/src/features/magic-context/dreamer/task-gates.ts b/packages/plugin/src/features/magic-context/dreamer/task-gates.ts index ce381db7c5..37c5cee9bf 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-gates.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-gates.ts @@ -43,6 +43,18 @@ export interface TaskGateContext { messageActivity?: MessageActivityProvider; } +/** True when any ROOT session of the project has a message newer than + * `sinceMs` (null → any root session). Conservative: a missing provider or + * an unavailable message store returns true so legacy gates still apply. */ +function hasSessionActivitySince(ctx: TaskGateContext, sinceMs: number | null): boolean { + if (!ctx.messageActivity) return true; + const count = ctx.messageActivity.countRootSessionsWithMessagesSince( + ctx.projectIdentity, + sinceMs, + ); + return count === null ? true : count > 0; +} + /** Raw status count used only to let curate transition expired active rows. */ export function countActiveMemories(db: Database, projectPath: string): number { const row = db @@ -366,34 +378,42 @@ export function evaluateTaskGate(task: DreamTaskName, ctx: TaskGateContext): boo return countUnmappedActiveMemories(db, project) > 0; case "verify": - // The executor's file gate does the precise incremental partition; the - // scheduler only avoids taking the memory lease when there is no live pool. - return countLiveMemories(db, project) > 0; + // Two-part gate: a LIVE memory pool AND session activity since the last + // successful run. The executor's file gate does the precise incremental + // partition; the scheduler only avoids taking the memory lease when there + // is no pool or no new session activity. + return countLiveMemories(db, project) > 0 && hasSessionActivitySince(ctx, lastRunAt); case "verify-broad": // Keep an open cycle runnable even when another task removed the last // active memory; the executor then closes the now-empty cycle. A closed - // cycle still needs an active pool before taking the memory lease. + // cycle still needs a live pool AND session activity since the last + // successful run before taking the memory lease. return ( getTaskScheduleState(db, project, "verify-broad")?.lastBroadRunAt != null || - countLiveMemories(db, project) > 0 + (countLiveMemories(db, project) > 0 && hasSessionActivitySince(ctx, lastRunAt)) ); case "curate": // Curate owns expiry hygiene, so its gate intentionally uses the raw // status pool: an expired-only project still needs one transition run. - return countActiveMemories(db, project) > 0; + // It additionally requires session activity since the last successful + // run, so a quiet project defers its expired-memory archive rather than + // spending a whole-pool LLM pass on an untouched project. + return countActiveMemories(db, project) > 0 && hasSessionActivitySince(ctx, lastRunAt); case "compress-cues": - // Cheap pre-gate: only take the memory lease when a live pool exists. The - // executor's selectCandidates does the precise NULL/stale-hash cue - // partition and no-ops when everything is already compressed. - return countLiveMemories(db, project) > 0; + // Two-part gate: a live pool AND session activity since the last + // successful run. The executor's selectCandidates does the precise + // NULL/stale-hash cue partition and no-ops when everything is already + // compressed. + return countLiveMemories(db, project) > 0 && hasSessionActivitySince(ctx, lastRunAt); case "classify-memories": - // Classification scores the live project memory pool directly. It has - // no file gate, watermark, or completeness prerequisites. - return countLiveMemories(db, project) > 0; + // Two-part gate: a live pool AND session activity since the last + // successful run — classification scores the live project memory pool + // directly. + return countLiveMemories(db, project) > 0 && hasSessionActivitySince(ctx, lastRunAt); case "retrospective": // Cheap pre-gate: any project ROOT session with a message newer than the diff --git a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts index 3f67c3018f..0e8049b55e 100644 --- a/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts +++ b/packages/plugin/src/features/magic-context/dreamer/task-scheduler.test.ts @@ -714,3 +714,62 @@ describe("task-scheduler — message-activity provider threading", () => { expect(result.backlogBefore.retrospective).toEqual({ pending: 3, total: 3 }); }); }); + +describe("task-scheduler — memory-activity gating", () => { + it("a provider reporting zero activity skips a due task even with a pool (pre-gate)", async () => { + db = freshDb(); + seedActiveMemory(db); + const now = Date.now(); + const tasks = [cfg("verify", "0 3 * * *")]; + planDueTasks(db, PROJECT, tasks, now); + forceDue(db, "verify", now); + + let ran = false; + const executor = async (): Promise => { + ran = true; + return { status: "completed" }; + }; + const count = await runDueTasksForProject({ + db, + projectIdentity: PROJECT, + tasks, + executor, + now, + messageActivity: stubActivity(0), + }); + expect(count).toBe(0); + expect(ran).toBe(false); + expect(getTaskScheduleState(db, PROJECT, "verify")?.lastStatus).toBe("skipped"); + }); + + it("the post-lease re-gate consults the provider (activity consumed mid-run)", async () => { + db = freshDb(); + seedActiveMemory(db); + const now = Date.now(); + const tasks = [cfg("verify", "0 3 * * *"), cfg("curate", "0 4 * * 0")]; + planDueTasks(db, PROJECT, tasks, now); + forceDue(db, "verify", now); + forceDue(db, "curate", now); + + // Both tasks pass the pre-gate with activity present; verify's executor + // consumes the new activity, so curate's POST-lease re-gate must fail. + let activity = 1; + const provider = { countRootSessionsWithMessagesSince: () => activity }; + const ran: string[] = []; + const executor = async (c: DreamTaskRuntimeConfig): Promise => { + ran.push(c.task); + if (c.task === "verify") activity = 0; + return { status: "completed" }; + }; + await runDueTasksForProject({ + db, + projectIdentity: PROJECT, + tasks, + executor, + now, + messageActivity: provider, + }); + expect(ran).toEqual(["verify"]); + expect(getTaskScheduleState(db, PROJECT, "curate")?.lastStatus).toBe("skipped"); + }); +});