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 000000000..d5f161b54 --- /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 000000000..10d0f3078 --- /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 95ce4710f..bea902c79 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 3a17c9fd6..15d4087da 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 f92ba4a1f..c6f61b72f 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 0a44005ff..ce381db7c 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 2c302ddf5..3f67c3018 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 a7bad2b05..8a313520a 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 8b274efeb..ba18e06f9 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);