basant307/AI_Governance_Project
048
1/**2 * @license3 * Copyright 2026 Qwen Team4 * SPDX-License-Identifier: Apache-2.05 */6 7/**8 * MemoryManager — the single entry-point for all memory module operations.9 *10 * # Design11 * All background-task state (in-flight promises, per-project extraction queues,12 * per-project dream-scan timestamps, task records) is owned directly by13 * MemoryManager using plain Maps and sets. There are no separate14 * BackgroundTaskRegistry / BackgroundTaskDrainer / BackgroundTaskScheduler15 * helper classes; those abstractions are replaced by straightforward inline16 * state management inside this class.17 *18 * Public API — everything external callers need:19 * config.getMemoryManager().scheduleExtract(params)20 * config.getMemoryManager().scheduleDream(params)21 * config.getMemoryManager().recall(projectRoot, query, options)22 * config.getMemoryManager().forget(projectRoot, query, options)23 * config.getMemoryManager().getStatus(projectRoot)24 * config.getMemoryManager().drain(options?)25 * config.getMemoryManager().appendToUserMemory(userMemory, projectRoot)26 *27 * # Task records28 * Each scheduled operation is tracked as a lightweight MemoryTaskRecord.29 * These are queryable by type and projectRoot for status display.30 *31 * # Injection for tests32 * Production code uses `config.getMemoryManager()`. Tests that need isolation33 * construct `new MemoryManager()` directly.34 */35 36import * as fs from 'node:fs/promises';37import * as path from 'node:path';38import { randomUUID } from 'node:crypto';39import type { Content, Part } from '@google/genai';40import type { Config } from '../config/config.js';41import { Storage } from '../config/storage.js';42import { atomicWriteFile } from '../utils/atomicFileWrite.js';43import { createDebugLogger } from '../utils/debugLogger.js';44import {45 logMemoryDream,46 logMemoryExtract,47 MemoryDreamEvent,48 MemoryExtractEvent,49} from '../telemetry/index.js';50import { isAnyAutoMemPath, isTeamAutoMemPath } from './paths.js';51import {52 getAutoMemoryConsolidationLockPath,53 getAutoMemoryMetadataPath,54} from './paths.js';55import { ensureAutoMemoryScaffold } from './store.js';56import { runAutoMemoryExtract } from './extract.js';57import { runManagedAutoMemoryDream } from './dream.js';58import {59 forgetManagedAutoMemoryEntries,60 forgetManagedAutoMemoryMatches,61 selectManagedAutoMemoryForgetCandidates,62 type AutoMemoryForgetMatch,63 type AutoMemoryForgetResult,64 type AutoMemoryForgetSelectionResult,65} from './forget.js';66import {67 resolveRelevantAutoMemoryPromptForQuery,68 type RelevantAutoMemoryPromptResult,69 type ResolveRelevantAutoMemoryPromptOptions,70} from './recall.js';71import { getManagedAutoMemoryStatus } from './status.js';72import {73 appendManagedAutoMemoryToUserMemory,74 type BuildMemoryPromptOptions,75 type UserAutoMemorySection,76 type TeamAutoMemorySection,77} from './prompt.js';78import { writeDreamManualRunToMetadata } from './dream.js';79import { buildConsolidationTaskPrompt } from './dreamAgentPlanner.js';80import {81 runSkillReviewByAgent,82 listExistingSkillDirNames,83} from './skillReviewAgentPlanner.js';84import {85 stageSkillDirs,86 acceptPendingSkill,87 rejectPendingSkill,88 type PendingSkill,89} from './pending-skills.js';90import type { AutoMemoryMetadata } from './types.js';91 92const debugLogger = createDebugLogger('AUTO_MEMORY_MANAGER');93 94// ─── Re-export public types consumed by callers ───────────────────────────────95 96export type {97 AutoMemoryForgetResult,98 AutoMemoryForgetMatch,99 AutoMemoryForgetSelectionResult,100};101export type {102 RelevantAutoMemoryPromptResult,103 ResolveRelevantAutoMemoryPromptOptions,104};105export type { ManagedAutoMemoryStatus } from './status.js';106 107// ─── Task record ──────────────────────────────────────────────────────────────108 109export type MemoryTaskStatus =110 | 'pending'111 | 'running'112 | 'completed'113 | 'failed'114 | 'cancelled'115 | 'skipped';116 117export interface MemoryTaskRecord {118 id: string;119 taskType: 'extract' | 'dream' | 'skill-review';120 projectRoot: string;121 sessionId?: string;122 status: MemoryTaskStatus;123 createdAt: string;124 updatedAt: string;125 progressText?: string;126 error?: string;127 metadata?: Record<string, unknown>;128}129 130// ─── Extract params / result ──────────────────────────────────────────────────131 132export interface ScheduleExtractParams {133 projectRoot: string;134 sessionId: string;135 history: Content[];136 now?: Date;137 config?: Config;138}139 140export interface ScheduleSkillReviewParams {141 projectRoot: string;142 sessionId: string;143 history: Content[];144 toolCallCount: number;145 skillsModified: boolean;146 now?: Date;147 config?: Config;148 enabled?: boolean;149 threshold?: number;150 maxTurns?: number;151 timeoutMs?: number;152 /** When true, stage created skills for user confirmation instead of153 * leaving them live in the skills root. Sourced from154 * Config.getAutoSkillConfirmEnabled(). */155 confirmBeforePersist?: boolean;156}157 158export interface SkillReviewScheduleResult {159 status: 'scheduled' | 'skipped';160 taskId?: string;161 skippedReason?:162 | 'below_threshold'163 | 'skills_modified_in_session'164 | 'disabled'165 | 'already_running'166 | 'memory_pressure';167 promise?: Promise<MemoryTaskRecord>;168}169 170// AutoMemoryExtractResult is re-used as the return type171export type { AutoMemoryExtractResult as ExtractResult } from './extract.js';172 173// ─── Dream params / result ────────────────────────────────────────────────────174 175export interface ScheduleDreamParams {176 projectRoot: string;177 sessionId: string;178 config?: Config;179 now?: Date;180 minHoursBetweenDreams?: number;181 minSessionsBetweenDreams?: number;182}183 184export interface DreamScheduleResult {185 status: 'scheduled' | 'skipped';186 taskId?: string;187 skippedReason?:188 | 'disabled'189 | 'same_session'190 | 'min_hours'191 | 'min_sessions'192 | 'scan_throttled'193 | 'locked'194 | 'running'195 | 'memory_pressure';196 promise?: Promise<MemoryTaskRecord>;197}198 199/** Function type for scanning session files by mtime. Injected for testing. */200export type SessionScannerFn = (201 projectRoot: string,202 sinceMs: number,203 excludeSessionId: string,204) => Promise<string[]>;205 206// ─── Drain options ────────────────────────────────────────────────────────────207 208export interface DrainOptions {209 timeoutMs?: number;210}211 212// ─── Constants ────────────────────────────────────────────────────────────────213 214export const EXTRACT_TASK_TYPE = 'managed-auto-memory-extraction' as const;215export const DREAM_TASK_TYPE = 'managed-auto-memory-dream' as const;216export const SKILL_REVIEW_TASK_TYPE = 'managed-skill-extractor' as const;217export const AUTO_SKILL_THRESHOLD = 20;218 219export const DEFAULT_AUTO_DREAM_MIN_HOURS = 24;220export const DEFAULT_AUTO_DREAM_MIN_SESSIONS = 5;221 222const DREAM_LOCK_STALE_MS = 60 * 60 * 1000; // 1 hour223const SESSION_SCAN_INTERVAL_MS = 10 * 60 * 1000; // 10 minutes224 225const WRITE_TOOL_NAMES = new Set([226 'write_file',227 'edit',228 'replace',229 'create_file',230]);231 232// ─── Internal helpers ─────────────────────────────────────────────────────────233 234function makeTaskRecord(235 type: MemoryTaskRecord['taskType'],236 projectRoot: string,237 sessionId?: string,238): MemoryTaskRecord {239 const now = new Date().toISOString();240 return {241 id: randomUUID(),242 taskType: type,243 projectRoot,244 sessionId,245 status: 'pending',246 createdAt: now,247 updatedAt: now,248 };249}250 251// INVARIANT: mutates `record` in place. `resolvePendingSkill`'s concurrency252// safety (concurrent Keep-all/Discard-all each removing only their own entry)253// relies on this — it re-reads `record.metadata.pendingSkills` after its await254// and expects to see writes from sibling calls. A refactor to immutable255// record updates would reintroduce the "all-but-one left behind" race.256function updateRecord(257 record: MemoryTaskRecord,258 patch: Partial<259 Pick<MemoryTaskRecord, 'status' | 'progressText' | 'error' | 'metadata'>260 >,261): void {262 if (patch.status !== undefined) record.status = patch.status;263 if (patch.progressText !== undefined)264 record.progressText = patch.progressText;265 if (patch.error !== undefined) record.error = patch.error;266 if (patch.metadata !== undefined) {267 record.metadata = { ...(record.metadata ?? {}), ...patch.metadata };268 }269 record.updatedAt = new Date().toISOString();270}271 272function partWritesToMemory(part: Part, projectRoot: string): boolean {273 const name = part.functionCall?.name;274 if (name && WRITE_TOOL_NAMES.has(name)) {275 const args = part.functionCall?.args as Record<string, unknown> | undefined;276 const filePath =277 args?.['file_path'] ?? args?.['path'] ?? args?.['target_file'];278 if (279 typeof filePath === 'string' &&280 (isAnyAutoMemPath(filePath, projectRoot) ||281 isTeamAutoMemPath(filePath, projectRoot))282 ) {283 return true;284 }285 }286 return false;287}288 289function historyWritesToMemory(290 history: Content[],291 projectRoot: string,292): boolean {293 return history.some((msg) =>294 (msg.parts ?? []).some((p) => partWritesToMemory(p, projectRoot)),295 );296}297 298function isProcessRunning(pid: number): boolean {299 try {300 process.kill(pid, 0);301 return true;302 } catch {303 return false;304 }305}306 307async function readDreamMetadata(308 projectRoot: string,309): Promise<AutoMemoryMetadata> {310 const content = await fs.readFile(311 getAutoMemoryMetadataPath(projectRoot),312 'utf-8',313 );314 return JSON.parse(content) as AutoMemoryMetadata;315}316 317async function writeDreamMetadata(318 projectRoot: string,319 metadata: AutoMemoryMetadata,320): Promise<void> {321 await atomicWriteFile(322 getAutoMemoryMetadataPath(projectRoot),323 `${JSON.stringify(metadata, null, 2)}\n`,324 { encoding: 'utf-8' },325 );326}327 328function hoursSince(lastDreamAt: string | undefined, now: Date): number | null {329 if (!lastDreamAt) return null;330 const timestamp = Date.parse(lastDreamAt);331 if (Number.isNaN(timestamp)) return null;332 return (now.getTime() - timestamp) / (1000 * 60 * 60);333}334 335const SESSION_FILE_PATTERN = /^[0-9a-fA-F-]{32,36}\.jsonl$/;336 337async function defaultSessionScanner(338 projectRoot: string,339 sinceMs: number,340 excludeSessionId: string,341): Promise<string[]> {342 const chatsDir = path.join(new Storage(projectRoot).getProjectDir(), 'chats');343 let names: string[];344 try {345 names = await fs.readdir(chatsDir);346 } catch {347 return [];348 }349 const results: string[] = [];350 await Promise.all(351 names.map(async (name) => {352 if (!SESSION_FILE_PATTERN.test(name)) return;353 const sessionId = name.slice(0, -'.jsonl'.length);354 if (sessionId === excludeSessionId) return;355 try {356 const stats = await fs.stat(path.join(chatsDir, name));357 if (stats.mtimeMs > sinceMs) results.push(sessionId);358 } catch {359 // skip unreadable files360 }361 }),362 );363 return results;364}365 366async function dreamLockExists(projectRoot: string): Promise<boolean> {367 const lockPath = getAutoMemoryConsolidationLockPath(projectRoot);368 let mtimeMs: number;369 let holderPid: number | undefined;370 try {371 const [stats, content] = await Promise.all([372 fs.stat(lockPath),373 fs.readFile(lockPath, 'utf-8').catch(() => ''),374 ]);375 mtimeMs = stats.mtimeMs;376 const parsed = parseInt(content.trim(), 10);377 holderPid = Number.isFinite(parsed) && parsed > 0 ? parsed : undefined;378 } catch {379 return false; // ENOENT — no lock380 }381 const ageMs = Date.now() - mtimeMs;382 if (ageMs <= DREAM_LOCK_STALE_MS) {383 if (holderPid !== undefined && isProcessRunning(holderPid)) return true;384 await fs.rm(lockPath, { force: true });385 return false;386 }387 await fs.rm(lockPath, { force: true });388 return false;389}390 391async function acquireDreamLock(projectRoot: string): Promise<void> {392 await fs.writeFile(393 getAutoMemoryConsolidationLockPath(projectRoot),394 String(process.pid),395 { flag: 'wx' },396 );397}398 399async function releaseDreamLock(projectRoot: string): Promise<void> {400 await fs.rm(getAutoMemoryConsolidationLockPath(projectRoot), {401 force: true,402 });403}404 405// ─── MemoryManager ────────────────────────────────────────────────────────────406 407/**408 * MemoryManager owns all runtime state for the memory subsystem and exposes a409 * clean, stable API. It is created once per Config instance and returned by410 * `config.getMemoryManager()`. Tests pass a fresh `new MemoryManager()`.411 */412export class MemoryManager {413 // ── Task records ────────────────────────────────────────────────────────────414 private readonly tasks = new Map<string, MemoryTaskRecord>();415 // ── Subscribers (useSyncExternalStore / custom listeners) ────────────────416 // Subscribers without a taskType filter receive every notify; those417 // with a filter receive only notifies whose changed record matches418 // (extract OR dream). Filtered subscribers exist so high-frequency419 // consumers (e.g. the bg-tasks UI hook, which only cares about420 // dream) can skip the per-extract O(n) work that would otherwise421 // run on every UserQuery.422 private readonly subscribers = new Set<() => void>();423 private readonly subscribersByType = new Map<424 'extract' | 'dream' | 'skill-review',425 Set<() => void>426 >();427 // ── In-flight promises (for drain) ──────────────────────────────────────────428 private readonly inFlight = new Map<string, Promise<unknown>>();429 430 // ── Extract scheduling state ─────────────────────────────────────────────────431 private readonly extractRunning = new Set<string>();432 private readonly extractCurrentTaskId = new Map<string, string>();433 private readonly extractQueued = new Map<434 string,435 { taskId: string; params: ScheduleExtractParams }436 >();437 438 // ── Skill-review in-flight dedup ─────────────────────────────────────────────439 private readonly skillReviewInFlightByProject = new Map<string, string>();440 441 // ── Dream scheduling state ───────────────────────────────────────────────────442 private readonly dreamInFlightByKey = new Map<string, string>();443 private readonly dreamLastSessionScanAt = new Map<string, number>();444 // AbortControllers for in-flight dream tasks, keyed by record id.445 // cancelTask() looks up the controller, aborts it (the abort signal446 // propagates into runForkedAgent), and marks the record cancelled.447 // The runDream finally block clears the entry on settle.448 private readonly dreamAbortControllers = new Map<string, AbortController>();449 // Set to true when releaseDreamLock() throws (e.g., Windows EPERM,450 // ENOENT race, disk full). The lock file is then left on disk and451 // dreamLockExists() sees a fresh-mtime lock owned by a still-alive452 // PID (us!), suppressing every subsequent scheduleDream() call as453 // `{status: 'skipped', skippedReason: 'locked'}` — invisible to the454 // user once the surfacing UI just shows "Lock release failed" without455 // re-firing. Setting this flag tells the next scheduleDream() to456 // force-clean the leaked lock file before the existence check, so457 // scheduling resumes within the same session instead of waiting for458 // next session start's staleness sweep.459 private dreamLockReleaseFailed = false;460 private readonly sessionScanner: SessionScannerFn;461 462 constructor(sessionScanner: SessionScannerFn = defaultSessionScanner) {463 this.sessionScanner = sessionScanner;464 }465 // ─── Subscribe ───────────────────────────────────────────────────────────────────466 467 /**468 * Register a listener that is called whenever any task record changes.469 * Compatible with React’s `useSyncExternalStore`.470 * Returns an unsubscribe function.471 *472 * Pass `{ taskType: 'dream' }` (or `'extract'`) to receive only473 * notifies whose changed record matches that type. Filtered474 * subscribers skip the wakeup entirely for unrelated transitions —475 * the dream-only UI hook uses this to avoid doing O(n) signature476 * work on every per-UserQuery extract notify.477 */478 subscribe(479 listener: () => void,480 opts?: { taskType?: 'extract' | 'dream' | 'skill-review' },481 ): () => void {482 if (opts?.taskType) {483 const type = opts.taskType;484 let set = this.subscribersByType.get(type);485 if (!set) {486 set = new Set();487 this.subscribersByType.set(type, set);488 }489 set.add(listener);490 return () => {491 set!.delete(listener);492 // Drop the Map entry when the per-type bucket is empty so the493 // long-lived MemoryManager doesn't accumulate empty Sets across494 // repeated subscribe/unsubscribe cycles (e.g. React mount /495 // unmount in the bg-tasks UI hook).496 if (set!.size === 0) this.subscribersByType.delete(type);497 };498 }499 this.subscribers.add(listener);500 return () => this.subscribers.delete(listener);501 }502 503 /**504 * Notify subscribers. Pass the changed task's type so type-filtered505 * subscribers can be reached too; the unfiltered subscriber set506 * always receives the wakeup either way.507 */508 private notify(taskType?: 'extract' | 'dream' | 'skill-review'): void {509 for (const fn of this.subscribers) fn();510 if (taskType) {511 const typed = this.subscribersByType.get(taskType);512 if (typed) for (const fn of typed) fn();513 }514 }515 516 /** Update a record and notify subscribers. */517 private update(518 record: MemoryTaskRecord,519 patch: Partial<520 Pick<MemoryTaskRecord, 'status' | 'progressText' | 'error' | 'metadata'>521 >,522 ): void {523 updateRecord(record, patch);524 this.notify(record.taskType);525 }526 527 /**528 * Register a brand-new record in the task map and notify once.529 * Use this for records that start in 'pending' and need no immediate patch.530 */531 private store(record: MemoryTaskRecord): void {532 this.tasks.set(record.id, record);533 this.notify(record.taskType);534 }535 536 /**537 * Register a brand-new record AND apply an initial status patch in a single538 * notify. Avoids the double-render that separate store()+update() causes.539 */540 private storeWith(541 record: MemoryTaskRecord,542 patch: Partial<543 Pick<MemoryTaskRecord, 'status' | 'progressText' | 'error' | 'metadata'>544 >,545 ): void {546 updateRecord(record, patch);547 this.tasks.set(record.id, record);548 this.notify(record.taskType);549 }550 // ─── Task record query ────────────────────────────────────────────────────────551 552 /** Return task records filtered by type and optionally by projectRoot. */553 listTasksByType(554 taskType: MemoryTaskRecord['taskType'],555 projectRoot?: string,556 ): MemoryTaskRecord[] {557 return [...this.tasks.values()]558 .filter(559 (t) =>560 t.taskType === taskType &&561 (!projectRoot || t.projectRoot === projectRoot),562 )563 .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));564 }565 566 // ─── Drain ────────────────────────────────────────────────────────────────────567 568 /** Wait for all in-flight tasks to settle, with optional timeout. */569 async drain(options: DrainOptions = {}): Promise<boolean> {570 const promises = [...this.inFlight.values()];571 if (promises.length === 0) return true;572 const waitAll = Promise.allSettled(promises).then(() => true);573 if (!options.timeoutMs || options.timeoutMs <= 0) return waitAll;574 return Promise.race<boolean>([575 waitAll,576 new Promise<boolean>((resolve) =>577 setTimeout(() => resolve(false), options.timeoutMs),578 ),579 ]);580 }581 582 private track<T>(taskId: string, promise: Promise<T>): Promise<T> {583 this.inFlight.set(taskId, promise);584 void promise.finally(() => this.inFlight.delete(taskId));585 return promise;586 }587 588 // ─── Extract ──────────────────────────────────────────────────────────────────589 590 /**591 * Schedule a managed auto-memory extraction for the given session turn.592 *593 * Returns immediately with a skipped result if:594 * - The last history turn wrote to a memory file (memory_tool)595 * - Extraction is already running for this project (queues trailing request)596 *597 * The trailing request starts automatically when the active extraction598 * completes.599 */600 async scheduleExtract(601 params: ScheduleExtractParams,602 ): Promise<603 ReturnType<typeof runAutoMemoryExtract> extends Promise<infer T> ? T : never604 > {605 if (historyWritesToMemory(params.history, params.projectRoot)) {606 const record = makeTaskRecord(607 'extract',608 params.projectRoot,609 params.sessionId,610 );611 this.storeWith(record, {612 status: 'skipped',613 progressText: 'Skipped: main agent wrote to memory files this turn.',614 metadata: {615 skippedReason: 'memory_tool',616 historyLength: params.history.length,617 },618 });619 return {620 touchedTopics: [],621 skippedReason: 'memory_tool' as const,622 cursor: {623 sessionId: params.sessionId,624 updatedAt: (params.now ?? new Date()).toISOString(),625 },626 } as never;627 }628 629 if (this.extractRunning.has(params.projectRoot)) {630 const currentTaskId = this.extractCurrentTaskId.get(params.projectRoot);631 if (!currentTaskId) {632 return {633 touchedTopics: [],634 skippedReason: 'already_running' as const,635 cursor: {636 sessionId: params.sessionId,637 updatedAt: (params.now ?? new Date()).toISOString(),638 },639 } as never;640 }641 642 const queued = this.extractQueued.get(params.projectRoot);643 if (queued) {644 // Supersede the existing queued request with newer params645 queued.params = params;646 const queuedRecord = this.tasks.get(queued.taskId);647 if (queuedRecord) {648 this.update(queuedRecord, {649 status: 'pending',650 progressText:651 'Updated trailing managed auto-memory extraction request while another extraction is running.',652 metadata: {653 queuedBehindTaskId: currentTaskId,654 historyLength: params.history.length,655 supersededAt: new Date().toISOString(),656 },657 });658 }659 } else {660 const record = makeTaskRecord(661 'extract',662 params.projectRoot,663 params.sessionId,664 );665 this.storeWith(record, {666 status: 'pending',667 progressText:668 'Queued trailing managed auto-memory extraction until the active extraction completes.',669 metadata: {670 trailing: true,671 queuedBehindTaskId: currentTaskId,672 historyLength: params.history.length,673 },674 });675 this.extractQueued.set(params.projectRoot, {676 taskId: record.id,677 params,678 });679 }680 681 return {682 touchedTopics: [],683 skippedReason: 'queued' as const,684 cursor: {685 sessionId: params.sessionId,686 updatedAt: (params.now ?? new Date()).toISOString(),687 },688 } as never;689 }690 691 const record = makeTaskRecord(692 'extract',693 params.projectRoot,694 params.sessionId,695 );696 this.store(record);697 return this.track(record.id, this.runExtract(record.id, params)) as never;698 }699 700 /**701 * True when the runtime is under hard or critical memory pressure, as702 * reported by the shared MemoryPressureMonitor (#5147). The monitor is703 * cgroup-aware and compares RSS/heap against their actual limits as a704 * ratio, so this adapts to `--max-old-space-size`, containers, and large705 * hosts alike — unlike an absolute megabyte threshold. Returns false when706 * no monitor is wired (e.g. unit tests, headless), so extraction proceeds707 * normally in those contexts.708 */709 private isUnderMemoryPressure(config?: Config): boolean {710 const level = config?.getMemoryPressureMonitor?.()?.getPressureLevel?.();711 return level === 'hard' || level === 'critical';712 }713 714 private async runExtract(715 taskId: string,716 params: ScheduleExtractParams,717 ): Promise<Awaited<ReturnType<typeof runAutoMemoryExtract>>> {718 const record = this.tasks.get(taskId)!;719 720 this.extractCurrentTaskId.set(params.projectRoot, taskId);721 this.extractRunning.add(params.projectRoot);722 this.update(record, {723 status: 'running',724 progressText: 'Running managed auto-memory extraction.',725 metadata: { historyLength: params.history.length },726 });727 728 const t0 = Date.now();729 try {730 // Memory-pressure gate. Checked inside try so the finally block731 // always runs — extractRunning/extractCurrentTaskId are cleaned up732 // and startQueuedExtract is called regardless of the gate outcome.733 if (this.isUnderMemoryPressure(params.config)) {734 debugLogger.warn('Skipping extract: memory pressure too high.');735 this.update(record, {736 status: 'skipped',737 progressText: 'Skipped: memory pressure too high for extraction.',738 metadata: { skippedReason: 'memory_pressure' },739 });740 if (params.config) {741 logMemoryExtract(742 params.config,743 new MemoryExtractEvent({744 trigger: 'auto',745 status: 'skipped',746 skipped_reason: 'memory_pressure',747 patches_count: 0,748 touched_topics: [],749 duration_ms: 0,750 }),751 );752 }753 return {754 touchedTopics: [],755 skippedReason: 'memory_pressure' as const,756 cursor: {757 sessionId: params.sessionId,758 updatedAt: (params.now ?? new Date()).toISOString(),759 },760 };761 }762 763 const result = await runAutoMemoryExtract(params);764 const durationMs = Date.now() - t0;765 this.update(record, {766 status: result.skippedReason ? 'skipped' : 'completed',767 progressText:768 result.systemMessage ??769 (result.touchedTopics.length > 0770 ? `Managed auto-memory updated: ${result.touchedTopics.join(', ')}.`771 : 'Managed auto-memory extraction completed without durable changes.'),772 metadata: {773 touchedTopics: result.touchedTopics,774 processedOffset: result.cursor.processedOffset,775 skippedReason: result.skippedReason,776 },777 });778 if (params.config) {779 logMemoryExtract(780 params.config,781 new MemoryExtractEvent({782 trigger: 'auto',783 status: 'completed',784 patches_count: result.touchedTopics.length,785 touched_topics: result.touchedTopics,786 duration_ms: durationMs,787 }),788 );789 }790 return result;791 } catch (error) {792 const durationMs = Date.now() - t0;793 this.update(record, {794 status: 'failed',795 error: error instanceof Error ? error.message : String(error),796 });797 if (params.config) {798 logMemoryExtract(799 params.config,800 new MemoryExtractEvent({801 trigger: 'auto',802 status: 'failed',803 patches_count: 0,804 touched_topics: [],805 duration_ms: durationMs,806 }),807 );808 }809 throw error;810 } finally {811 this.extractCurrentTaskId.delete(params.projectRoot);812 this.extractRunning.delete(params.projectRoot);813 void this.startQueuedExtract(params.projectRoot);814 }815 }816 817 private async startQueuedExtract(projectRoot: string): Promise<void> {818 if (this.extractRunning.has(projectRoot)) return;819 const queued = this.extractQueued.get(projectRoot);820 if (!queued) return;821 this.extractQueued.delete(projectRoot);822 await this.track(823 queued.taskId,824 this.runExtract(queued.taskId, queued.params),825 );826 }827 828 // ─── Skill review ─────────────────────────────────────────────────────────────829 830 scheduleSkillReview(831 params: ScheduleSkillReviewParams,832 ): SkillReviewScheduleResult {833 if (params.enabled === false) {834 return { status: 'skipped', skippedReason: 'disabled' };835 }836 837 if (params.skillsModified) {838 return { status: 'skipped', skippedReason: 'skills_modified_in_session' };839 }840 841 const threshold = params.threshold ?? AUTO_SKILL_THRESHOLD;842 if (params.toolCallCount < threshold) {843 return { status: 'skipped', skippedReason: 'below_threshold' };844 }845 846 if (!params.config) {847 return { status: 'skipped', skippedReason: 'disabled' };848 }849 850 const existingTaskId = this.skillReviewInFlightByProject.get(851 params.projectRoot,852 );853 if (existingTaskId) {854 return {855 status: 'skipped',856 skippedReason: 'already_running',857 taskId: existingTaskId,858 };859 }860 861 const record = makeTaskRecord(862 'skill-review',863 params.projectRoot,864 params.sessionId,865 );866 this.storeWith(record, {867 status: 'running',868 progressText: 'Running managed skill review.',869 metadata: {870 historyLength: params.history.length,871 toolCallCount: params.toolCallCount,872 threshold,873 },874 });875 876 const promise = this.track(record.id, this.runSkillReview(record, params));877 return { status: 'scheduled', taskId: record.id, promise };878 }879 880 private async runSkillReview(881 record: MemoryTaskRecord,882 params: ScheduleSkillReviewParams,883 ): Promise<MemoryTaskRecord> {884 this.skillReviewInFlightByProject.set(params.projectRoot, record.id);885 886 try {887 // Memory-pressure gate — inside try so finally always cleans up888 // the skillReviewInFlightByProject entry.889 if (this.isUnderMemoryPressure(params.config)) {890 this.update(record, {891 status: 'skipped',892 progressText: 'Skipped: memory pressure too high.',893 metadata: { skippedReason: 'memory_pressure' },894 });895 debugLogger.warn('Skipping skill review: memory pressure too high.');896 return record;897 }898 899 // Snapshot existing skill dirs BEFORE the agent runs so staging can tell900 // newly-created skills from in-place edits of already-confirmed ones901 // (only new skills should enter the confirmation flow).902 const preExistingSkillDirs = params.confirmBeforePersist903 ? new Set(await listExistingSkillDirNames(params.projectRoot))904 : undefined;905 906 const result = await runSkillReviewByAgent({907 config: params.config!,908 projectRoot: params.projectRoot,909 history: params.history,910 maxTurns: params.maxTurns,911 timeoutMs: params.timeoutMs,912 });913 914 if (params.confirmBeforePersist && result.touchedSkillFiles.length > 0) {915 const pending = await stageSkillDirs(916 result.touchedSkillFiles,917 params.projectRoot,918 preExistingSkillDirs,919 record.id,920 );921 this.update(record, {922 status: 'completed',923 progressText:924 pending.length > 0925 ? `${pending.length} skill(s) awaiting review.`926 : (result.systemMessage ??927 'Managed skill review completed without durable changes.'),928 metadata: {929 touchedSkillFiles: result.touchedSkillFiles,930 ...(pending.length > 0 ? { pendingSkills: pending } : {}),931 },932 });933 } else {934 this.update(record, {935 status: 'completed',936 progressText:937 result.systemMessage ??938 'Managed skill review completed without durable changes.',939 metadata: { touchedSkillFiles: result.touchedSkillFiles },940 });941 }942 } catch (error) {943 this.update(record, {944 status: 'failed',945 error: error instanceof Error ? error.message : String(error),946 });947 throw error;948 } finally {949 this.skillReviewInFlightByProject.delete(params.projectRoot);950 }951 return record;952 }953 954 // ─── Dream ────────────────────────────────────────────────────────────────────955 956 /**957 * Maybe schedule a managed auto-memory dream (consolidation).958 * Returns immediately if preconditions aren't met (time gate, session count,959 * lock, or duplicate).960 */961 async scheduleDream(962 params: ScheduleDreamParams,963 ): Promise<DreamScheduleResult> {964 // `params.config` is optional only because some test paths omit it;965 // production callers always pass it. Without a config the966 // fork-agent execution can't start (`runManagedAutoMemoryDream`967 // throws). Skip early so a missing-config call doesn't surface a968 // failed dream entry in the bg-tasks dialog.969 if (!params.config || !params.config.getManagedAutoDreamEnabled()) {970 return { status: 'skipped', skippedReason: 'disabled' };971 }972 973 // Also skip dream under memory pressure — dream does its own974 // structuredClone of full history, and shouldn't add extra pressure975 // when the heap is already under hard/critical load.976 if (this.isUnderMemoryPressure(params.config)) {977 debugLogger.warn('Skipping dream: memory pressure too high.');978 return { status: 'skipped', skippedReason: 'memory_pressure' };979 }980 981 const now = params.now ?? new Date();982 const minHours =983 params.minHoursBetweenDreams ?? DEFAULT_AUTO_DREAM_MIN_HOURS;984 const minSessions =985 params.minSessionsBetweenDreams ?? DEFAULT_AUTO_DREAM_MIN_SESSIONS;986 987 await ensureAutoMemoryScaffold(params.projectRoot, now);988 const metadata = await readDreamMetadata(params.projectRoot);989 990 if (metadata.lastDreamSessionId === params.sessionId) {991 return { status: 'skipped', skippedReason: 'same_session' };992 }993 994 const elapsedHours = hoursSince(metadata.lastDreamAt, now);995 if (elapsedHours !== null && elapsedHours < minHours) {996 return { status: 'skipped', skippedReason: 'min_hours' };997 }998 999 // Throttle the expensive session-count filesystem scan.1000 // Return a distinct reason so callers can tell the difference between1001 // "we know there aren't enough sessions" and "we haven't checked yet".1002 const lastScan = this.dreamLastSessionScanAt.get(params.projectRoot) ?? 0;1003 if (now.getTime() - lastScan < SESSION_SCAN_INTERVAL_MS) {1004 return { status: 'skipped', skippedReason: 'scan_throttled' };1005 }1006 1007 const lastDreamMs = metadata.lastDreamAt1008 ? Date.parse(metadata.lastDreamAt)1009 : 0;1010 const sessionIds = await this.sessionScanner(1011 params.projectRoot,1012 lastDreamMs,1013 params.sessionId,1014 );1015 // Record scan time only after we actually performed the filesystem scan.1016 this.dreamLastSessionScanAt.set(params.projectRoot, now.getTime());1017 if (sessionIds.length < minSessions) {1018 return { status: 'skipped', skippedReason: 'min_sessions' };1019 }1020 1021 // If the previous dream's release failed (lockReleaseError surfaced1022 // on the dialog), the lock file is still on disk and dreamLockExists()1023 // would silently suppress every subsequent dream until next process1024 // start. Force-clean it here so the same session recovers.1025 if (this.dreamLockReleaseFailed) {1026 await fs1027 .rm(getAutoMemoryConsolidationLockPath(params.projectRoot), {1028 force: true,1029 })1030 .catch(() => {1031 // Best-effort recovery — if even the forced rm fails (truly1032 // unrecoverable filesystem state), fall through and let the1033 // existence check below report 'locked' as before.1034 });1035 this.dreamLockReleaseFailed = false;1036 }1037 if (await dreamLockExists(params.projectRoot)) {1038 return { status: 'skipped', skippedReason: 'locked' };1039 }1040 1041 // Deduplication — only one dream per projectRoot at a time1042 const dedupeKey = `${DREAM_TASK_TYPE}:${params.projectRoot}`;1043 const existingId = this.dreamInFlightByKey.get(dedupeKey);1044 if (existingId) {1045 return {1046 status: 'skipped',1047 skippedReason: 'running',1048 taskId: existingId,1049 };1050 }1051 1052 const record = makeTaskRecord(1053 'dream',1054 params.projectRoot,1055 params.sessionId,1056 );1057 // Register the AbortController BEFORE storeWith. storeWith fires1058 // a notify which can synchronously call cancelTask via subscribers1059 // (e.g. a UI listener). If the controller isn't in1060 // `dreamAbortControllers` by then, cancelTask falls into the1061 // missing-controller defensive warn-and-return-false path and the1062 // model gets a phantom failure on a brand-new dream. Registering1063 // first means any reentrant cancel sees a complete state.1064 const abortController = new AbortController();1065 this.dreamAbortControllers.set(record.id, abortController);1066 this.dreamInFlightByKey.set(dedupeKey, record.id);1067 this.storeWith(record, {1068 status: 'running',1069 // Set the initial progressText so the dialog's Progress section1070 // has something to show during the in-flight window — fork-agent1071 // execution exposes no per-turn callback today, so without this1072 // the section stays empty until completion.1073 progressText: 'Scheduled managed auto-memory dream.',1074 metadata: { sessionCount: sessionIds.length },1075 });1076 1077 const promise = this.track(1078 record.id,1079 this.runDream(record, dedupeKey, params, now, abortController.signal),1080 );1081 1082 return { status: 'scheduled', taskId: record.id, promise };1083 }1084 1085 /**1086 * Look up a single task record by id. Used by `task_stop` and other1087 * cross-cutting consumers that have a task id but no project root.1088 */1089 getTask(taskId: string): MemoryTaskRecord | undefined {1090 return this.tasks.get(taskId);1091 }1092 1093 /** Promote one staged skill (by dir name) for the given skill-review task. */1094 async acceptPendingSkillFromTask(1095 taskId: string,1096 skillName: string,1097 ): Promise<void> {1098 await this.resolvePendingSkill(taskId, skillName, 'accept');1099 }1100 1101 /** Discard one staged skill (by dir name) for the given skill-review task. */1102 async rejectPendingSkillFromTask(1103 taskId: string,1104 skillName: string,1105 ): Promise<void> {1106 await this.resolvePendingSkill(taskId, skillName, 'reject');1107 }1108 1109 private async resolvePendingSkill(1110 taskId: string,1111 skillName: string,1112 action: 'accept' | 'reject',1113 ): Promise<void> {1114 const record = this.tasks.get(taskId);1115 if (!record) {1116 debugLogger.warn(`Cannot resolve pending skill: no task ${taskId}.`);1117 return;1118 }1119 const target = (1120 (record.metadata?.['pendingSkills'] as PendingSkill[]) ?? []1121 ).find((p) => p.name === skillName);1122 if (!target) {1123 debugLogger.warn(1124 `Cannot resolve pending skill "${skillName}": not pending on task ${taskId}.`,1125 );1126 return;1127 }1128 try {1129 if (action === 'accept') {1130 await acceptPendingSkill(target);1131 } else {1132 await rejectPendingSkill(target);1133 }1134 } catch (error) {1135 // Leave the skill in pendingSkills so the user can retry, and surface the1136 // failure instead of silently dropping it.1137 debugLogger.warn(1138 `Failed to ${action} pending skill "${skillName}": ${1139 error instanceof Error ? error.message : String(error)1140 }`,1141 );1142 throw error;1143 }1144 // Re-read pendingSkills AFTER the await: concurrent Keep-all/Discard-all1145 // calls each remove only their own entry. read+filter+update runs with no1146 // intervening await, so it is atomic under the single-threaded event loop.1147 const remaining = (1148 (record.metadata?.['pendingSkills'] as PendingSkill[]) ?? []1149 ).filter((p) => p.name !== skillName);1150 this.update(record, { metadata: { pendingSkills: remaining } });1151 }1152 1153 /**1154 * Cancel a running dream task. Aborts the dream's fork agent (the1155 * abort signal threads through `runForkedAgent`), marks the record1156 * cancelled immediately so the UI reflects user intent, and lets the1157 * existing `runDream` finally block release the consolidation lock1158 * via the natural error propagation path.1159 *1160 * Returns true if a running task was aborted, false if the task is1161 * unknown / already terminal / not a dream. Currently only dream1162 * tasks support cancellation — extract is short-lived and runs1163 * synchronously through the request loop; cancelling it would1164 * interfere with the user's own turn.1165 */1166 cancelTask(taskId: string): boolean {1167 const record = this.tasks.get(taskId);1168 if (!record) return false;1169 if (record.taskType !== 'dream') return false;1170 if (record.status !== 'running') return false;1171 1172 // The AbortController is registered synchronously alongside the1173 // status='running' transition in scheduleDream and only cleared in1174 // runDream's finally block (which only runs after a terminal1175 // status transition has already happened). So under normal flow1176 // an entry that is `running` MUST have a controller. Treat the1177 // missing-controller case as a contract violation: don't flip1178 // status (a cancelled record without an aborted fork would leak1179 // the consolidation lock until the agent finishes naturally) and1180 // return false so the caller knows the abort didn't take. Log at1181 // warn level so the inconsistency is observable in debug bundles1182 // — silent failure here would leave a runaway dream burning tokens1183 // with no signal to the user or to telemetry.1184 const ac = this.dreamAbortControllers.get(taskId);1185 if (!ac) {1186 debugLogger.warn(1187 `cancelTask: AbortController missing for running dream task ${taskId}; ` +1188 `not flipping status. This indicates a logic bug — the controller ` +1189 `should have been registered in scheduleDream and only cleared ` +1190 `after a terminal status transition.`,1191 );1192 return false;1193 }1194 1195 // Mark cancelled BEFORE aborting so the runDream catch path can1196 // detect the user-cancel intent (signal.aborted + status already1197 // 'cancelled') and avoid overwriting with a generic 'failed'.1198 this.update(record, {1199 status: 'cancelled',1200 progressText: 'Cancelled by user.',