feat(orchestrator): restore pending summary markers (#12199) (#12273)

neo-gpt · tobiu · web-flow · commit aa099876a1e9 · 2026-05-31T22:56:54.000+02:00
Co-authored-by: tobiu &lt;tobiasuhlig78@gmail.com&gt;
diff --git a/ai/daemons/orchestrator/scheduling/summary.mjs b/ai/daemons/orchestrator/scheduling/summary.mjs
@@ -3,12 +3,44 @@ import {
     markNodesAsRead
 } from '../../bridge/queries.mjs';
 
+/**
+ * Reads pending low-latency summarization markers from the coordinator table.
+ *
+ * @summary Keeps the scheduler's pending-lane check cheap: it only needs to know
+ * whether pending rows exist, while the spawned summary child remains responsible
+ * for lease-claiming and draining the actual jobs.
+ * @param {Object} db SQLite database handle.
+ * @param {Object} [options]
+ * @param {Number} [options.limit=50] Maximum marker ids to return.
+ * @returns {String[]}
+ */
+export function getPendingSummarizationJobs(db, {limit = 50} = {}) {
+    if (!db?.prepare) {
+        return [];
+    }
+
+    const numericLimit = Number.isInteger(limit) && limit > 0 ? limit : 50;
+
+    try {
+        return db.prepare(`
+            SELECT session_id
+            FROM SummarizationJobs
+            WHERE status = 'pending'
+            ORDER BY rowid ASC
+            LIMIT ?
+        `).all(numericLimit).map(row => row.session_id).filter(Boolean);
+    } catch {
+        return [];
+    }
+}
+
 /**
  * Builds the task trigger for the summarization sweep lane.
  *
- * Two wake-up sources, in priority order:
+ * Three wake-up sources, in priority order:
  * 1. Unread sunset handovers — priority because they unblock the next agent boot
- * 2. Periodic sweep — fallback for ordinary unsummarized sessions
+ * 2. Pending disconnect markers — targeted low-latency session close handling
+ * 3. Periodic sweep — fallback for ordinary unsummarized sessions
  *
  * Pure function. Test the scheduling contract without mounting the SQLite graph.
  *
@@ -17,9 +49,10 @@ import {
  * @param {Number} options.lastRunAt Last summary task start timestamp.
  * @param {Number} options.intervalMs Periodic sweep interval; `0` disables the interval source.
  * @param {Object[]} [options.handovers=[]] Unread sunset-handover message nodes.
+ * @param {String[]} [options.pendingJobs=[]] Pending SummarizationJobs session ids.
  * @returns {Object|null} A summary task trigger or null when no work is due.
  */
-export function buildSummaryTrigger({now, lastRunAt, intervalMs, handovers = []}) {
+export function buildSummaryTrigger({now, lastRunAt, intervalMs, handovers = [], pendingJobs = []}) {
     if (handovers.length > 0) {
         return {
             taskName     : 'summary',
@@ -29,6 +62,15 @@ export function buildSummaryTrigger({now, lastRunAt, intervalMs, handovers = []}
         };
     }
 
+    if (pendingJobs.length > 0) {
+        return {
+            taskName    : 'summary',
+            source      : 'pending-summarization',
+            reason      : `pending-summarization:${pendingJobs.length}`,
+            pendingCount: pendingJobs.length
+        };
+    }
+
     if (intervalMs > 0 && now - lastRunAt >= intervalMs) {
         return {
             taskName: 'summary',
@@ -50,6 +92,7 @@ export function buildSummaryTrigger({now, lastRunAt, intervalMs, handovers = []}
  * @param {Number} options.now Current timestamp in milliseconds.
  * @param {Number} options.summarySweepIntervalMs Periodic summary sweep interval.
  * @param {Function} [options.getUnreadSunsetHandoversFn] Test seam for handover reads.
+ * @param {Function} [options.getPendingSummarizationJobsFn] Test seam for pending marker reads.
  * @param {Function} [options.markNodesAsReadFn] Test seam for handover mark-read writes.
  * @param {Function} [options.log] Optional orchestrator log function.
  * @returns {Object|null} Task trigger with optional `onSuccess` callback, or null.
@@ -60,13 +103,16 @@ export function getDueTask({
     now,
     summarySweepIntervalMs,
     getUnreadSunsetHandoversFn = getUnreadSunsetHandovers,
+    getPendingSummarizationJobsFn = getPendingSummarizationJobs,
     markNodesAsReadFn          = markNodesAsRead,
     log
 }) {
     const handovers = getUnreadSunsetHandoversFn(db);
+    const pendingJobs = handovers.length > 0 ? [] : getPendingSummarizationJobsFn(db);
     const trigger   = buildSummaryTrigger({
         now,
         handovers,
+        pendingJobs,
         intervalMs: summarySweepIntervalMs,
         lastRunAt : state.summary?.lastRunAt || 0
     });
diff --git a/ai/mcp/server/memory-core/Server.mjs b/ai/mcp/server/memory-core/Server.mjs
@@ -221,15 +221,18 @@ class Server extends BaseServer {
     /**
      * @summary SSE-only hook fired by `TransportService` on session disconnect. Removes the
      * per-session McpServer from `CoalescingEngineService`'s broadcast set (counterpart to
-     * `createMcpServer`'s `addMcpServer` registration). Summarization is orchestrator-driven
-     * (the `summary` task), not triggered on disconnect.
+     * `createMcpServer`'s `addMcpServer` registration) and writes a cheap idempotent
+     * `SummarizationJobs.pending` marker. The orchestrator `summary` task drains that marker;
+     * this hook never summarizes inline.
      * @param {String} sessionId
      * @param {Object} mcpServerInstance
      */
     onSessionClosed(sessionId, mcpServerInstance) {
         if (mcpServerInstance) {
             CoalescingEngineService.removeMcpServer(mcpServerInstance);
         }
+
+        SessionService.queueSummarizationJob(sessionId);
     }
 
     /**
diff --git a/ai/scripts/lifecycle/summarize-sessions.mjs b/ai/scripts/lifecycle/summarize-sessions.mjs
@@ -27,7 +27,17 @@ async function summarize() {
     try {
         await Memory_SessionService.initAsync();
 
-        console.log('[summarize-sessions] Starting session summarization...');
+        console.log('[summarize-sessions] Draining pending session summarization markers...');
+        const pendingResult = await Memory_SessionService.summarizePendingSessions();
+
+        if (pendingResult && pendingResult.error) {
+            console.error(`[summarize-sessions] Pending summarization failed: ${pendingResult.message}`);
+            process.exit(1);
+        }
+
+        console.log(`[summarize-sessions] Pending drain complete. Pending: ${pendingResult?.pending || 0}; processed: ${pendingResult?.processed || 0}`);
+
+        console.log('[summarize-sessions] Starting drift-detection session summarization...');
         // includeAll: false will only summarize sessions from the last 30 days
         const result = await Memory_SessionService.summarizeSessions({ includeAll: false });
 
diff --git a/ai/services/memory-core/SessionService.mjs b/ai/services/memory-core/SessionService.mjs
@@ -953,6 +953,109 @@ ${aggregatedContent}
         };
     }
 
+    /**
+     * Queues a low-latency summarization marker for a disconnected session.
+     *
+     * @summary Writes an idempotent `pending` SummarizationJobs row without running
+     * summarization inline. Multiple MCP server instances can race this cheap marker
+     * safely; the orchestrator-owned summary lane remains the only drainer and the
+     * existing lease transition serializes the expensive summarization work.
+     * @param {String} sessionId Session id whose transport disconnected.
+     * @returns {Boolean} true when the marker was written, false otherwise.
+     */
+    queueSummarizationJob(sessionId) {
+        if (!sessionId || typeof sessionId !== 'string') {
+            return false;
+        }
+
+        const db = GraphService.db?.storage?.db;
+        if (!db) {
+            logger.warn(`[SessionService] queueSummarizationJob skipped for ${sessionId}: SQLite graph is not available.`);
+            return false;
+        }
+
+        try {
+            db.prepare(`
+                INSERT INTO SummarizationJobs (session_id, status, lease_token, expires_at, retry_count)
+                VALUES (?, 'pending', NULL, NULL, 0)
+                ON CONFLICT(session_id) DO UPDATE SET
+                    status      = 'pending',
+                    lease_token = NULL,
+                    expires_at  = NULL
+                WHERE SummarizationJobs.status != 'completed'
+                    AND SummarizationJobs.status != 'in_progress'
+            `).run(sessionId);
+
+            return true;
+        } catch (e) {
+            logger.warn(`[SessionService] Error queueing summarization job for ${sessionId}: ${e.message}`);
+            return false;
+        }
+    }
+
+    /**
+     * Reads pending low-latency summarization job ids in insertion order.
+     *
+     * @summary Provides the orchestrator child process with the exact pending lane
+     * payload before it falls back to the broader drift sweep.
+     * @param {Object} [options]
+     * @param {Number} [options.limit=50] Maximum pending rows to drain in one child run.
+     * @returns {String[]}
+     */
+    getPendingSummarizationJobIds({limit = 50} = {}) {
+        const db = GraphService.db?.storage?.db;
+        if (!db) return [];
+
+        const numericLimit = Number.isInteger(limit) && limit > 0 ? limit : 50;
+
+        try {
+            return db.prepare(`
+                SELECT session_id
+                FROM SummarizationJobs
+                WHERE status = 'pending'
+                ORDER BY rowid ASC
+                LIMIT ?
+            `).all(numericLimit).map(row => row.session_id).filter(Boolean);
+        } catch (e) {
+            logger.warn(`[SessionService] Error reading pending summarization jobs: ${e.message}`);
+            return [];
+        }
+    }
+
+    /**
+     * Drains pending summarization markers through the existing lease-backed summarizer.
+     *
+     * @summary Reuses {@link summarizeSessions} with explicit session ids so each
+     * `pending` row travels through `claimSummarizationJob()` and cannot be processed
+     * twice by overlapping pending/drift drains.
+     * @param {Object} [options]
+     * @param {Number} [options.limit=50] Maximum pending rows to drain in one child run.
+     * @returns {Promise<{processed: number, sessions: object[], pending: number}>}
+     */
+    async summarizePendingSessions({limit = 50} = {}) {
+        const sessionIds = this.getPendingSummarizationJobIds({limit});
+        const processed  = [];
+
+        for (const sessionId of sessionIds) {
+            const result = await this.summarizeSessions({sessionId});
+
+            if (result?.error) {
+                logger.warn(`[SessionService] Pending summarization failed for ${sessionId}: ${result.message}`);
+                continue;
+            }
+
+            if (Array.isArray(result?.sessions)) {
+                processed.push(...result.sessions);
+            }
+        }
+
+        return {
+            pending  : sessionIds.length,
+            processed: processed.length,
+            sessions : processed
+        };
+    }
+
     /**
      * Claims an exclusive lease on a summarization job using the SummarizationJobs table.
      * Prevents race conditions across concurrent MCP instances.
diff --git a/test/playwright/unit/ai/daemons/orchestrator/scheduling/summary.spec.mjs b/test/playwright/unit/ai/daemons/orchestrator/scheduling/summary.spec.mjs
@@ -1,6 +1,7 @@
 import {test, expect} from '@playwright/test';
 import {
     buildSummaryTrigger,
+    getPendingSummarizationJobs,
     getDueTask
 } from '../../../../../../../ai/daemons/orchestrator/scheduling/summary.mjs';
 
@@ -19,6 +20,36 @@ test.describe('orchestrator/scheduling/summary (#11864 / Epic #11831)', () => {
         });
     });
 
+    test('#12199: buildSummaryTrigger prioritizes pending summarization markers before the periodic sweep', () => {
+        expect(buildSummaryTrigger({
+            now        : 600000,
+            lastRunAt  : 0,
+            intervalMs : 600000,
+            handovers  : [],
+            pendingJobs: ['session-1', 'session-2']
+        })).toEqual({
+            taskName    : 'summary',
+            source      : 'pending-summarization',
+            reason      : 'pending-summarization:2',
+            pendingCount: 2
+        });
+    });
+
+    test('#12199: sunset handovers stay higher priority than pending summarization markers', () => {
+        expect(buildSummaryTrigger({
+            now        : 600000,
+            lastRunAt  : 0,
+            intervalMs : 600000,
+            handovers  : [{id: 'MESSAGE:1'}],
+            pendingJobs: ['session-1']
+        })).toEqual({
+            taskName     : 'summary',
+            source       : 'sunset-handover',
+            reason       : 'sunset-handover:1',
+            handoverCount: 1
+        });
+    });
+
     test('buildSummaryTrigger returns a periodic sweep trigger only when the interval is due', () => {
         expect(buildSummaryTrigger({now: 599999, lastRunAt: 0, intervalMs: 600000, handovers: []})).toBeNull();
 
@@ -60,6 +91,24 @@ test.describe('orchestrator/scheduling/summary (#11864 / Epic #11831)', () => {
         expect(result.onSuccess).toBeUndefined();
     });
 
+    test('#12199: getDueTask reads pending markers when no handover is waiting', () => {
+        const result = getDueTask({
+            db                            : 'mock-db',
+            state                         : {summary: {lastRunAt: 0}},
+            now                           : 100,
+            summarySweepIntervalMs        : 600000,
+            getUnreadSunsetHandoversFn    : () => [],
+            getPendingSummarizationJobsFn : () => ['session-1']
+        });
+
+        expect(result).toEqual({
+            taskName    : 'summary',
+            source      : 'pending-summarization',
+            reason      : 'pending-summarization:1',
+            pendingCount: 1
+        });
+    });
+
     test('getDueTask returns sunset-handover trigger with onSuccess callback', () => {
         const markCalls = [];
         const handovers = [{id: 'MESSAGE:1'}, {id: 'MESSAGE:2'}];
@@ -79,4 +128,23 @@ test.describe('orchestrator/scheduling/summary (#11864 / Epic #11831)', () => {
         result.onSuccess();
         expect(markCalls).toEqual([{db: 'mock-db', nodes: handovers}]);
     });
+
+    test('#12199: getPendingSummarizationJobs reads pending session ids only', () => {
+        const calls = [];
+        const db = {
+            prepare: (sql) => {
+                calls.push(sql);
+                return {
+                    all: (limit) => {
+                        calls.push(limit);
+                        return [{session_id: 'session-1'}, {session_id: 'session-2'}, {session_id: null}];
+                    }
+                };
+            }
+        };
+
+        expect(getPendingSummarizationJobs(db, {limit: 2})).toEqual(['session-1', 'session-2']);
+        expect(calls[0]).toContain("WHERE status = 'pending'");
+        expect(calls[1]).toBe(2);
+    });
 });
diff --git a/test/playwright/unit/ai/mcp/server/memory-core/Server.spec.mjs b/test/playwright/unit/ai/mcp/server/memory-core/Server.spec.mjs
@@ -141,4 +141,34 @@ test.describe('Neo.ai.mcp.server.memory-core.Server', () => {
 
         serverInstance.destroy();
     });
+
+    test('#12199: onSessionClosed writes a pending summarization marker without summarizing inline', async () => {
+        const SDK = await import('../../../../../../../ai/services.mjs');
+        const serverInstance = Neo.create('Neo.ai.mcp.server.memory-core.Server');
+        const mcpServerInstance = {id: 'mcp-session-server'};
+        const calls = [];
+        const removed = [];
+
+        const originalQueue = SDK.Memory_SessionService.queueSummarizationJob;
+        const originalRemove = SDK.Memory_CoalescingEngineService.removeMcpServer;
+
+        SDK.Memory_SessionService.queueSummarizationJob = (sessionId) => {
+            calls.push(sessionId);
+            return true;
+        };
+        SDK.Memory_CoalescingEngineService.removeMcpServer = (instance) => {
+            removed.push(instance);
+        };
+
+        try {
+            serverInstance.onSessionClosed('closed-session-1', mcpServerInstance);
+
+            expect(removed).toEqual([mcpServerInstance]);
+            expect(calls).toEqual(['closed-session-1']);
+        } finally {
+            SDK.Memory_SessionService.queueSummarizationJob = originalQueue;
+            SDK.Memory_CoalescingEngineService.removeMcpServer = originalRemove;
+            serverInstance.destroy();
+        }
+    });
 });
diff --git a/test/playwright/unit/ai/services/memory-core/SessionService.ResumeValidation.spec.mjs b/test/playwright/unit/ai/services/memory-core/SessionService.ResumeValidation.spec.mjs
