From abfdcc112b52b73a07500c640ed6f7a91c46eb86 Mon Sep 17 00:00:00 2001 From: Zqc <24S103279@stu.hit.edu.cn> Date: Fri, 19 Jun 2026 08:17:45 +0800 Subject: [PATCH] fix(core): prevent OOM in auto-memory extraction during /quit (#5147) (#5181) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The auto-memory extraction background task caused FATAL ERROR: Reached heap limit after /quit because buildTranscriptMessages() ran .replace() over every history message synchronously, but the resulting text was never consumed downstream. Changes: - extract.ts: delete buildTranscriptMessages/loadUnprocessedTranscriptSlice. Replace with a zero-stringify cursor scan on the unprocessed slice only. - manager.ts: add isUnderMemoryPressure() using existing MemoryPressureMonitor. Gates in runExtract(), scheduleDream(), and scheduleSkillReview(). - client.ts: add shutdownRequested flag + requestShutdown(). - AppContainer.tsx: call requestShutdown() in /quit callback. - Tests: 46 passed (11 extract + 34 manager + 2 client). - Docs: update memory-system.md flowchart to cursor-first flow. Co-authored-by: 俊良 --- docs/design/auto-memory/memory-system.md | 31 +- packages/cli/src/ui/AppContainer.tsx | 5 + packages/core/src/core/client.test.ts | 119 +++++++ packages/core/src/core/client.ts | 19 ++ packages/core/src/memory/extract.test.ts | 344 ++++++++++++++++++-- packages/core/src/memory/extract.ts | 78 ++--- packages/core/src/memory/manager.test.ts | 396 +++++++++++++++++++++++ packages/core/src/memory/manager.ts | 75 ++++- packages/core/src/telemetry/types.ts | 12 +- 9 files changed, 987 insertions(+), 92 deletions(-) diff --git a/docs/design/auto-memory/memory-system.md b/docs/design/auto-memory/memory-system.md index c3c9624604..06af35af5e 100644 --- a/docs/design/auto-memory/memory-system.md +++ b/docs/design/auto-memory/memory-system.md @@ -215,27 +215,28 @@ flowchart TD ```mermaid flowchart TD A[runAutoMemoryExtract] --> B[ensureAutoMemoryScaffold\n初始化目录和文件] - B --> C[buildTranscriptMessages\n将 Content[] 转换为带 offset 的消息列表] - C --> D[readExtractCursor\n读取上次处理到的位置] - D --> E[loadUnprocessedTranscriptSlice\n截取未处理的消息段] - E --> F{slice 为空?} - F -- 是 --> G[返回无 patches 结果] - F -- 否 --> H[runAutoMemoryExtractionByAgent\n调用 forked agent 提取 patches] - H --> I[dedupeExtractPatches\n去重+规范化] - I --> J{有 touched topics?} - J -- 是 --> K[bumpMetadata\n更新 meta.json] - K --> L[rebuildManagedAutoMemoryIndex\n重建 MEMORY.md] - L --> M[writeExtractCursor\n记录最新 offset] - J -- 否 --> M - M --> N[返回 AutoMemoryExtractResult] + B --> C[readExtractCursor\n读取上次处理到的位置] + C --> D[history.slice startOffset\n只取未处理的消息切片] + D --> E{slice 有新的 user 消息?} + E -- 否 --> F[更新 cursor\n返回无 patches 结果] + E -- 是 --> G[runAutoMemoryExtractionByAgent\n调用 forked agent 提取] + G --> H{有 touched topics?} + H -- 是 --> I[bumpMetadata\n更新 meta.json] + I --> J[rebuildManagedAutoMemoryIndex\n重建 MEMORY.md] + J --> K[writeExtractCursor\n记录最新 offset = history.length] + H -- 否 --> K + K --> L[返回 AutoMemoryExtractResult] ``` +> **注意：** `isUnderMemoryPressure` 门控位于 `MemoryManager.runExtract()` 中，不在本流程内。当 monitor 报告 hard/critical 压力时，`MemoryManager` 会跳过 extract 调用，不推进 cursor。 + **提取游标（Cursor）**： - 字段：`{ sessionId, processedOffset, updatedAt }` -- 每次提取后更新 `processedOffset` 为当前历史长度 -- 下次提取时，只处理 `offset >= processedOffset` 的消息 +- 提取前先通过 `readExtractCursor` 读取当前进度，再用 `history.slice(processedOffset)` 仅处理未读部分 +- 每次提取后更新 `processedOffset` 为当前历史长度（`params.history.length`） - 跨会话时（`sessionId` 变化）从偏移量 0 重新开始 +- 注意：不再通过 `buildTranscriptMessages` / `loadUnprocessedTranscriptSlice` 构建转录文本——`hasNewUserMessages` 通过 `history.slice(startOffset).some(m => m.role === 'user' && partToString(m.parts).trim().length > 0)` 判断，仅在未读切片上做轻量字符串化，全量历史不再处理 **Patch 过滤规则**： diff --git a/packages/cli/src/ui/AppContainer.tsx b/packages/cli/src/ui/AppContainer.tsx index 48d78dfe4b..6abbe7f86e 100644 --- a/packages/cli/src/ui/AppContainer.tsx +++ b/packages/cli/src/ui/AppContainer.tsx @@ -1155,6 +1155,10 @@ export const AppContainer = (props: AppContainerProps) => { openApprovalModeDialog, quit: (messages: HistoryItem[]) => { setQuittingMessages(messages); + // Signal the client to skip background memory tasks (extract, dream, + // skill review) so the process can exit without spawning new agent + // work during the exit window. + config.getGeminiClient()?.requestShutdown(); setTimeout(async () => { await runExitCleanup(); process.exit(0); @@ -1206,6 +1210,7 @@ export const AppContainer = (props: AppContainerProps) => { openDeleteDialog, openHelpDialog, openDiffDialog, + config, ], ); diff --git a/packages/core/src/core/client.test.ts b/packages/core/src/core/client.test.ts index bfcf47fd37..d72e0c4f48 100644 --- a/packages/core/src/core/client.test.ts +++ b/packages/core/src/core/client.test.ts @@ -8103,4 +8103,123 @@ Other open files: }); /* eslint-enable @typescript-eslint/no-explicit-any */ }); + + describe('#5147 shutdown gate', () => { + /** + * C1: requestShutdown() makes runManagedAutoMemoryBackgroundTasks a + * no-op. We drive the private method directly: before shutdown it + * schedules extract + dream; after shutdown it schedules neither. + */ + it('skips background memory tasks after shutdown is requested', () => { + const scheduleExtractSpy = vi.fn().mockResolvedValue({ + touchedTopics: [], + cursor: { sessionId: 'sess', updatedAt: new Date().toISOString() }, + }); + const scheduleDreamSpy = vi + .fn() + .mockResolvedValue({ status: 'skipped', skippedReason: 'locked' }); + + const mgr = { + scheduleExtract: scheduleExtractSpy, + scheduleDream: scheduleDreamSpy, + recall: vi.fn(), + scheduleSkillReview: vi + .fn() + .mockReturnValue({ status: 'skipped', skippedReason: 'disabled' }), + }; + + const client = new GeminiClient(makeMockConfigForShutdown(mgr)); + // Avoid needing a real chat — the method calls getHistoryShallow(). + ( + client as unknown as { getHistoryShallow: () => unknown[] } + ).getHistoryShallow = () => []; + + const runBgTasks = ( + client as unknown as { + runManagedAutoMemoryBackgroundTasks: (t: SendMessageType) => void; + } + ).runManagedAutoMemoryBackgroundTasks.bind(client); + + // Before shutdown: a completed UserQuery turn schedules extract + dream. + runBgTasks(SendMessageType.UserQuery); + expect(scheduleExtractSpy).toHaveBeenCalledTimes(1); + expect(scheduleDreamSpy).toHaveBeenCalledTimes(1); + + scheduleExtractSpy.mockClear(); + scheduleDreamSpy.mockClear(); + + // After shutdown: the gate short-circuits before any scheduling. + client.requestShutdown(); + runBgTasks(SendMessageType.UserQuery); + expect(scheduleExtractSpy).not.toHaveBeenCalled(); + expect(scheduleDreamSpy).not.toHaveBeenCalled(); + }); + + /** + * C2: requestShutdown() is idempotent — calling it multiple times + * should not throw or have side effects. + */ + it('is idempotent when called multiple times', () => { + const mgr = { + scheduleExtract: vi.fn(), + scheduleDream: vi.fn(), + recall: vi.fn(), + scheduleSkillReview: vi + .fn() + .mockReturnValue({ status: 'skipped', skippedReason: 'disabled' }), + }; + const cfg = makeMockConfigForShutdown(mgr); + const client = new GeminiClient(cfg); + + // Should not throw on first call + expect(() => client.requestShutdown()).not.toThrow(); + // Should not throw on second call + expect(() => client.requestShutdown()).not.toThrow(); + // Should not throw on third call + expect(() => client.requestShutdown()).not.toThrow(); + }); + }); }); + +function makeMockConfigForShutdown( + mgr: Record>, +): Config { + return { + isBareMode: vi.fn().mockReturnValue(false), + getGeminiClient: vi.fn().mockReturnValue(undefined), + getProjectRoot: vi.fn().mockReturnValue('/project'), + getSessionId: vi.fn().mockReturnValue('session-1'), + getMemoryManager: vi.fn().mockReturnValue(mgr), + getManagedAutoMemoryEnabled: vi.fn().mockReturnValue(true), + getManagedAutoDreamEnabled: vi.fn().mockReturnValue(true), + getAutoSkillEnabled: vi.fn().mockReturnValue(false), + getModel: vi.fn().mockReturnValue('test-model'), + getBaseLlmClient: vi.fn().mockReturnValue({ + generateContent: vi.fn(), + }), + getContentGenerator: vi.fn().mockReturnValue({ + generateContent: vi.fn(), + }), + getToolRegistry: vi.fn().mockReturnValue({ + getDeclarations: vi.fn().mockReturnValue([]), + getTools: vi.fn().mockReturnValue([]), + }), + getPromptRegistry: vi.fn().mockReturnValue({ + getDeclarations: vi.fn().mockReturnValue([]), + }), + getFileReadCache: vi.fn().mockReturnValue({ + clear: vi.fn(), + }), + getExtensionLoader: vi.fn().mockReturnValue(undefined), + getWorkspaceContext: vi.fn().mockReturnValue(undefined), + getDebugMode: vi.fn().mockReturnValue(false), + getApprovalMode: vi.fn().mockReturnValue('default'), + logEvent: vi.fn(), + getTelemetryService: vi.fn().mockReturnValue(undefined), + getHookSystem: vi.fn().mockReturnValue(undefined), + getMaxSessionTurns: vi.fn().mockReturnValue(100), + getChatRecordingService: vi.fn().mockReturnValue(undefined), + isInteractive: vi.fn().mockReturnValue(false), + getStdinReader: vi.fn().mockReturnValue(undefined), + } as unknown as Config; +} diff --git a/packages/core/src/core/client.ts b/packages/core/src/core/client.ts index c79ce048c9..0d11092bf4 100644 --- a/packages/core/src/core/client.ts +++ b/packages/core/src/core/client.ts @@ -211,6 +211,7 @@ export class GeminiClient { private skillsModifiedInSession = false; private cachedGitStatus: string | null | undefined; private readonly surfacedRelevantAutoMemoryPaths = new Set(); + private shutdownRequested = false; private readonly loopDetector: LoopDetectionService; private lastPromptId: string | undefined = undefined; @@ -596,6 +597,15 @@ export class GeminiClient { }); } + /** + * Signal that shutdown is imminent. Subsequent calls to background memory + * tasks (extract, dream, skill review) will be skipped so the process can + * exit cleanly without spawning new work. + */ + requestShutdown(): void { + this.shutdownRequested = true; + } + /** * Abort and release the pending auto-memory prefetch in one step. * Safe to call when no prefetch is pending — does nothing. Centralises @@ -1380,6 +1390,15 @@ export class GeminiClient { private runManagedAutoMemoryBackgroundTasks( messageType: SendMessageType, ): void { + // During shutdown, skip all background memory tasks so the process + // can exit cleanly without spawning new work. + if (this.shutdownRequested) { + debugLogger.debug( + 'Skipping background memory tasks: shutdown requested.', + ); + return; + } + // autoSkill counts tool calls and can trigger on both UserQuery and // ToolResult turns so the threshold can fire mid-session. if ( diff --git a/packages/core/src/memory/extract.test.ts b/packages/core/src/memory/extract.test.ts index e229b632fe..1909e6af1b 100644 --- a/packages/core/src/memory/extract.test.ts +++ b/packages/core/src/memory/extract.test.ts @@ -9,12 +9,9 @@ import * as os from 'node:os'; import * as path from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { Config } from '../config/config.js'; +import type { Content } from '@google/genai'; import { getAutoMemoryExtractCursorPath } from './paths.js'; -import { - buildTranscriptMessages, - loadUnprocessedTranscriptSlice, - runAutoMemoryExtract, -} from './extract.js'; +import { runAutoMemoryExtract } from './extract.js'; import { runAutoMemoryExtractionByAgent } from './extractionAgentPlanner.js'; import { ensureAutoMemoryScaffold } from './store.js'; import { @@ -57,24 +54,6 @@ describe('auto-memory extraction', () => { }); }); - it('builds transcript slices from history and cursor state', () => { - const transcript = buildTranscriptMessages([ - { role: 'user', parts: [{ text: 'hello' }] }, - { role: 'model', parts: [{ text: 'world' }] }, - { role: 'user', parts: [{ text: 'I prefer terse responses.' }] }, - ]); - - const slice = loadUnprocessedTranscriptSlice('session-1', transcript, { - sessionId: 'session-1', - processedOffset: 2, - updatedAt: new Date().toISOString(), - }); - - expect(slice.messages).toHaveLength(1); - expect(slice.messages[0]?.text).toBe('I prefer terse responses.'); - expect(slice.nextProcessedOffset).toBe(3); - }); - it('updates cursor and avoids duplicate writes for repeated extraction', async () => { vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ touchedTopics: [], @@ -235,4 +214,323 @@ describe('auto-memory extraction', () => { expect(rebuildUserAutoMemoryIndex).not.toHaveBeenCalled(); }); }); + + describe('#5147 OOM regression', () => { + /** + * A1: cursor-first — runAutoMemoryExtract only processes the unread + * portion of history. The first call processes all messages; the second + * call (with only a few new messages appended) should NOT reprocess + * the already-processed prefix. + */ + it('only processes unread messages via cursor-first ordering', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: ['user'], + touchedProjectScope: true, + touchedUserScope: false, + systemMessage: undefined, + }); + + // Build 20 messages (10 turns of user+model) + const history: Content[] = []; + for (let i = 0; i < 20; i++) { + history.push({ + role: i % 2 === 0 ? 'user' : 'model', + parts: [ + { text: `[MSG${i}] `.padEnd(16, '-') + `content for message ${i}` }, + ], + }); + } + + // First extract: processes all 20 messages + const first = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + expect(first.cursor.processedOffset).toBe(20); + expect(runAutoMemoryExtractionByAgent).toHaveBeenCalledTimes(1); + + // Add 2 new messages (1 turn) + history.push( + { role: 'user', parts: [{ text: 'new user question?' }] }, + { role: 'model', parts: [{ text: 'new assistant answer.' }] }, + ); + + // Second extract: should detect only the 2 new messages + const agentCallsBefore = vi.mocked(runAutoMemoryExtractionByAgent).mock + .calls.length; + const second = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + // Fork agent should have been called again (new user message found) + expect(vi.mocked(runAutoMemoryExtractionByAgent).mock.calls.length).toBe( + agentCallsBefore + 1, + ); + // Cursor advances to full history length + expect(second.cursor.processedOffset).toBe(22); + }); + + /** + * A2: when the cursor is already at the end of history (no new user + * messages), runAutoMemoryExtract skips without calling the fork agent. + */ + it('skips extract when cursor is already up to date', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: [], + touchedProjectScope: false, + touchedUserScope: false, + systemMessage: undefined, + }); + + const history: Content[] = [ + { role: 'user', parts: [{ text: 'hello' }] }, + { role: 'model', parts: [{ text: 'hi' }] }, + ]; + + // First extract: cursor → 2 + await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + const agentCallsBefore = vi.mocked(runAutoMemoryExtractionByAgent).mock + .calls.length; + + // Second extract with same 2 messages: no new user messages + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + // Fork agent should NOT be called again + expect(vi.mocked(runAutoMemoryExtractionByAgent).mock.calls.length).toBe( + agentCallsBefore, + ); + expect(result.touchedTopics).toEqual([]); + expect(result.cursor.processedOffset).toBe(2); + }); + + /** + * A3: a huge single message does not OOM the cursor scan. The cursor + * path no longer stringifies history with the global whitespace regex; + * it only does a bounded partToString().trim() on the unprocessed slice + * to detect new user content. A 5MB message must be handled without the + * old full-history .replace(/\s+/g) blow-up. + */ + it('handles a huge single message without OOM in the cursor scan', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: ['user'], + touchedProjectScope: true, + touchedUserScope: false, + systemMessage: undefined, + }); + + const hugeText = 'x '.repeat(2_500_000); // ~5MB with whitespace + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [{ role: 'user', parts: [{ text: hugeText }] }], + }); + + // New user content detected → fork agent invoked, cursor advanced. + expect(runAutoMemoryExtractionByAgent).toHaveBeenCalled(); + expect(result.cursor.processedOffset).toBe(1); + }); + + /** + * A4: when the session ID changes between extracts, the cursor from the + * old session is ignored, and the full history is treated as unprocessed. + */ + it('reprocesses full history when session changes', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: ['user'], + touchedProjectScope: true, + touchedUserScope: false, + systemMessage: undefined, + }); + + const history: Content[] = [ + { role: 'user', parts: [{ text: 'first session query' }] }, + { role: 'model', parts: [{ text: 'first session answer' }] }, + ]; + + // Session 1: cursor advances to 2 + await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + const agentCallsBefore = vi.mocked(runAutoMemoryExtractionByAgent).mock + .calls.length; + + // Session 2: cursor ignored, full history treated as unprocessed + await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-2', + config: mockConfig, + history: [...history], + }); + + // Fork agent called again (session changed, so messages are "new") + expect(vi.mocked(runAutoMemoryExtractionByAgent).mock.calls.length).toBe( + agentCallsBefore + 1, + ); + }); + + /** + * A5: verify that cursor-first ordering prevents OOM by processing only + * the unread portion. Constructs a large history where most messages + * have already been processed, then verifies that the extract completes + * without processing the full-history text through .replace(). + */ + it('avoids full-history regex replace when most messages are already processed', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: ['user'], + touchedProjectScope: true, + touchedUserScope: false, + systemMessage: undefined, + }); + + // Build 50 messages, each with unique text to prevent string interning. + const history: Content[] = []; + for (let i = 0; i < 50; i++) { + const prefix = `[MSG${i}] `.padEnd(16, '-'); + history.push({ + role: i % 2 === 0 ? 'user' : 'model', + parts: [{ text: prefix + `${i}: the quick brown fox `.repeat(200) }], + }); + } + + // Process all 50 in the first extract + await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + // Add 2 more messages — only these need processing + history.push( + { role: 'user', parts: [{ text: 'final question?'.repeat(50) }] }, + { role: 'model', parts: [{ text: 'final answer.'.repeat(50) }] }, + ); + + const agentCallsBefore = vi.mocked(runAutoMemoryExtractionByAgent).mock + .calls.length; + + // This should complete without OOM — it only processes 2 messages, + // not the full 52. + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + expect(vi.mocked(runAutoMemoryExtractionByAgent).mock.calls.length).toBe( + agentCallsBefore + 1, + ); + expect(result.cursor.processedOffset).toBe(52); + }); + + /** + * A6: the cursor scan must not count empty or whitespace-only user + * messages as "new user content". The partToString().trim().length > 0 + * filter should cause the extract to be skipped, just like the old + * buildTranscriptMessages().filter() did. + */ + it('skips extract when unprocessed user messages are whitespace-only', async () => { + const history: Content[] = [{ role: 'user', parts: [{ text: ' ' }] }]; + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + expect(runAutoMemoryExtractionByAgent).not.toHaveBeenCalled(); + expect(result.touchedTopics).toEqual([]); + }); + + it('skips extract when unprocessed user messages have empty parts', async () => { + const history: Content[] = [{ role: 'user', parts: [] }]; + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...history], + }); + + expect(runAutoMemoryExtractionByAgent).not.toHaveBeenCalled(); + expect(result.touchedTopics).toEqual([]); + }); + + /** + * A7: when history shrinks between extract calls (e.g. compression + * reduces 50 → 17 entries), the stored processedOffset (50) exceeds + * history.length (17). The cursor-first logic must reset startOffset + * to 0 rather than passing 50 to history.slice(), which would return + * [] and permanently skip new messages. + */ + it('re-scans full history when stored offset exceeds current length (compression)', async () => { + vi.mocked(runAutoMemoryExtractionByAgent).mockResolvedValue({ + touchedTopics: ['user'], + touchedProjectScope: true, + touchedUserScope: false, + systemMessage: undefined, + }); + + // First extract: 20 messages, cursor advances to 20. + const fullHistory: Content[] = []; + for (let i = 0; i < 20; i++) { + fullHistory.push({ + role: i % 2 === 0 ? 'user' : 'model', + parts: [{ text: `compression msg ${i}` }], + }); + } + await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...fullHistory], + }); + + // Simulate compression: history shrinks from 20 to 5, but cursor + // still says processedOffset = 20. Then a new user message is added. + const compressedHistory = fullHistory.slice(0, 5); + compressedHistory.push({ + role: 'user', + parts: [{ text: 'new question after compression' }], + }); + + const agentCallsBefore = vi.mocked(runAutoMemoryExtractionByAgent).mock + .calls.length; + + const result = await runAutoMemoryExtract({ + projectRoot, + sessionId: 'session-1', + config: mockConfig, + history: [...compressedHistory], // 6 messages, cursor says 20 + }); + + // The new user message must be detected — startOffset was clamped to 0 + // instead of using the stale 20 that exceeds history.length (6). + expect(vi.mocked(runAutoMemoryExtractionByAgent).mock.calls.length).toBe( + agentCallsBefore + 1, + ); + expect(result.cursor.processedOffset).toBe(compressedHistory.length); + }); + }); }); diff --git a/packages/core/src/memory/extract.ts b/packages/core/src/memory/extract.ts index a66b0aeab0..a42db33dd2 100644 --- a/packages/core/src/memory/extract.ts +++ b/packages/core/src/memory/extract.ts @@ -31,50 +31,17 @@ import { const debugLogger = createDebugLogger('AUTO_MEMORY_EXTRACT'); -export interface AutoMemoryTranscriptMessage { - offset: number; - role: 'user' | 'model'; - text: string; -} - export interface AutoMemoryExtractResult { touchedTopics: AutoMemoryType[]; - skippedReason?: 'already_running' | 'queued' | 'memory_tool'; + skippedReason?: + | 'already_running' + | 'queued' + | 'memory_tool' + | 'memory_pressure'; systemMessage?: string; cursor: AutoMemoryExtractCursor; } -export function buildTranscriptMessages( - history: Content[], -): AutoMemoryTranscriptMessage[] { - return history - .map((message, index) => ({ - offset: index, - role: message.role, - text: partToString(message.parts ?? []) - .replace(/\s+/g, ' ') - .trim(), - })) - .filter( - (message): message is AutoMemoryTranscriptMessage => - (message.role === 'user' || message.role === 'model') && - message.text.length > 0, - ); -} - -export function loadUnprocessedTranscriptSlice( - sessionId: string, - messages: AutoMemoryTranscriptMessage[], - cursor: AutoMemoryExtractCursor, -): { messages: AutoMemoryTranscriptMessage[]; nextProcessedOffset: number } { - const startOffset = - cursor.sessionId === sessionId ? (cursor.processedOffset ?? 0) : 0; - return { - messages: messages.filter((message) => message.offset >= startOffset), - nextProcessedOffset: messages.length, - }; -} - async function readExtractCursor( projectRoot: string, ): Promise { @@ -153,26 +120,37 @@ export async function runAutoMemoryExtract(params: { ); } - const transcript = buildTranscriptMessages(params.history); - const currentCursor = await readExtractCursor(params.projectRoot); - const slice = loadUnprocessedTranscriptSlice( - params.sessionId, - transcript, - currentCursor, - ); - if (!params.config) { throw new Error( 'Managed auto-memory extraction requires config for forked-agent execution.', ); } - // Skip if no new user messages in the unprocessed slice. - const hasNewUserMessages = slice.messages.some((m) => m.role === 'user'); + // Read the cursor first, then scan only the unprocessed slice. The old + // code ran partToString().replace() over EVERY message but the resulting + // text was never read — fork agent context comes from getCacheSafeParams(). + const currentCursor = await readExtractCursor(params.projectRoot); + const rawOffset = + currentCursor.sessionId === params.sessionId + ? (currentCursor.processedOffset ?? 0) + : 0; + // History may shrink between extract calls (compression). Clamp to length + // so new messages after compression are not permanently skipped. + const startOffset = rawOffset > params.history.length ? 0 : rawOffset; + + // Skip if there are no new, non-empty user messages in the unprocessed + // slice. partToString runs only on this small slice and without the + // global whitespace regex — the .trim().length check preserves the old + // behaviour of ignoring empty-text user turns. + const hasNewUserMessages = params.history + .slice(startOffset) + .some( + (m) => m.role === 'user' && partToString(m.parts ?? []).trim().length > 0, + ); if (!hasNewUserMessages) { const cursor: AutoMemoryExtractCursor = { sessionId: params.sessionId, - processedOffset: slice.nextProcessedOffset, + processedOffset: params.history.length, updatedAt: now.toISOString(), }; await writeExtractCursor(params.projectRoot, cursor); @@ -221,7 +199,7 @@ export async function runAutoMemoryExtract(params: { const cursor: AutoMemoryExtractCursor = { sessionId: params.sessionId, - processedOffset: slice.nextProcessedOffset, + processedOffset: params.history.length, updatedAt: now.toISOString(), }; await writeExtractCursor(params.projectRoot, cursor); diff --git a/packages/core/src/memory/manager.test.ts b/packages/core/src/memory/manager.test.ts index 83c1d11413..11de7c689a 100644 --- a/packages/core/src/memory/manager.test.ts +++ b/packages/core/src/memory/manager.test.ts @@ -946,4 +946,400 @@ describe('MemoryManager', () => { }); }); }); + + // ─── #5147 regression: trailing queue + memory pressure ───────────────── + + describe('scheduleExtract #5147', () => { + /** + * B1: When an extract is already running and a new extract is queued, + * superseding the trailing request drops the old params reference (the + * old history becomes GC-eligible). Verify that only the latest params + * are retained and the trailing extract executes correctly. + */ + it('supersedes trailing queue without leaking old history refs', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + + const mgr = new MemoryManager(); + + let resolveFirst: ( + value: Awaited>, + ) => void; + let resolveTrailing: ( + value: Awaited>, + ) => void; + const firstPromise = new Promise< + Awaited> + >((r) => { + resolveFirst = r; + }); + const trailingPromise = new Promise< + Awaited> + >((r) => { + resolveTrailing = r; + }); + + // First call → starts running + vi.mocked(runAutoMemoryExtract).mockReturnValueOnce(firstPromise); + + void mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + history: [ + { role: 'user', parts: [{ text: 'first history' }] }, + { role: 'model', parts: [{ text: 'first response' }] }, + ], + }); + + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + + // Second call while first is running → queues trailing + const secondResult = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + history: [ + { role: 'user', parts: [{ text: 'second history' }] }, + { role: 'model', parts: [{ text: 'second response' }] }, + ], + }); + expect(secondResult.skippedReason).toBe('queued'); + + // Third call while first is STILL running → supersedes trailing + vi.mocked(runAutoMemoryExtract).mockReturnValueOnce(trailingPromise); + const thirdResult = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + history: [ + { role: 'user', parts: [{ text: 'third history' }] }, + { role: 'model', parts: [{ text: 'third response' }] }, + ], + }); + expect(thirdResult.skippedReason).toBe('queued'); + // Still only 1 actual extract call (first is still running) + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + + // Finish the first extract + resolveFirst!({ + touchedTopics: [], + cursor: { + sessionId: 'sess', + processedOffset: 2, + updatedAt: new Date().toISOString(), + }, + }); + // Wait for the trailing to be picked up and started + await vi.waitFor(() => { + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(2); + }); + + // Verify the trailing extract received the third call's params, + // not the second call's stale history reference. + expect(runAutoMemoryExtract).toHaveBeenLastCalledWith( + expect.objectContaining({ + history: [ + { role: 'user', parts: [{ text: 'third history' }] }, + { role: 'model', parts: [{ text: 'third response' }] }, + ], + }), + ); + + // Finish the trailing (should use third history, not second) + resolveTrailing!({ + touchedTopics: ['user'], + cursor: { + sessionId: 'sess', + processedOffset: 2, + updatedAt: new Date().toISOString(), + }, + }); + + // Drain to ensure everything settles + await mgr.drain({ timeoutMs: 500 }); + }); + + /** + * B2: extract is skipped with 'memory_pressure' when the shared + * MemoryPressureMonitor reports hard/critical pressure. The cursor is + * NOT advanced (runAutoMemoryExtract is never called), so the unread + * messages are retried on a later, lower-pressure turn. + */ + it('skips extract with memory_pressure when the monitor reports critical', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('critical'), + }), + } as Partial); + + const mgr = new MemoryManager(); + const result = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + }); + + expect(result.skippedReason).toBe('memory_pressure'); + expect(result.touchedTopics).toEqual([]); + // The cursor is deliberately NOT advanced (no processedOffset) so + // unprocessed messages are retried on a later lower-pressure turn. + expect(result.cursor.processedOffset).toBeUndefined(); + // Gate fired before invoking the real extract → cursor untouched. + expect(runAutoMemoryExtract).not.toHaveBeenCalled(); + }); + + /** + * B3: extract proceeds normally when the monitor reports normal/soft + * pressure (only hard/critical gate it). + */ + it('does not skip extract when pressure is normal', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + vi.mocked(runAutoMemoryExtract).mockResolvedValueOnce({ + touchedTopics: ['user'], + cursor: { + sessionId: 'sess', + processedOffset: 1, + updatedAt: new Date().toISOString(), + }, + }); + + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('soft'), + }), + } as Partial); + + const mgr = new MemoryManager(); + const result = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + }); + + expect(result.skippedReason).toBeUndefined(); + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + }); + + /** + * B3c: when getMemoryPressureMonitor() returns undefined, the gate + * allows extraction to proceed — the optional-chain returns undefined + * (falsy), so isUnderMemoryPressure returns false. + */ + it('does not skip extract when monitor is absent', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + vi.mocked(runAutoMemoryExtract).mockResolvedValueOnce({ + touchedTopics: ['user'], + cursor: { + sessionId: 'sess', + processedOffset: 1, + updatedAt: new Date().toISOString(), + }, + }); + + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue(undefined), + } as Partial); + + const mgr = new MemoryManager(); + const result = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + }); + + expect(result.skippedReason).toBeUndefined(); + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + }); + + /** + * B3b: 'hard' pressure level also gates extract (not just 'critical'). + * In production 'hard' is the first level to fire as memory climbs, so + * it needs the same coverage as 'critical'. + */ + it('skips extract when monitor reports hard pressure', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('hard'), + }), + } as Partial); + + const mgr = new MemoryManager(); + const result = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + }); + + expect(result.skippedReason).toBe('memory_pressure'); + expect(result.cursor.processedOffset).toBeUndefined(); + expect(runAutoMemoryExtract).not.toHaveBeenCalled(); + }); + + /** + * B4: a queued (trailing) extract is also gated. Because the gate lives + * in runExtract — the choke point both the direct and queued paths funnel + * through — a trailing extract started after pressure spikes is skipped + * rather than bypassing the gate via startQueuedExtract. + */ + it('gates queued trailing extracts under memory pressure', async () => { + vi.mocked(runAutoMemoryExtract).mockClear(); + + let pressure: 'normal' | 'critical' = 'normal'; + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn(() => pressure), + }), + } as Partial); + + let resolveFirst: ( + value: Awaited>, + ) => void; + const firstPromise = new Promise< + Awaited> + >((r) => { + resolveFirst = r; + }); + vi.mocked(runAutoMemoryExtract).mockReturnValueOnce(firstPromise); + + const mgr = new MemoryManager(); + + // First extract starts running (pressure normal). + void mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'first' }] }], + }); + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + + // Queue a trailing extract while the first is still running. + const queuedResult = await mgr.scheduleExtract({ + projectRoot: '/project', + sessionId: 'sess', + config, + history: [{ role: 'user', parts: [{ text: 'trailing' }] }], + }); + expect(queuedResult.skippedReason).toBe('queued'); + + // Pressure spikes, then the first extract finishes → trailing dequeues. + pressure = 'critical'; + resolveFirst!({ + touchedTopics: [], + cursor: { + sessionId: 'sess', + processedOffset: 1, + updatedAt: new Date().toISOString(), + }, + }); + + // The trailing extract must NOT call the real runAutoMemoryExtract a + // second time — the gate in runExtract skips it under pressure. + await mgr.drain({ timeoutMs: 500 }); + expect(runAutoMemoryExtract).toHaveBeenCalledTimes(1); + }); + + /** + * B4b: skill review pressure gate lives in runSkillReview (mirroring + * the extract pattern), producing a skipped task record. + */ + it('skips skill review when monitor reports hard pressure', async () => { + vi.mocked(runSkillReviewByAgent).mockClear(); + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('hard'), + }), + } as Partial); + + const mgr = new MemoryManager(); + const result = mgr.scheduleSkillReview({ + projectRoot: '/project', + sessionId: 'sess', + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + toolCallCount: 25, + threshold: 2, + skillsModified: false, + config, + }); + + expect(result.status).toBe('scheduled'); + const record = await result.promise!; + expect(record.status).toBe('skipped'); + expect(record.metadata?.['skippedReason']).toBe('memory_pressure'); + expect(runSkillReviewByAgent).not.toHaveBeenCalled(); + }); + + /** + * B4c: after the gate fires, the finally block must clean up the + * skillReviewInFlightByProject Map entry. A second call to + * scheduleSkillReview must NOT return already_running. + */ + it('cleans up Map entry after pressure gate fires', async () => { + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('hard'), + }), + } as Partial); + + const mgr = new MemoryManager(); + + // First call: gate fires, skipped record pushed to promise. + const first = mgr.scheduleSkillReview({ + projectRoot: '/project', + sessionId: 'sess', + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + toolCallCount: 25, + threshold: 2, + skillsModified: false, + config, + }); + expect(first.status).toBe('scheduled'); + await first.promise!; + + vi.mocked(runSkillReviewByAgent).mockClear(); + + // Second call: must not return already_running — the Map entry was + // cleaned up by the finally block. + const second = mgr.scheduleSkillReview({ + projectRoot: '/project', + sessionId: 'sess', + history: [{ role: 'user', parts: [{ text: 'hi' }] }], + toolCallCount: 25, + threshold: 2, + skillsModified: false, + config, + }); + + expect(second.status).toBe('scheduled'); + expect(second.skippedReason).toBeUndefined(); + }); + + /** + * B5: scheduleDream also gates on memory pressure. The dream path does + * its own structuredClone of full history, so hard/critical pressure + * should skip it alongside extract. + */ + it('skips dream with memory_pressure when monitor reports critical', async () => { + const config = makeMockConfig({ + getMemoryPressureMonitor: vi.fn().mockReturnValue({ + getPressureLevel: vi.fn().mockReturnValue('critical'), + }), + getManagedAutoDreamEnabled: vi.fn().mockReturnValue(true), + } as Partial); + + const mgr = new MemoryManager(); + const result = await mgr.scheduleDream({ + projectRoot: '/project', + sessionId: 'sess', + config, + }); + + expect(result.status).toBe('skipped'); + expect(result.skippedReason).toBe('memory_pressure'); + }); + }); }); diff --git a/packages/core/src/memory/manager.ts b/packages/core/src/memory/manager.ts index 95846aa479..793f23ef5a 100644 --- a/packages/core/src/memory/manager.ts +++ b/packages/core/src/memory/manager.ts @@ -147,7 +147,8 @@ export interface SkillReviewScheduleResult { | 'below_threshold' | 'skills_modified_in_session' | 'disabled' - | 'already_running'; + | 'already_running' + | 'memory_pressure'; promise?: Promise; } @@ -175,7 +176,8 @@ export interface DreamScheduleResult { | 'min_sessions' | 'scan_throttled' | 'locked' - | 'running'; + | 'running' + | 'memory_pressure'; promise?: Promise; } @@ -674,11 +676,26 @@ export class MemoryManager { return this.track(record.id, this.runExtract(record.id, params)) as never; } + /** + * True when the runtime is under hard or critical memory pressure, as + * reported by the shared MemoryPressureMonitor (#5147). The monitor is + * cgroup-aware and compares RSS/heap against their actual limits as a + * ratio, so this adapts to `--max-old-space-size`, containers, and large + * hosts alike — unlike an absolute megabyte threshold. Returns false when + * no monitor is wired (e.g. unit tests, headless), so extraction proceeds + * normally in those contexts. + */ + private isUnderMemoryPressure(config?: Config): boolean { + const level = config?.getMemoryPressureMonitor?.()?.getPressureLevel?.(); + return level === 'hard' || level === 'critical'; + } + private async runExtract( taskId: string, params: ScheduleExtractParams, ): Promise>> { const record = this.tasks.get(taskId)!; + this.extractCurrentTaskId.set(params.projectRoot, taskId); this.extractRunning.add(params.projectRoot); this.update(record, { @@ -689,6 +706,39 @@ export class MemoryManager { const t0 = Date.now(); try { + // Memory-pressure gate. Checked inside try so the finally block + // always runs — extractRunning/extractCurrentTaskId are cleaned up + // and startQueuedExtract is called regardless of the gate outcome. + if (this.isUnderMemoryPressure(params.config)) { + debugLogger.warn('Skipping extract: memory pressure too high.'); + this.update(record, { + status: 'skipped', + progressText: 'Skipped: memory pressure too high for extraction.', + metadata: { skippedReason: 'memory_pressure' }, + }); + if (params.config) { + logMemoryExtract( + params.config, + new MemoryExtractEvent({ + trigger: 'auto', + status: 'skipped', + skipped_reason: 'memory_pressure', + patches_count: 0, + touched_topics: [], + duration_ms: 0, + }), + ); + } + return { + touchedTopics: [], + skippedReason: 'memory_pressure' as const, + cursor: { + sessionId: params.sessionId, + updatedAt: (params.now ?? new Date()).toISOString(), + }, + }; + } + const result = await runAutoMemoryExtract(params); const durationMs = Date.now() - t0; this.update(record, { @@ -811,7 +861,20 @@ export class MemoryManager { params: ScheduleSkillReviewParams, ): Promise { this.skillReviewInFlightByProject.set(params.projectRoot, record.id); + try { + // Memory-pressure gate — inside try so finally always cleans up + // the skillReviewInFlightByProject entry. + if (this.isUnderMemoryPressure(params.config)) { + this.update(record, { + status: 'skipped', + progressText: 'Skipped: memory pressure too high.', + metadata: { skippedReason: 'memory_pressure' }, + }); + debugLogger.warn('Skipping skill review: memory pressure too high.'); + return record; + } + const result = await runSkillReviewByAgent({ config: params.config!, projectRoot: params.projectRoot, @@ -857,6 +920,14 @@ export class MemoryManager { return { status: 'skipped', skippedReason: 'disabled' }; } + // Also skip dream under memory pressure — dream does its own + // structuredClone of full history, and shouldn't add extra pressure + // when the heap is already under hard/critical load. + if (this.isUnderMemoryPressure(params.config)) { + debugLogger.warn('Skipping dream: memory pressure too high.'); + return { status: 'skipped', skippedReason: 'memory_pressure' }; + } + const now = params.now ?? new Date(); const minHours = params.minHoursBetweenDreams ?? DEFAULT_AUTO_DREAM_MIN_HOURS; diff --git a/packages/core/src/telemetry/types.ts b/packages/core/src/telemetry/types.ts index ed698c7ae3..64ae817eb8 100644 --- a/packages/core/src/telemetry/types.ts +++ b/packages/core/src/telemetry/types.ts @@ -1288,7 +1288,11 @@ export class MemoryExtractEvent implements BaseTelemetryEvent { /** 'auto' = triggered by session turn; 'manual' = user-initiated */ trigger: 'auto' | 'manual'; status: 'completed' | 'skipped' | 'failed'; - skipped_reason?: 'already_running' | 'queued' | 'memory_tool'; + skipped_reason?: + | 'already_running' + | 'queued' + | 'memory_tool' + | 'memory_pressure'; patches_count: number; touched_topics: string; duration_ms: number; @@ -1296,7 +1300,11 @@ export class MemoryExtractEvent implements BaseTelemetryEvent { constructor(params: { trigger: 'auto' | 'manual'; status: 'completed' | 'skipped' | 'failed'; - skipped_reason?: 'already_running' | 'queued' | 'memory_tool'; + skipped_reason?: + | 'already_running' + | 'queued' + | 'memory_tool' + | 'memory_pressure'; patches_count: number; touched_topics: string[]; duration_ms: number;