diff --git a/apps/cli/src/lib/AGENTS.md b/apps/cli/src/lib/AGENTS.md index 1eb42195d..2c316817f 100644 --- a/apps/cli/src/lib/AGENTS.md +++ b/apps/cli/src/lib/AGENTS.md @@ -162,6 +162,14 @@ control-plane path is DEPRECATED; do not add functionality to it. frames and MUST sample without a cloud transport. Cloud observers/snapshots attach only after the authorized remote bridge attaches and detach on offline/revocation. Sampling is observer-lease driven; never start OS probes permanently or persist snapshots. + Recent resource history retains only bounded observed samples in memory. The local + `machine/get-resource-history` Machine RPC reads that buffer without probing; negotiate + `resourceHistory` v1. History is workspace-scoped and excludes command lines/environment. + Per-process attribution pins Windows creation-identity strings after first observation; + coarse POSIX `lstart` timestamps cannot establish identity, so those session process rows + are omitted. Aggregate resource estimates remain separate. History schemas reject unknown + nested resource fields and unsupported session statuses. Diagnostic attribution never + authorizes process termination. - Machine Flock writes for this CLI's own machine must be local-first: after `repo.flush()`, call `LoroDocumentManager.markMachineFlockDocDirty(...)` (or pass the manager as the sync scheduler) instead of awaiting `handle.syncOnce()` in the user/RPC request path. @@ -252,6 +260,12 @@ control-plane path is DEPRECATED; do not add functionality to it. (`pressureRecheckAttempts`) because reclaim returns cache in milliseconds. Eviction is bounded per call (`maxEvictionsPerCall`) because the caller awaits it on the prompt hot path. The threshold is a safety MARGIN, never "what a turn needs" — do not phrase it that way to users. + GC eligibility captures runtime, history-mirror and metadata-version identity before awaited + reads, then checks again immediately before termination. Inspection errors protect only the + affected session. Cleanup holds dispatch, execution and manager admission leases through + document teardown and transient-store deletion. + Release manager and execution first, then dispatch so deferred RPC/meta work opens fresh state. + Direct start/continue/steer must reserve execution admission before accessing session documents. - `provider-setup-manager.ts` owns durable default managed-builtin creation; setup rows with executable runtime overrides are invalid. The future config stays under `['providerSetup', configId]` while runtime/auth/live-probe @@ -301,6 +315,14 @@ control-plane path is DEPRECATED; do not add functionality to it. `tool_call` items in history — the CLI persists NO extra scheduled-task state (not in `SessionMeta`, not a new history item); see `@lody/shared` `collectPendingScheduledTasksFromHistory` + `nextCronFireMs`. + GC uses one history snapshot for active goals and background protection. Completed + scheduling tool calls protect the owning runtime until an explicit persisted cancellation; + an elapsed fire time is never proof of completion. Live pending/terminal work is checked + before and after asynchronous reads. Missing runtime ownership releases stale task-only + protection, while active goals remain persistent; a replacement runtime requires a fresh + eligibility check. If task completion/liveness is not observable for a live runtime, + preserve it conservatively instead of inventing a TTL. A failed history read protects only + its session and must not abort the sweep or become an unhandled interval rejection. INVARIANT: `history-apply.ts` strips `rawInput`/`rawOutput` from ALL generic tool calls (unstructured by spec) EXCEPT the four scheduling tools in `SCHEDULING_TOOL_NAMES` (`CronCreate/CronDelete/CronList/ScheduleWakeup`, matched via `_meta.lody.toolName`), diff --git a/apps/cli/src/lib/loro/doc.ts b/apps/cli/src/lib/loro/doc.ts index 9d6a9ce7f..444ed925a 100644 --- a/apps/cli/src/lib/loro/doc.ts +++ b/apps/cli/src/lib/loro/doc.ts @@ -1241,7 +1241,7 @@ export class LoroDocumentManager { // any one-shot renderer reconciliation without unloading the shared doc. this.cancelLocalDocRoomBridge(docId); const existing = this.sessions.get(sessionId); - if (existing) { + if (existing && !existing.isDestroyed) { return existing; } @@ -1256,6 +1256,20 @@ export class LoroDocumentManager { // open a fresh handle; otherwise the unload could evict the document that // the newly activated SessionDocument is about to retain. const initPromise = this.withLocalDocOwnership(docId, async () => { + const stale = this.sessions.get(sessionId); + if (stale) { + if (!stale.isDestroyed) return stale; + // Failed destruction may have already detached the room. Never expose + // that wrapper: retry teardown before opening a new repo handle. + await stale.destroy({ preserveStatus: true }); + if (this.sessions.get(sessionId) === stale) this.sessions.delete(sessionId); + const replacement = this.sessions.get(sessionId); + if (replacement) { + if (replacement.isDestroyed) + throw new Error('Replacement session document requires cleanup'); + return replacement; + } + } const sessionDoc = new SessionDocument( this.repo, sessionId, @@ -1671,23 +1685,31 @@ export class LoroDocumentManager { sessionId: SessionId, options: { preserveStatus?: boolean } = {} ): Promise { - // Also await any in-flight init for this session + // Await initialization separately: only init failure means nothing to clean. + // A destroy failure must propagate and leave the exact wrapper owned for retry. const pending = this.pendingSessionDocs.get(sessionId); if (pending) { + const existing = this.sessions.get(sessionId); + let doc: SessionDocument | undefined; try { - const doc = await pending; - await doc.destroy({ preserveStatus: options.preserveStatus }); - this.sessions.delete(sessionId); + doc = await pending; } catch { - // Init failed — nothing to clean up + // A recovery attempt can fail while its destroyed wrapper stays owned. + doc = existing; } - this.pendingSessionDocs.delete(sessionId); + if (doc) { + await doc.destroy({ preserveStatus: options.preserveStatus }); + if (this.sessions.get(sessionId) === doc) this.sessions.delete(sessionId); + } + if (this.pendingSessionDocs.get(sessionId) === pending) + this.pendingSessionDocs.delete(sessionId); + return; } const sessionDoc = this.sessions.get(sessionId); if (sessionDoc) { await sessionDoc.destroy({ preserveStatus: options.preserveStatus }); - this.sessions.delete(sessionId); + if (this.sessions.get(sessionId) === sessionDoc) this.sessions.delete(sessionId); } } } @@ -1750,6 +1772,7 @@ export class SessionDocument implements LoroDocument void>(); private historyAutoReadHandle: AutoMarkLatestUserHistoryAsReadHandle | null = null; private destroyed = false; + private pendingDestroy: Promise | undefined; get isDestroyed(): boolean { return this.destroyed; @@ -3061,7 +3084,20 @@ export class SessionDocument implements LoroDocument { + if (this.pendingDestroy) return this.pendingDestroy; + const pending = this.destroyOnce(options); + this.pendingDestroy = pending; + const release = () => { + if (this.pendingDestroy === pending) this.pendingDestroy = undefined; + }; + // A failed unload stays retryable, but concurrent callers must never start + // a second unload that could outlive the replacement document's creation. + void pending.then(release, release); + return pending; + } + + private async destroyOnce(options: { preserveStatus?: boolean }) { if (!this.mirror) { return; } diff --git a/apps/cli/src/lib/machine-runtime.ts b/apps/cli/src/lib/machine-runtime.ts index 04782d1cd..36e29ee23 100644 --- a/apps/cli/src/lib/machine-runtime.ts +++ b/apps/cli/src/lib/machine-runtime.ts @@ -343,6 +343,14 @@ export class MachineRuntime { async dispatchLocalMachineRpc( message: LocalMachineRpcRequestValidated ): Promise { + if (message.method === 'machine/get-resource-history') { + if (message.ownerSessionId) { + return { ok: false, error: 'Machine resource history requires workspace-level access' }; + } + return this.resourceMonitor + ? { ok: true, result: this.resourceMonitor.getHistory() } + : { ok: false, error: 'Resource monitor stopped' }; + } const handler = this.requireHandler(); return await handler.handleLocalMachineRpc(message); } @@ -358,11 +366,12 @@ export class MachineRuntime { this.gcManager = new SessionGCManager(gcConfig, { getSessionLastActivity: (sessionId) => handler.getLastActivity(sessionId), hasActiveTurn: (sessionId) => handler.hasActiveTurn(sessionId), - hasActiveGoal: async (sessionId) => await handler.hasActiveGoal(sessionId), + hasProtectedWork: async (sessionId) => await handler.hasProtectedWork(sessionId), + captureCleanupGuard: (sessionId) => handler.captureGCCleanupGuard(sessionId), hasPendingUpdates: (sessionId) => handler.hasPendingUpdates(sessionId), hasPendingUserWork: async (sessionId) => await handler.hasPendingUserWork(sessionId), isArchiveInFlight: (sessionId) => handler.isArchiveInFlight(sessionId), - cleanSession: (sessionId) => handler.cleanSessionForGC(sessionId), + cleanSession: (sessionId, isCurrent) => handler.cleanSessionForGC(sessionId, isCurrent), getSessionIds: () => handler.getTrackedSessionIds(), memoryPressure: this.options.memoryPressure, logger: this.options.logger, diff --git a/apps/cli/src/lib/message-handler.ts b/apps/cli/src/lib/message-handler.ts index 03ecdadf6..0a09c3af2 100644 --- a/apps/cli/src/lib/message-handler.ts +++ b/apps/cli/src/lib/message-handler.ts @@ -1,3 +1,4 @@ +import { hasBackgroundWorkFromHistory } from './session-background-work'; import os from 'os'; import fs from 'fs'; import path from 'path'; @@ -6551,6 +6552,8 @@ export class MessageHandler { }; switch (request.method) { + case 'machine/get-resource-history': + throw new Error('Resource history is served by the machine runtime'); case 'code-collab/get-file-index': await assertOwner(request.params.sessionId as SessionId); return await this.codeCollabV2Service.getFileIndex(request.params); @@ -9931,17 +9934,29 @@ export class MessageHandler { return hasPendingUserTurnActivation(meta); } - /** - * Persistent active goals may drive a later autonomous ACP cycle even while - * no prompt is running. They are not a live-presence signal, but evicting the - * ACP process would discard that resumable session state. - */ - async hasActiveGoal(sessionId: SessionId): Promise { + /** One history read covers goals and tasks; runtime ownership bounds task protection. */ + async hasProtectedWork(sessionId: SessionId): Promise { + const runtime = this.sessionManager.getSession(sessionId); + if (runtime?.terminalManager.hasRunningTerminals?.()) { + return true; + } const sessionDoc = await this.workspaceDocument.getOrCreateSessionDoc(sessionId); + const history = await sessionDoc.getHistory(); + const historyGoal = resolveLatestSessionGoalFromHistory(history); const meta = await sessionDoc.getMetaState(); const legacyMeta = meta as SessionLegacyMetaFields | null | undefined; - const historyGoal = resolveLatestSessionGoalFromHistory(await sessionDoc.getHistory()); - return isSessionGoalActive(historyGoal ?? legacyMeta?.latestGoal); + // Goals intentionally persist beyond runtime exit. Task snapshots do not: + // once the owning runtime is gone they cannot pin its transient state. + if (isSessionGoalActive(historyGoal ?? legacyMeta?.latestGoal)) return true; + const currentRuntime = this.sessionManager.getSession(sessionId); + if (!currentRuntime) return false; + // A replacement arrived while reading. This snapshot cannot establish that + // replacement's idleness; defer its cleanup until a fresh eligibility check. + if (currentRuntime !== runtime) return true; + return ( + currentRuntime.terminalManager.hasRunningTerminals?.() === true || + hasBackgroundWorkFromHistory(history) + ); } /** @@ -10025,27 +10040,73 @@ export class MessageHandler { * Clean all transient state for a session. * Called by GC manager when a session has been idle or evicted under memory pressure. */ - async cleanSessionForGC(sessionId: SessionId): Promise { + captureGCCleanupGuard(sessionId: SessionId): () => boolean { + const runtime = this.sessionManager.getSession(sessionId); + const sessionDoc = this.workspaceDocument.sessions.get(sessionId); + const mirror = sessionDoc?.mirror; + // Mirror.getState returns its immutable current state, not getHistory's + // normalized copy. Identity changes invalidate this eligibility read without + // another history scan or a retained per-session cache. + const state = mirror?.getState(); + const metadata = this.workspaceDocument.repo.getMeta(); + // Metadata is a separate Flock room. Its version is a fresh plain object, + // so compare sorted clock values, not the object identity or history mirror. + const metadataVersion = () => + JSON.stringify( + Object.entries(metadata.version()) + .sort(([a], [b]) => a.localeCompare(b)) + .map(([peer, clock]) => [peer, clock?.physicalTime, clock?.logicalCounter]) + ); + const version = metadataVersion(); + return () => + this.sessionManager.getSession(sessionId) === runtime && + this.workspaceDocument.sessions.get(sessionId) === sessionDoc && + sessionDoc?.mirror === mirror && + mirror?.getState() === state && + this.workspaceDocument.repo.getMeta() === metadata && + metadataVersion() === version && + !this.sessionDispatchWatcher.hasPendingDispatch(sessionId) && + !this.hasActiveTurn(sessionId) && + !this.hasPendingUpdates(sessionId) && + !this.isArchiveInFlight(sessionId) && + runtime?.terminalManager.hasRunningTerminals?.() !== true; + } + + async cleanSessionForGC(sessionId: SessionId, isCurrent: () => boolean): Promise { + if (!isCurrent()) return false; this.logger.debug(`[GC] Cleaning session ${sessionId}`); - // 1. Clear active presence - this.clearSessionActivePresence(sessionId); - await this.previewService.closeSessionPreviewForCleanup(sessionId, 'Session cleaned by GC'); - // 2. Terminate session process first — if later steps throw, the process - // is already gone and the session stays tracked for retry/cleanup. - if (this.sessionManager.hasSession(sessionId)) { - await this.sessionManager.terminateSession(sessionId, true); + // The preview close may yield to a new turn, terminal, goal or replacement. + // No await may separate this guard from terminateSession: it synchronously + // invokes Session.terminate, which closes admission before its first await. + if (!isCurrent()) return false; + const releaseDispatch = this.sessionDispatchWatcher.tryAcquireGCCleanupLease(sessionId); + if (!releaseDispatch) return false; + let releaseExecution: (() => void) | null = null; + let releaseManager: (() => void) | null = null; + try { + releaseExecution = this.executionService.tryAcquireGCCleanupLease(sessionId); + if (!releaseExecution) return false; + releaseManager = this.sessionManager.tryAcquireGCCleanupLease(sessionId); + if (!releaseManager || !isCurrent()) return false; + const termination = this.sessionManager.terminateSession(sessionId, true); + this.clearSessionActivePresence(sessionId); + await termination; + + // Dispatch, direct turn writes and replacement creation stay excluded + // across document destruction's awaits. Newly arriving RPC/meta work is + // retained by the watcher and resumes against a fresh doc after release. + await this.workspaceDocument.cleanSessionDoc(sessionId); + this.store.deleteSession(sessionId); + + this.logger.debug(`[GC] Session ${sessionId} cleaned`); + return true; + } finally { + releaseManager?.(); + releaseExecution?.(); + releaseDispatch(); } - - // 3. Clean Loro documents (main memory savings) - await this.workspaceDocument.cleanSessionDoc(sessionId); - - // 4. Drop transient tracking last — only after all cleanup succeeded, - // so getTrackedSessionIds() can still see it for retry if steps above throw. - this.store.deleteSession(sessionId); - - this.logger.debug(`[GC] Session ${sessionId} cleaned`); } } diff --git a/apps/cli/src/lib/session-background-work.test.ts b/apps/cli/src/lib/session-background-work.test.ts new file mode 100644 index 000000000..3616fe19c --- /dev/null +++ b/apps/cli/src/lib/session-background-work.test.ts @@ -0,0 +1,100 @@ +import { expect, it } from 'vitest'; +import { hasBackgroundWorkFromHistory } from './session-background-work'; + +const task = (taskId: string, status: string, taskKind?: string) => ({ + type: 'subagent_task', + taskId, + status, + ...(taskKind ? { taskKind } : {}), +}); + +it.each(['subagent', 'background', 'scheduled', undefined])('protects active %s tasks', (kind) => { + for (const status of ['pending', 'in_progress']) { + expect(hasBackgroundWorkFromHistory([{ items: [task('a', status, kind)] }])).toBe(true); + } +}); + +it.each(['completed', 'failed'])('does not pin terminal task state %s', (status) => { + expect(hasBackgroundWorkFromHistory([{ items: [task('a', status)] }])).toBe(false); +}); + +it('uses the latest snapshot per task and supports resumed agents', () => { + const history = [{ items: [task('a', 'in_progress')] }, { items: [task('a', 'completed')] }]; + expect(hasBackgroundWorkFromHistory(history)).toBe(false); + expect(hasBackgroundWorkFromHistory([...history, { items: [task('a', 'pending')] }])).toBe(true); + expect(hasBackgroundWorkFromHistory([...history, { items: [task('b', 'in_progress')] }])).toBe( + true + ); +}); + +it('ignores malformed cron and generic tool calls', () => { + expect( + hasBackgroundWorkFromHistory([ + { + items: [ + { type: 'tool_call', toolName: 'CronCreate', status: 'completed' }, + { type: 'tool_call', status: 'in_progress' }, + null, + { type: 'subagent_task' }, + ], + }, + ]) + ).toBe(false); +}); + +it('protects a completed scheduling call until explicit cron deletion', () => { + const create = { + type: 'tool_call', + toolName: 'CronCreate', + status: 'completed', + toolCallId: 'create-1', + rawInput: { cron: '* * * * *', recurring: true }, + rawOutput: 'id: job-123', + }; + const remove = { + type: 'tool_call', + toolName: 'CronDelete', + status: 'completed', + rawInput: { id: 'job-123' }, + }; + expect(hasBackgroundWorkFromHistory([{ items: [create] }])).toBe(true); + expect(hasBackgroundWorkFromHistory([{ items: [create, remove] }])).toBe(false); + expect(hasBackgroundWorkFromHistory([{ items: [create, { ...remove, status: 'failed' }] }])).toBe( + true + ); +}); + +it('protects wakeups and one-shot schedules even after their expected fire time', () => { + const schedules = [ + { + type: 'tool_call', + toolName: 'ScheduleWakeup', + status: 'completed', + recordedAtMs: 1, + rawInput: { delaySeconds: 1 }, + }, + { + type: 'tool_call', + toolName: 'CronCreate', + status: 'completed', + toolCallId: 'old-cron', + recordedAtMs: 1, + rawInput: { cron: '0 0 * * *', recurring: false }, + rawOutput: 'id: old-job\nnextFireAt: 2000-01-01T00:00:00Z', + }, + ]; + for (const schedule of schedules) { + expect(hasBackgroundWorkFromHistory([{ items: [schedule] }])).toBe(true); + expect(hasBackgroundWorkFromHistory([{ items: [{ ...schedule, status: 'failed' }] }])).toBe( + false + ); + } +}); + +it('keeps scheduled pending tasks protected regardless of historical timestamps', () => { + expect( + hasBackgroundWorkFromHistory([ + { items: [{ ...task('a', 'pending', 'scheduled'), startedAtEpochSeconds: 1 }] }, + ]) + ).toBe(true); +}); diff --git a/apps/cli/src/lib/session-background-work.ts b/apps/cli/src/lib/session-background-work.ts new file mode 100644 index 000000000..176928420 --- /dev/null +++ b/apps/cli/src/lib/session-background-work.ts @@ -0,0 +1,24 @@ +import { + collectPendingScheduledTasksFromHistory, + isRecord, + SubagentTaskPayloadSchema, + type SubagentTaskStatus, +} from '@lody/shared'; + +/** Persisted task state and schedules; elapsed fire times never prove completion. */ +export function hasBackgroundWorkFromHistory( + history: readonly { items?: readonly unknown[] }[] +): boolean { + const latest = new Map(); + for (const entry of history) { + for (const item of entry.items ?? []) { + if (!isRecord(item) || item.type !== 'subagent_task') continue; + const parsed = SubagentTaskPayloadSchema.safeParse(item); + if (parsed.success) latest.set(parsed.data.taskId, parsed.data.status); + } + } + return ( + [...latest.values()].some((status) => status === 'pending' || status === 'in_progress') || + collectPendingScheduledTasksFromHistory(history).length > 0 + ); +} diff --git a/apps/cli/src/lib/session-gc-manager.test.ts b/apps/cli/src/lib/session-gc-manager.test.ts index c442414af..1c02027fb 100644 --- a/apps/cli/src/lib/session-gc-manager.test.ts +++ b/apps/cli/src/lib/session-gc-manager.test.ts @@ -34,15 +34,17 @@ describe('SessionGCManager', () => { let sessionActivities: Map; let activeTurns: Set; let activeGoals: Set; + let backgroundWork: ReturnType; let pendingUpdates: Set; let pendingUserWork: Set; + let pendingUserWorkRead: ReturnType; let archiveInFlight: Set; let sleepCalls: number[]; beforeEach(() => { vi.useFakeTimers(); sleepCalls = []; - cleanMock = vi.fn().mockResolvedValue(undefined); + cleanMock = vi.fn().mockResolvedValue(true); loggerMock = { info: vi.fn(), debug: vi.fn(), @@ -52,8 +54,10 @@ describe('SessionGCManager', () => { sessionActivities = new Map(); activeTurns = new Set(); activeGoals = new Set(); + backgroundWork = vi.fn(async () => false); pendingUpdates = new Set(); pendingUserWork = new Set(); + pendingUserWorkRead = vi.fn(async (id: SessionId) => pendingUserWork.has(id)); archiveInFlight = new Set(); mockedGetMemoryPressureSnapshot.mockResolvedValue({ availableMemoryBytes: 4 * 1024 * 1024 * 1024, @@ -86,10 +90,12 @@ describe('SessionGCManager', () => { { getSessionLastActivity: (sessionId) => sessionActivities.get(sessionId), hasActiveTurn: (sessionId) => activeTurns.has(sessionId), - hasActiveGoal: async (sessionId) => activeGoals.has(sessionId), + hasProtectedWork: async (sessionId) => + activeGoals.has(sessionId) || (await backgroundWork(sessionId)), hasPendingUpdates: (sessionId) => pendingUpdates.has(sessionId), - hasPendingUserWork: async (sessionId) => pendingUserWork.has(sessionId), + hasPendingUserWork: pendingUserWorkRead, isArchiveInFlight: (sessionId) => archiveInFlight.has(sessionId), + captureCleanupGuard: () => () => true, cleanSession: cleanMock, getSessionIds: () => [...sessionActivities.keys()], memoryPressure: { @@ -106,6 +112,131 @@ describe('SessionGCManager', () => { ); }; + it.each(['idle', 'pressure'] as const)( + 'rechecks active work after the final %s metadata await', + async (mode) => { + const id = 'last-await' as SessionId; + sessionActivities.set(id, Date.now() - 60000); + pendingUserWorkRead.mockImplementation(async () => { + activeTurns.add(id); + return false; + }); + mockedGetMemoryPressureSnapshot.mockResolvedValue({ + availableMemoryBytes: 1, + effectiveMemoryLimitBytes: 32 * 1024 ** 3, + }); + const manager = createManager({ idleTimeoutMs: 1000 }); + if (mode === 'idle') await manager.sweep(); + else expect((await manager.evictForMemoryPressure()).evictedSessionIds).toEqual([]); + expect(cleanMock).not.toHaveBeenCalled(); + } + ); + + it('does not count a cleanup guard refusal as a pressure eviction', async () => { + sessionActivities.set('refused' as SessionId, Date.now() - 60000); + cleanMock.mockResolvedValue(false); + mockedGetMemoryPressureSnapshot.mockResolvedValue({ + availableMemoryBytes: 1, + effectiveMemoryLimitBytes: 32 * 1024 ** 3, + }); + const result = await createManager().evictForMemoryPressure(); + expect(cleanMock).toHaveBeenCalledOnce(); + expect(result.evictedSessionIds).toEqual([]); + }); + + it.each(['capture', 'metadata'] as const)( + 'isolates %s inspection failure to one session', + async (failure) => { + const broken = 'broken-read' as SessionId; + const healthy = 'healthy-read' as SessionId; + sessionActivities.set(broken, Date.now() - 60000); + sessionActivities.set(healthy, Date.now() - 60000); + const manager = createManager({ idleTimeoutMs: 1000 }); + if (failure === 'capture') { + (manager as any).deps.captureCleanupGuard = (id: SessionId) => { + if (id === broken) throw new Error('destroyed'); + return () => true; + }; + } else { + pendingUserWorkRead.mockImplementation(async (id) => { + if (id === broken) throw new Error('unreadable metadata'); + return false; + }); + } + await manager.sweep(); + expect(cleanMock.mock.calls.map(([id]) => id)).toEqual([healthy]); + } + ); + + describe('background task eviction guard', () => { + it.each(['idle', 'pressure'] as const)( + 'protects running tasks during %s eviction', + async (mode) => { + const manager = createManager({ idleTimeoutMs: 1000 }); + const id = 'background' as SessionId; + sessionActivities.set(id, Date.now() - 60000); + backgroundWork.mockResolvedValue(true); + mockedGetMemoryPressureSnapshot.mockResolvedValue({ + availableMemoryBytes: 500 * 1024 * 1024, + effectiveMemoryLimitBytes: 32 * 1024 ** 3, + }); + if (mode === 'idle') await manager.sweep(); + else await manager.evictForMemoryPressure(); + expect(cleanMock).not.toHaveBeenCalled(); + } + ); + + it('allows cleanup once background work completes', async () => { + const manager = createManager({ idleTimeoutMs: 1000 }); + const id = 'background' as SessionId; + sessionActivities.set(id, Date.now() - 60000); + backgroundWork.mockResolvedValue(true); + await manager.sweep(); + backgroundWork.mockResolvedValue(false); + await manager.sweep(); + expect(cleanMock).toHaveBeenCalledWith(id, expect.any(Function)); + }); + + it('rechecks background work before cleanup', async () => { + const manager = createManager({ idleTimeoutMs: 1000 }); + sessionActivities.set('background' as SessionId, Date.now() - 60000); + backgroundWork.mockResolvedValueOnce(false).mockResolvedValue(true); + await manager.sweep(); + expect(cleanMock).not.toHaveBeenCalled(); + }); + + it.each(['idle', 'pressure', 'periodic'] as const)( + 'protects unreadable history while continuing %s cleanup for other sessions', + async (mode) => { + const manager = createManager({ idleTimeoutMs: 1000, sweepIntervalMs: 500 }); + const unreadable = 'unreadable' as SessionId; + const eligible = 'eligible' as SessionId; + sessionActivities.set(unreadable, Date.now() - 60000); + sessionActivities.set(eligible, Date.now() - 60000); + backgroundWork.mockImplementation(async (sessionId) => { + if (sessionId === unreadable) throw new Error('history unavailable'); + return false; + }); + mockedGetMemoryPressureSnapshot.mockResolvedValue({ + availableMemoryBytes: 500 * 1024 * 1024, + effectiveMemoryLimitBytes: 32 * 1024 ** 3, + }); + try { + if (mode === 'idle') await manager.sweep(); + else if (mode === 'pressure') await manager.evictForMemoryPressure(); + else { + manager.start(); + await vi.advanceTimersByTimeAsync(600); + } + expect(cleanMock).toHaveBeenCalledTimes(1); + expect(cleanMock).toHaveBeenCalledWith(eligible, expect.any(Function)); + expect(loggerMock.warn).toHaveBeenCalledWith(expect.stringContaining('unreadable')); + } finally { + manager.stop(); + } + } + ); + }); describe('loadGCConfig', () => { it('returns default config with 20 minute timeout', () => { const config = loadGCConfig(); @@ -158,7 +289,7 @@ describe('SessionGCManager', () => { await manager.sweep(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s1); + expect(cleanMock).toHaveBeenCalledWith(s1, expect.any(Function)); }); it('does not clean sessions with pending updates', async () => { @@ -256,6 +387,7 @@ describe('SessionGCManager', () => { // Simulate the session becoming active during cleanup cleanMock.mockImplementationOnce(async () => { sessionActivities.set(s1, Date.now()); // touch the session + return true; }); await manager.sweep(); @@ -308,7 +440,7 @@ describe('SessionGCManager', () => { await manager.evictForMemoryPressure(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s1); // longest idle first + expect(cleanMock).toHaveBeenCalledWith(s1, expect.any(Function)); // longest idle first }); it('evicts multiple sessions until memory is above threshold', async () => { @@ -373,7 +505,7 @@ describe('SessionGCManager', () => { await manager.evictForMemoryPressure(s1); // exclude s1 expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s2); // s1 excluded, so s2 is evicted + expect(cleanMock).toHaveBeenCalledWith(s2, expect.any(Function)); // s1 excluded, so s2 is evicted }); it('skips sessions with pending updates', async () => { @@ -403,7 +535,7 @@ describe('SessionGCManager', () => { await manager.evictForMemoryPressure(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s2); + expect(cleanMock).toHaveBeenCalledWith(s2, expect.any(Function)); }); it('skips sessions with pending user work', async () => { @@ -433,7 +565,7 @@ describe('SessionGCManager', () => { await manager.evictForMemoryPressure(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s2); + expect(cleanMock).toHaveBeenCalledWith(s2, expect.any(Function)); }); it('skips sessions with active goals', async () => { @@ -463,7 +595,7 @@ describe('SessionGCManager', () => { await manager.evictForMemoryPressure(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s2); + expect(cleanMock).toHaveBeenCalledWith(s2, expect.any(Function)); }); it('returns stillUnderPressure when nothing eligible can be evicted', async () => { @@ -559,7 +691,7 @@ describe('SessionGCManager', () => { const result = await manager.evictForMemoryPressure(); expect(cleanMock).toHaveBeenCalledTimes(1); - expect(cleanMock).toHaveBeenCalledWith(s1); + expect(cleanMock).toHaveBeenCalledWith(s1, expect.any(Function)); expect(result.evictedSessionIds).toEqual([s1]); expect(result.stillUnderPressure).toBe(false); expect(result.pressureReason).toBeNull(); @@ -598,6 +730,24 @@ describe('SessionGCManager', () => { }); describe('start/stop', () => { + it('handles unexpected periodic sweep rejection and continues the next interval', async () => { + const manager = createManager({ sweepIntervalMs: 500 }); + const sweep = vi + .spyOn(manager, 'sweep') + .mockRejectedValueOnce(new Error('unexpected sweep failure')) + .mockResolvedValue(undefined); + try { + manager.start(); + await vi.advanceTimersByTimeAsync(1100); + expect(sweep).toHaveBeenCalledTimes(2); + expect(loggerMock.error).toHaveBeenCalledWith( + '[GC] Sweep failed: unexpected sweep failure' + ); + } finally { + manager.stop(); + } + }); + it('runs periodic sweep', async () => { const manager = createManager({ idleTimeoutMs: 1000, diff --git a/apps/cli/src/lib/session-gc-manager.ts b/apps/cli/src/lib/session-gc-manager.ts index c39ecdf76..59393eb6d 100644 --- a/apps/cli/src/lib/session-gc-manager.ts +++ b/apps/cli/src/lib/session-gc-manager.ts @@ -95,12 +95,14 @@ export interface SessionGCDeps { getSessionLastActivity: (sessionId: SessionId) => number | undefined; /** Whether the session has an active turn (prompting or finalizing) */ hasActiveTurn: (sessionId: SessionId) => boolean; - /** Whether the session has an active background goal that still needs its ACP runtime */ - hasActiveGoal: (sessionId: SessionId) => boolean | Promise; + /** Goals, live terminals and background tasks from one history snapshot. */ + hasProtectedWork: (sessionId: SessionId) => boolean | Promise; + /** Synchronous runtime/history identity check captured before awaited eligibility reads. */ + captureCleanupGuard: (sessionId: SessionId) => () => boolean; hasPendingUpdates: (sessionId: SessionId) => boolean; hasPendingUserWork: (sessionId: SessionId) => boolean | Promise; isArchiveInFlight: (sessionId: SessionId) => boolean; - cleanSession: (sessionId: SessionId) => Promise; + cleanSession: (sessionId: SessionId, isCurrent: () => boolean) => Promise; getSessionIds: () => SessionId[]; memoryPressure: MemoryPressureSnapshotSource; logger: Logger; @@ -337,7 +339,11 @@ export class SessionGCManager { `pressureSignal=${pressureSignal}, ` + `maxEvictionsPerCall=${this.config.maxEvictionsPerCall})` ); - this.sweepInterval = setInterval(() => void this.sweep(), this.config.sweepIntervalMs); + this.sweepInterval = setInterval(() => { + void this.sweep().catch((error: unknown) => { + this.deps.logger.error(`[GC] Sweep failed: ${formatErrorMessage(error)}`); + }); + }, this.config.sweepIntervalMs); } stop(): void { @@ -365,14 +371,15 @@ export class SessionGCManager { let cleaned = 0; let skipped = 0; for (const { sessionId } of candidates) { - if (!(await this.isStillEligibleForGC(sessionId))) { + const isCurrent = await this.isStillEligibleForGC(sessionId); + if (!isCurrent) { skipped++; continue; } try { - await this.deps.cleanSession(sessionId); - cleaned++; + if (await this.deps.cleanSession(sessionId, isCurrent)) cleaned++; + else skipped++; } catch (error) { this.deps.logger.error( `[GC] Failed to clean session ${sessionId}: ${formatErrorMessage(error)}` @@ -462,7 +469,8 @@ export class SessionGCManager { continue; } - if (!(await this.isEligibleForCleanup(sessionId))) { + const isCurrent = await this.isEligibleForCleanup(sessionId); + if (!isCurrent) { continue; } @@ -470,7 +478,7 @@ export class SessionGCManager { this.deps.logger.debug( `[GC] Evicting session ${sessionId} (idle ${Math.round(idleMs / 1000)}s) due to memory pressure` ); - await this.deps.cleanSession(sessionId); + if (!(await this.deps.cleanSession(sessionId, isCurrent))) continue; evictedSessionIds.push(sessionId); // Re-check memory after eviction memorySnapshot = await this.deps.memoryPressure.refresh(); @@ -661,49 +669,72 @@ export class SessionGCManager { /** * Check if a session is eligible for cleanup. * A session is NOT eligible if it has an active turn, active goal, - * pending updates, pending user work, or archive in flight. + * background work, pending updates, pending user work, or archive in flight. */ - private async isEligibleForCleanup(sessionId: SessionId): Promise { - if (this.deps.hasActiveTurn(sessionId)) { - return false; - } + private async isEligibleForCleanup(sessionId: SessionId): Promise<(() => boolean) | null> { + try { + const protectionUnchanged = this.deps.captureCleanupGuard(sessionId); + const lastActivity = this.deps.getSessionLastActivity(sessionId); + const isCurrent = () => { + try { + return ( + protectionUnchanged() && + this.deps.getSessionLastActivity(sessionId) === lastActivity && + !this.deps.hasActiveTurn(sessionId) && + !this.deps.hasPendingUpdates(sessionId) && + !this.deps.isArchiveInFlight(sessionId) + ); + } catch (error) { + this.deps.logger.warn(`[GC] Cannot recheck ${sessionId}: ${formatErrorMessage(error)}`); + return false; + } + }; + if (this.deps.hasActiveTurn(sessionId)) { + return null; + } - if (await this.deps.hasActiveGoal(sessionId)) { - return false; - } + if (await this.deps.hasProtectedWork(sessionId)) return null; - if (this.deps.hasPendingUpdates(sessionId)) { - return false; - } + if (this.deps.hasPendingUpdates(sessionId)) { + return null; + } - if (await this.deps.hasPendingUserWork(sessionId)) { - return false; - } + if (await this.deps.hasPendingUserWork(sessionId)) { + return null; + } - if (this.deps.isArchiveInFlight(sessionId)) { - return false; - } + if (this.deps.isArchiveInFlight(sessionId)) { + return null; + } - return true; + return isCurrent() ? isCurrent : null; + } catch (error) { + // Protect an unreadable session without aborting the remaining sweep. + this.deps.logger.warn( + `[GC] Cannot inspect protected work for ${sessionId}: ${formatErrorMessage(error)}` + ); + return null; + } } /** * Re-check eligibility right before cleanup to guard against races. * Also verifies the session hasn't become active since candidate selection. */ - private async isStillEligibleForGC(sessionId: SessionId): Promise { - if (!(await this.isEligibleForCleanup(sessionId))) { - return false; + private async isStillEligibleForGC(sessionId: SessionId): Promise<(() => boolean) | null> { + const isCurrent = await this.isEligibleForCleanup(sessionId); + if (!isCurrent) { + return null; } const lastActivity = this.deps.getSessionLastActivity(sessionId); if (lastActivity !== undefined) { const idleMs = Date.now() - lastActivity; if (idleMs < this.config.idleTimeoutMs) { - return false; + return null; } } - return true; + return isCurrent() ? isCurrent : null; } } diff --git a/apps/cli/src/monitor/cli-resource-monitor.test.ts b/apps/cli/src/monitor/cli-resource-monitor.test.ts new file mode 100644 index 000000000..3a1f643a0 --- /dev/null +++ b/apps/cli/src/monitor/cli-resource-monitor.test.ts @@ -0,0 +1,23 @@ +import { expect, it, vi } from 'vitest'; +import type { MachineId } from '@lody/shared'; +import { CliResourceMonitor } from './cli-resource-monitor'; + +it('reads history without probing and records unavailable samples without private errors', async () => { + const listMonitorSessions = vi.fn(async () => { + throw new Error('private diagnostics'); + }); + const monitor = new CliResourceMonitor( + 'machine' as MachineId, + { listMonitorSessions } as never, + {} as never, + { getLatest: async () => ({}) } as never, + { debug: () => {} } as never + ); + expect(monitor.getHistory().samples).toEqual([]); + expect(listMonitorSessions).not.toHaveBeenCalled(); + await expect(monitor.sample()).rejects.toThrow('private diagnostics'); + const history = monitor.getHistory(); + expect(history.samples[0]?.source).toBe('unavailable'); + expect(JSON.stringify(history)).not.toContain('private diagnostics'); + expect(listMonitorSessions).toHaveBeenCalledOnce(); +}); diff --git a/apps/cli/src/monitor/cli-resource-monitor.ts b/apps/cli/src/monitor/cli-resource-monitor.ts index 0cc746963..8398c2715 100644 --- a/apps/cli/src/monitor/cli-resource-monitor.ts +++ b/apps/cli/src/monitor/cli-resource-monitor.ts @@ -12,7 +12,10 @@ import type { SessionManager, SessionMonitorRuntimeInfo } from '@/session/sessio import type { Logger } from '@/utils/logger'; import { formatErrorMessage } from '@/utils/format-error'; import type { MemoryPressureSnapshotSource } from './memory-pressure-sampler'; -import { aggregateProcessTreeUsage } from './process-tree'; +import { aggregateProcessTreeUsage, ObservedProcessAttribution } from './process-tree'; +import { ResourceHistory } from './resource-history'; +import type { MachineResourceHistory } from '@lody/shared'; +import type { ProcessTableEntry } from './process-table'; import { logicalCpuCount, readProcessTable } from './process-table'; import { sumResourceUsage, @@ -38,6 +41,12 @@ type Baseline = { sampledAtMs: number; cpuTimeMicros: number }; export class CliResourceMonitor { private readonly instanceId = uuidv4(); + private readonly history: ResourceHistory; + private readonly attribution = new ObservedProcessAttribution(); + + getHistory(): MachineResourceHistory { + return this.history.read(); + } private readonly sessionBaselines = new Map(); private cliBaseline: Baseline | null = null; private deviceCpuBaseline: DeviceCpuTimeSample | null = null; @@ -48,9 +57,29 @@ export class CliResourceMonitor { private readonly stateSource: SessionMonitorStateSource, private readonly memoryPressure: MemoryPressureSnapshotSource, private readonly logger: Logger - ) {} + ) { + this.history = new ResourceHistory(machineId, this.instanceId); + } async sample(): Promise { + try { + return await this.sampleOnce(); + } catch (error) { + this.history.append({ + sampledAtMs: Date.now(), + source: 'unavailable', + cliControlPlane: null, + memoryKind: process.platform === 'win32' ? 'working-set-sum' : 'rss-sum', + processes: [], + sessions: [], + processesTruncated: false, + sessionsTruncated: false, + }); + throw error; + } + } + + private async sampleOnce(): Promise { const sampledAtMs = Date.now(); const updatedAtMs = getServerNow(); const cpuCount = logicalCpuCount(); @@ -58,6 +87,8 @@ export class CliResourceMonitor { const deviceCpuCores = toDeviceCpuCores(deviceCpuSample, this.deviceCpuBaseline, cpuCount); this.deviceCpuBaseline = deviceCpuSample; const warnings: string[] = []; + let historySource: 'available' | 'unavailable' | 'not-sampled' = 'not-sampled'; + let historyProcesses: ProcessTableEntry[] = []; const [sessions, memoryPressure] = await Promise.all([ this.sessionManager.listMonitorSessions(), this.memoryPressure.getLatest(), @@ -81,6 +112,8 @@ export class CliResourceMonitor { if (processTreeSessions.length > 0 || process.platform === 'darwin') { try { const processTable = await readProcessTable(); + historySource = 'available'; + historyProcesses = processTable.entries; processMemoryKind = processTable.memoryKind; warnings.push(...processTable.warnings); if (processTable.memoryKind === 'physical-footprint-sum') { @@ -98,6 +131,7 @@ export class CliResourceMonitor { })) ); } catch (error) { + historySource = 'unavailable'; warnings.push('process_table_unavailable'); this.logger.debug( `Machine monitor process-table probe failed: ${formatErrorMessage(error)}` @@ -159,7 +193,7 @@ export class CliResourceMonitor { sessions.map((session) => session.accounting.kind).filter((kind) => kind !== 'unavailable') ); - return { + const snapshot: MachineMonitorSnapshot = { kind: 'snapshot', protocolVersion: 1, machineId: this.machineId, @@ -185,6 +219,40 @@ export class CliResourceMonitor { sessionsTruncated, warnings, }; + const owners = this.attribution.assign( + historyProcesses, + processTreeSessions.map((session) => ({ + sessionId: session.sessionId, + startedAtMs: session.startedAtMs, + rootPids: session.accounting.kind === 'process-tree' ? session.accounting.rootPids : [], + })) + ); + this.history.append({ + sampledAtMs, + source: historySource, + cliControlPlane, + memoryKind: processMemoryKind, + processesTruncated: false, + sessionsTruncated, + processes: historyProcesses + .filter((entry) => entry.pid === process.pid || owners.has(entry.pid)) + .map((entry) => ({ + pid: entry.pid, + startedAtMs: entry.startedAtMs, + sessionId: owners.get(entry.pid) ?? null, + memoryBytes: entry.memoryBytes, + cpuTimeMicros: entry.cpuTimeMicros, + })), + sessions: visibleSessions.map((session) => ({ + sessionId: session.sessionId, + parentSessionId: session.parentSessionId, + status: session.status, + resource: session.resource, + cleanup: + sessions.find((runtime) => runtime.sessionId === session.sessionId)?.cleanup ?? null, + })), + }); + return snapshot; } private buildSessionSnapshot(args: { diff --git a/apps/cli/src/monitor/process-table.ts b/apps/cli/src/monitor/process-table.ts index 339e0143b..e099f6f05 100644 --- a/apps/cli/src/monitor/process-table.ts +++ b/apps/cli/src/monitor/process-table.ts @@ -10,6 +10,8 @@ export type ProcessTableEntry = { parentPid: number; processGroupId: number | null; startedAtMs: number; + /** Provider identity with finer precision than display timestamps; absent means unpinnable. */ + processIdentity?: string; cpuTimeMicros: number; memoryBytes: number; }; @@ -25,6 +27,7 @@ const WindowsProcessSchema = z.object({ ProcessId: z.coerce.number().int().nonnegative(), ParentProcessId: z.coerce.number().int().nonnegative(), CreationDateMs: z.coerce.number().finite().nonnegative(), + CreationIdentity: z.string().regex(/^\d+$/), KernelModeTime: z.coerce.number().finite().nonnegative(), UserModeTime: z.coerce.number().finite().nonnegative(), WorkingSetSize: z.coerce.number().finite().nonnegative(), @@ -126,7 +129,7 @@ async function readDarwinProcessTable(): Promise { async function readWindowsProcessTable(): Promise { const command = [ "$ProgressPreference = 'SilentlyContinue'; Get-CimInstance -ClassName Win32_Process", - 'Select-Object ProcessId,ParentProcessId,KernelModeTime,UserModeTime,WorkingSetSize,@{Name="CreationDateMs";Expression={([DateTimeOffset]$_.CreationDate).ToUnixTimeMilliseconds()}}', + 'Select-Object ProcessId,ParentProcessId,KernelModeTime,UserModeTime,WorkingSetSize,@{Name="CreationDateMs";Expression={([DateTimeOffset]$_.CreationDate).ToUnixTimeMilliseconds()}},@{Name="CreationIdentity";Expression={$_.CreationDate.ToUniversalTime().Ticks.ToString()}}', 'ConvertTo-Json -Compress', ].join(' | '); const stdout = await runProbe( @@ -146,6 +149,7 @@ async function readWindowsProcessTable(): Promise { parentPid: item.ParentProcessId, processGroupId: null, startedAtMs: item.CreationDateMs, + processIdentity: item.CreationIdentity, cpuTimeMicros: (item.KernelModeTime + item.UserModeTime) / 10, memoryBytes: item.WorkingSetSize, })), diff --git a/apps/cli/src/monitor/process-tree.ts b/apps/cli/src/monitor/process-tree.ts index 85a0b9237..2d6059821 100644 --- a/apps/cli/src/monitor/process-tree.ts +++ b/apps/cli/src/monitor/process-tree.ts @@ -65,3 +65,36 @@ function resolveOwner( } return null; } + +/** Only provider identities may pin roots; coarse ps lstart timestamps remain unassigned. */ +export class ObservedProcessAttribution { + private readonly roots = new Map(); + + assign( + entries: readonly ProcessTableEntry[], + roots: readonly (ProcessTreeRootSet & { startedAtMs: number | null })[] + ): Map { + const byPid = new Map(entries.map((entry) => [entry.pid, entry])); + const owners = new Map(); + const currentKeys = new Set(); + for (const root of roots) { + for (const pid of root.rootPids) { + const key = `${root.sessionId}:${root.startedAtMs}:${pid}`; + currentKeys.add(key); + const entry = byPid.get(pid); + if (!entry?.processIdentity) continue; + const pinned = this.roots.get(key); + if (pinned === undefined) this.roots.set(key, entry.processIdentity); + if ((pinned ?? entry.processIdentity) === entry.processIdentity) + owners.set(pid, root.sessionId); + } + } + for (const key of this.roots.keys()) if (!currentKeys.has(key)) this.roots.delete(key); + const result = new Map(); + for (const entry of entries) { + const owner = resolveOwner(entry, byPid, owners); + if (owner) result.set(entry.pid, owner); + } + return result; + } +} diff --git a/apps/cli/src/monitor/resource-history.test.ts b/apps/cli/src/monitor/resource-history.test.ts new file mode 100644 index 000000000..2b685e7f9 --- /dev/null +++ b/apps/cli/src/monitor/resource-history.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from 'vitest'; +import { MachineResourceHistorySchema } from '@lody/shared'; +import type { SessionId } from '@lody/shared'; +import { ResourceHistory } from './resource-history'; +import { ObservedProcessAttribution } from './process-tree'; + +const sample = (sampledAtMs: number) => ({ + sampledAtMs, + source: 'available' as const, + cliControlPlane: null, + memoryKind: 'working-set-sum' as const, + processesTruncated: false, + sessionsTruncated: false, + processes: [], + sessions: [], +}); + +describe('resource history', () => { + it('owns immutable cgroup and CLI metrics even without a process table', () => { + const history = new ResourceHistory('machine', 'instance'); + const resource = { + memoryBytes: 123, + cpuCores: 0.5, + cpuPercentOfMachine: 25, + processCount: 2, + memoryKind: 'cgroup-current' as const, + quality: 'exact-cgroup' as const, + }; + const cleanup = { state: 'running' as const, attemptedAtMs: 1 }; + history.append({ + ...sample(1), + source: 'not-sampled', + cliControlPlane: { ...resource, memoryKind: 'rss', quality: 'exact-process' }, + sessions: [ + { sessionId: 'session', parentSessionId: null, status: 'stopping', cleanup, resource }, + ], + }); + resource.memoryBytes = 999; + cleanup.attemptedAtMs = 999; + const receipt = history.read(1).samples[0]; + expect(receipt?.cliControlPlane?.memoryBytes).toBe(123); + expect(receipt?.sessions[0]?.resource.memoryBytes).toBe(123); + expect(receipt?.sessions[0]?.cleanup?.attemptedAtMs).toBe(1); + expect(MachineResourceHistorySchema.safeParse(history.read(1)).success).toBe(true); + }); + it('bounds count and age and keeps observer gaps without fabricated samples', () => { + const history = new ResourceHistory('machine', 'instance'); + for (let i = 0; i < 150; i++) history.append(sample(i)); + expect(history.read(150).samples).toHaveLength(120); + expect(history.read(150).samples[0]?.sampledAtMs).toBe(30); + expect(history.read(700_000).samples).toEqual([]); + }); + it('caps process rows, returns independent copies, and distinguishes unavailable from empty', () => { + const history = new ResourceHistory('machine', 'instance'); + history.append({ + ...sample(1), + processes: Array.from({ length: 300 }, (_, i) => ({ + pid: i + 1, + startedAtMs: 1, + sessionId: null, + memoryBytes: 1, + cpuTimeMicros: 1, + })), + }); + history.append({ ...sample(2), source: 'unavailable' }); + const result = history.read(2); + expect(result.samples[0]?.processes).toHaveLength(256); + expect(result.samples[0]?.processesTruncated).toBe(true); + expect(result.samples[1]?.source).toBe('unavailable'); + expect(MachineResourceHistorySchema.safeParse(result).success).toBe(true); + result.samples.length = 0; + expect(history.read(2).samples).toHaveLength(2); + }); + it('does not attribute a reused root PID or its new descendants to the old session', () => { + const attribution = new ObservedProcessAttribution(); + const roots = [{ sessionId: 'session' as SessionId, startedAtMs: 1, rootPids: [10] }]; + const root = { + pid: 10, + parentPid: 1, + processGroupId: null, + startedAtMs: 5, + processIdentity: '50000', + memoryBytes: 1, + cpuTimeMicros: 1, + }; + expect(attribution.assign([root], roots).get(10)).toBe('session'); + attribution.assign([], roots); + // Display timestamps may be equal even though the provider identifies a new process. + const reused = { ...root, processIdentity: '50001' }; + expect(attribution.assign([reused, { ...reused, pid: 11, parentPid: 10 }], roots).size).toBe(0); + }); + it('leaves coarse POSIX observations unassigned even when PID and lstart match', () => { + const attribution = new ObservedProcessAttribution(); + const roots = [{ sessionId: 'session' as SessionId, startedAtMs: 1, rootPids: [10] }]; + const root = { + pid: 10, + parentPid: 1, + processGroupId: 10, + startedAtMs: 1000, + memoryBytes: 1, + cpuTimeMicros: 1, + }; + expect(attribution.assign([root], roots).size).toBe(0); + expect(attribution.assign([root, { ...root, pid: 11, parentPid: 10 }], roots).size).toBe(0); + }); +}); diff --git a/apps/cli/src/monitor/resource-history.ts b/apps/cli/src/monitor/resource-history.ts new file mode 100644 index 000000000..1e9b6fb53 --- /dev/null +++ b/apps/cli/src/monitor/resource-history.ts @@ -0,0 +1,44 @@ +import type { MachineResourceHistory } from '@lody/shared'; + +type Sample = MachineResourceHistory['samples'][number]; +const MAX_AGE_MS = 10 * 60 * 1000; + +/** Observation-only history. Reading it never starts a process probe. */ +export class ResourceHistory { + private samples: Sample[] = []; + + constructor( + private readonly machineId: string, + private readonly instanceId: string + ) {} + + append(sample: Sample): void { + this.samples.push( + structuredClone({ + ...sample, + processes: sample.processes.slice(0, 256), + processesTruncated: sample.processesTruncated || sample.processes.length > 256, + sessions: sample.sessions.slice(0, 100), + sessionsTruncated: sample.sessionsTruncated || sample.sessions.length > 100, + }) + ); + this.trim(sample.sampledAtMs); + } + + read(now = Date.now()): MachineResourceHistory { + this.trim(now); + return structuredClone({ + type: 'machine/resource-history', + machineId: this.machineId, + instanceId: this.instanceId, + collectedWhileObserved: true, + samples: this.samples, + }); + } + + private trim(now: number): void { + this.samples = this.samples + .filter((sample) => sample.sampledAtMs >= now - MAX_AGE_MS) + .slice(-120); + } +} diff --git a/apps/cli/src/session/AGENTS.md b/apps/cli/src/session/AGENTS.md index 0cf175627..a0d2b3aca 100644 --- a/apps/cli/src/session/AGENTS.md +++ b/apps/cli/src/session/AGENTS.md @@ -254,6 +254,9 @@ the frozen identity. Never fall back to the Session owner when the driving Turn drains. Shutdown closes session admission before awaiting work. The process boundary reserves a separate forced-cleanup deadline and sweeps retained workspace/preparation owners concurrently before exiting. + Idle and memory-pressure eviction preserve pending/running first-class background tasks + and ACP terminals until observed exit, including terminal startup. Retained output from + an exited terminal is not liveness. Raw cron fire times are not completion evidence. - `session-preparation-service.ts` — process-local speculative ACP lease/state owner. Peek/claim are synchronous published-resource snapshots and must never delay cold fallback; peek never transfers ownership. A prepared resource may reuse its open diff --git a/apps/cli/src/session/session-dispatch-watcher.ts b/apps/cli/src/session/session-dispatch-watcher.ts index db7dde8c2..943e8f39b 100644 --- a/apps/cli/src/session/session-dispatch-watcher.ts +++ b/apps/cli/src/session/session-dispatch-watcher.ts @@ -371,6 +371,8 @@ export class SessionDispatchWatcher { * Separate from `sessionCheckChains` so cancels are not blocked by long dispatches. */ private readonly cancelCheckChains = new Map>(); + private readonly activeSessionWatchReconciles = new Map(); + private readonly gcCleanupLeases = new Map(); /** * Tracks the last `meta.lastCanceledTurn` value we have already processed per session. @@ -492,6 +494,7 @@ export class SessionDispatchWatcher { this.sessionCheckChains.clear(); this.cancelCheckChains.clear(); this.cancelSeenTurn.clear(); + this.gcCleanupLeases.clear(); this.rpcTurnStash.clear(); this.turnSourceHints.clear(); this.rpcTurnOfferSubscribers.clear(); @@ -547,6 +550,11 @@ export class SessionDispatchWatcher { * are coalesced: exactly one follow-up check will run after the current one finishes. */ enqueueSessionCheck(sessionId: SessionId, options: SessionCheckOptions = {}): Promise { + const cleanup = this.gcCleanupLeases.get(sessionId); + if (cleanup) { + cleanup.dispatch = true; + return Promise.resolve(); + } // Tests may drive an as-yet-unstarted watcher directly (generation 0). Once // a production watcher has ever started, every check is generation-bound, // including RPC/access callbacks that happen to enqueue after stop(). @@ -733,6 +741,42 @@ export class SessionDispatchWatcher { return (this.rpcTurnStash.get(sessionId)?.size ?? 0) > 0 || this.accessFibers.has(sessionId); } + /** Hold dispatch admission through GC's process, document and transient-state cleanup. */ + tryAcquireGCCleanupLease(sessionId: SessionId): (() => void) | null { + if ( + (this.lifecycleGeneration > 0 && !this.started) || + this.gcCleanupLeases.has(sessionId) || + this.sessionCheckChains.has(sessionId) || + this.cancelCheckChains.has(sessionId) || + this.activeSessionWatchReconciles.has(sessionId) || + this.pendingMetadataSessionIds.has(sessionId) || + this.hasPendingDispatch(sessionId) + ) + return null; + + const lease = { dispatch: false, cancel: false }; + const generation = this.lifecycleGeneration; + this.gcCleanupLeases.set(sessionId, lease); + return () => { + if (this.gcCleanupLeases.get(sessionId) !== lease) return; + this.gcCleanupLeases.delete(sessionId); + // A successful cleanup evicted this mirror; a failed cleanup can safely + // reattach it. Never keep a subscription to a destroyed SessionDocument. + this.watchedSessions.get(sessionId)?.unsubscribe(); + this.watchedSessions.delete(sessionId); + if (generation !== this.lifecycleGeneration) return; + if (this.started) { + // Reconcile metadata first so idle sessions stay unloaded and pending + // RPC/meta/cancel work opens and subscribes to the current document. + this.enqueueMetadataReconcile(sessionId); + } else if (generation === 0) { + // Match direct, not-yet-started watcher checks used by local callers. + if (lease.dispatch) void this.enqueueSessionCheck(sessionId); + if (lease.cancel) void this.enqueueCancelCheck(sessionId); + } + }; + } + /** Drop expired stashed RPC turns across all sessions (bounded cleanup). */ private sweepExpiredRpcTurns(): void { const now = Date.now(); @@ -844,6 +888,11 @@ export class SessionDispatchWatcher { /** Enqueue a cancel check (separate chain from dispatch — see class doc). */ private enqueueCancelCheck(sessionId: SessionId, lifecycleGeneration?: number): Promise { + const cleanup = this.gcCleanupLeases.get(sessionId); + if (cleanup) { + cleanup.cancel = true; + return Promise.resolve(); + } const previous = this.cancelCheckChains.get(sessionId) ?? Promise.resolve(); const next = previous .catch(() => {}) @@ -879,6 +928,7 @@ export class SessionDispatchWatcher { * otherwise monopolize the event loop and retain all of their cloud rooms. */ private enqueueMetadataReconcile(sessionId: SessionId): void { + if (this.gcCleanupLeases.has(sessionId)) return; // Reinsert so a fresh event moves ahead of stale catch-up work already in // the queue. The drain takes from the newest end in bounded batches. this.pendingMetadataSessionIds.delete(sessionId); @@ -1124,6 +1174,11 @@ export class SessionDispatchWatcher { if (!isActive()) { return; } + if (this.gcCleanupLeases.has(sessionId)) return; + this.activeSessionWatchReconciles.set( + sessionId, + (this.activeSessionWatchReconciles.get(sessionId) ?? 0) + 1 + ); const roomId = getSessionRoomId(sessionId); let phase: SessionReconcilePhase = 'read-doc-meta'; try { @@ -1220,6 +1275,10 @@ export class SessionDispatchWatcher { )}` ); throw error; + } finally { + const remaining = (this.activeSessionWatchReconciles.get(sessionId) ?? 1) - 1; + if (remaining === 0) this.activeSessionWatchReconciles.delete(sessionId); + else this.activeSessionWatchReconciles.set(sessionId, remaining); } } diff --git a/apps/cli/src/session/session-execution-service.ts b/apps/cli/src/session/session-execution-service.ts index 9c1f840c6..e60511777 100644 --- a/apps/cli/src/session/session-execution-service.ts +++ b/apps/cli/src/session/session-execution-service.ts @@ -651,6 +651,8 @@ export class SessionExecutionService { private readonly currentTurnBySession = new Map(); private readonly turnRuntimeBySession = new Map(); private readonly rewriteBarrierSessions = new Set(); + private readonly gcCleanupWaiters = new Map>(); + private readonly gcProtectedOperations = new Map(); private readonly rewriteConflictLeaseSessions = new Set(); private readonly turnReleaseWaiters = new Map void>>>(); // Serializes ownership mutations per session so prompt completion and steer @@ -1042,6 +1044,42 @@ export class SessionExecutionService { }; } + tryAcquireGCCleanupLease(sessionId: SessionId): (() => void) | null { + if (this.gcCleanupWaiters.has(sessionId) || this.gcProtectedOperations.has(sessionId)) + return null; + const releaseRewrite = this.tryAcquireSessionRewriteBarrier(sessionId); + if (!releaseRewrite) return null; + let resolve!: () => void; + this.gcCleanupWaiters.set( + sessionId, + new Promise((done) => { + resolve = done; + }) + ); + let released = false; + return () => { + if (released) return; + released = true; + this.gcCleanupWaiters.delete(sessionId); + releaseRewrite(); + resolve(); + }; + } + + private async acquireGCProtectedOperation(sessionId: SessionId): Promise<() => void> { + // Direct start/continue/steer enter before document access. A cleanup that + // already owns the session finishes first; otherwise GC must defer to us. + while (this.gcCleanupWaiters.has(sessionId)) { + await this.gcCleanupWaiters.get(sessionId); + } + this.gcProtectedOperations.set(sessionId, (this.gcProtectedOperations.get(sessionId) ?? 0) + 1); + return () => { + const remaining = (this.gcProtectedOperations.get(sessionId) ?? 1) - 1; + if (remaining === 0) this.gcProtectedOperations.delete(sessionId); + else this.gcProtectedOperations.set(sessionId, remaining); + }; + } + tryAcquireSessionRewriteBarrier(sessionId: SessionId): (() => void) | null { if ( this.rewriteBarrierSessions.has(sessionId) || @@ -1121,30 +1159,35 @@ export class SessionExecutionService { timestamp: string; inputConfig: SessionTurnInputConfig; }): Promise { - return await this.steerMutationQueue.enqueue(options.sessionId, async () => { - const releaseConflict = this.tryAcquireSessionRewriteConflictLease(options.sessionId); - if (!releaseConflict) { - // Nothing was submitted, so this guide is still ours to run. Only the - // dispatch pointer is written: the history flip needs the lease we just - // failed to take, and dispatch honors the pointer on its own. - await this.requeueUndeliveredSteer(options.sessionId, options.userTurnId, { - canWriteHistory: false, - }); - return { - type: 'session/steer_response', - sessionId: options.sessionId, - userTurnId: options.userTurnId, - applied: false, - disposition: 'busy', - error: 'The session history is being replaced.', - }; - } - try { - return await this.steerSessionLocked(options); - } finally { - releaseConflict(); - } - }); + const releaseGCOperation = await this.acquireGCProtectedOperation(options.sessionId); + try { + return await this.steerMutationQueue.enqueue(options.sessionId, async () => { + const releaseConflict = this.tryAcquireSessionRewriteConflictLease(options.sessionId); + if (!releaseConflict) { + // Nothing was submitted, so this guide is still ours to run. Only the + // dispatch pointer is written: the history flip needs the lease we just + // failed to take, and dispatch honors the pointer on its own. + await this.requeueUndeliveredSteer(options.sessionId, options.userTurnId, { + canWriteHistory: false, + }); + return { + type: 'session/steer_response', + sessionId: options.sessionId, + userTurnId: options.userTurnId, + applied: false, + disposition: 'busy', + error: 'The session history is being replaced.', + }; + } + try { + return await this.steerSessionLocked(options); + } finally { + releaseConflict(); + } + }); + } finally { + releaseGCOperation(); + } } private async steerSessionLocked(options: { @@ -3304,22 +3347,29 @@ export class SessionExecutionService { message: SessionChatRequestValidated, dispatchOptions?: SessionDispatchOptions ): Promise { - const turn = await this.prepareContinueSessionTurn(message, dispatchOptions); - if ( - dispatchOptions?.dispatchSource !== 'delivery' && - (await this.markCancelledUserTurnBeforeOwner({ - sessionId: message.sessionId, - sessionDoc: turn.options.sessionDoc, - userTurnId: message.userTurnId, - })) - ) { - return; + const releaseGCOperation = await this.acquireGCProtectedOperation(message.sessionId); + try { + const turn = await this.prepareContinueSessionTurn(message, dispatchOptions); + if ( + dispatchOptions?.dispatchSource !== 'delivery' && + (await this.markCancelledUserTurnBeforeOwner({ + sessionId: message.sessionId, + sessionDoc: turn.options.sessionDoc, + userTurnId: message.userTurnId, + })) + ) { + return; + } + const body = dispatchOptions?.onTurnClaimed + ? (ctx: VisibleSessionTurnContext) => + Effect.promise(dispatchOptions.onTurnClaimed!).pipe( + Effect.flatMap(() => turn.body(ctx)) + ) + : turn.body; + await this.runVisibleSessionTurn(turn.options, body); + } finally { + releaseGCOperation(); } - const body = dispatchOptions?.onTurnClaimed - ? (ctx: VisibleSessionTurnContext) => - Effect.promise(dispatchOptions.onTurnClaimed!).pipe(Effect.flatMap(() => turn.body(ctx))) - : turn.body; - await this.runVisibleSessionTurn(turn.options, body); } private async prepareContinueSessionTurn( @@ -4174,21 +4224,26 @@ export class SessionExecutionService { message: SessionCreateRequestValidated, dispatchOptions?: SessionDispatchOptions ): Promise { - const turn = await this.prepareStartSessionTurn(message, dispatchOptions); - const userTurnId = - typeof message.userTurnId === 'string' && message.userTurnId.trim() - ? message.userTurnId.trim() - : undefined; - if ( - await this.markCancelledUserTurnBeforeOwner({ - sessionId: message.sessionId, - sessionDoc: turn.options.sessionDoc, - userTurnId, - }) - ) { - return; + const releaseGCOperation = await this.acquireGCProtectedOperation(message.sessionId); + try { + const turn = await this.prepareStartSessionTurn(message, dispatchOptions); + const userTurnId = + typeof message.userTurnId === 'string' && message.userTurnId.trim() + ? message.userTurnId.trim() + : undefined; + if ( + await this.markCancelledUserTurnBeforeOwner({ + sessionId: message.sessionId, + sessionDoc: turn.options.sessionDoc, + userTurnId, + }) + ) { + return; + } + await this.runVisibleSessionTurn(turn.options, turn.body); + } finally { + releaseGCOperation(); } - await this.runVisibleSessionTurn(turn.options, turn.body); } private async prepareStartSessionTurn( diff --git a/apps/cli/src/session/session-manager.test.ts b/apps/cli/src/session/session-manager.test.ts index 485ac56f8..5ba3f89f9 100644 --- a/apps/cli/src/session/session-manager.test.ts +++ b/apps/cli/src/session/session-manager.test.ts @@ -230,6 +230,33 @@ describe('SessionManager cleanup phases', () => { return { manager, workspaceDocument, sessions: internals.sessions }; }; + it('keeps GC admission closed after terminated removes the runtime until the outer lease releases', async () => { + const { manager, sessions } = cleanupFixture(); + const sessionId = 'gc-lease' as SessionId; + const events = new EventEmitter(); + const runtime = Object.assign(events, { + sessionId, + terminate: async () => events.emit('terminated', { sessionId, exitCode: 0 }), + }) as unknown as ISession; + ( + manager as unknown as { registerSessionEvents(session: ISession): void } + ).registerSessionEvents(runtime); + sessions.set(sessionId, runtime); + const release = manager.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + expect(manager.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + await manager.terminateSession(sessionId, true); + expect(manager.getSession(sessionId)).toBeNull(); + await expect(manager.createSession(createSessionConfig({ sessionId }))).rejects.toThrow( + 'cleanup must complete' + ); + release!(); + release!(); + const next = manager.tryAcquireGCCleanupLease(sessionId); + expect(next).not.toBeNull(); + next!(); + }); + it('retains failed cleanup ownership, waits for all attempts, and retries before closing documents', async () => { const { manager, workspaceDocument, sessions } = cleanupFixture(); const successId = 'cleanup-success' as SessionId; diff --git a/apps/cli/src/session/session-manager.ts b/apps/cli/src/session/session-manager.ts index 51d8379fb..5737b2cd9 100644 --- a/apps/cli/src/session/session-manager.ts +++ b/apps/cli/src/session/session-manager.ts @@ -350,6 +350,7 @@ export type SessionMonitorRuntimeInfo = { agentType: string; startedAtMs: number; runtimeStatus: 'created' | 'failed' | 'running' | 'stopping' | 'terminated'; + cleanup?: { state: 'running' | 'failed' | 'completed'; attemptedAtMs: number } | null; accounting: SessionResourceAccounting; }; @@ -452,12 +453,16 @@ export class SessionManager extends EventEmitter { private gitCredentialBroker: GitCredentialBroker | null = null; private readonly sessions = new Map(); private readonly cleanupOwnedSessions = new WeakSet(); + private readonly gcCleanupSessions = new Set(); private shuttingDown = false; private assertSessionAdmission(sessionId?: SessionId): void { if (this.shuttingDown) throw new Error('Session manager is shutting down'); const resident = sessionId ? this.sessions.get(sessionId) : undefined; - if (resident && this.cleanupOwnedSessions.has(resident)) { + if ( + (sessionId && this.gcCleanupSessions.has(sessionId)) || + (resident && this.cleanupOwnedSessions.has(resident)) + ) { throw new Error('Session cleanup must complete before replacement'); } } @@ -2080,6 +2085,25 @@ export class SessionManager extends EventEmitter { return session; } + /** Keeps replacement admission closed through GC's document and store cleanup. */ + tryAcquireGCCleanupLease(sessionId: SessionId): (() => void) | null { + if ( + this.shuttingDown || + this.gcCleanupSessions.has(sessionId) || + this.pendingSessionCreates.has(sessionId) || + this.pendingTerminationPromises.has(sessionId) || + this.preparationSessions.has(sessionId) + ) + return null; + this.gcCleanupSessions.add(sessionId); + let released = false; + return () => { + if (released) return; + released = true; + this.gcCleanupSessions.delete(sessionId); + }; + } + async terminateSession(sessionId: SessionId, force: boolean = false): Promise { const session = this.sessions.get(sessionId); if (!session) { diff --git a/apps/cli/src/session/session.ts b/apps/cli/src/session/session.ts index f6549b5a4..1e3b29720 100644 --- a/apps/cli/src/session/session.ts +++ b/apps/cli/src/session/session.ts @@ -229,6 +229,7 @@ export class Session extends EventEmitter implements ISession { agentCliType: this.config.agentCliType, agentType: this.config.agentType, startedAtMs: this.startedAtMs, + cleanup: this.cleanupState ? { ...this.cleanupState } : null, runtimeStatus: this.status === 'existing' || this.status === 'stopped' ? 'created' : this.status, accounting, @@ -244,6 +245,7 @@ export class Session extends EventEmitter implements ISession { } private terminationPromise: Promise | null = null; + private cleanupState: SessionMonitorRuntimeInfo['cleanup'] = null; private terminationForceRequested = false; private forceTerminationSignal: Promise = Promise.resolve(); private requestForceTermination: (() => void) | null = null; @@ -263,11 +265,15 @@ export class Session extends EventEmitter implements ISession { }); if (force) this.requestForceTermination?.(); this.status = 'stopping'; + this.cleanupState = { state: 'running', attemptedAtMs: Date.now() }; const termination = Promise.resolve().then(() => this.terminateOnce()); this.terminationPromise = termination; void termination.then( - () => {}, () => { + if (this.cleanupState) this.cleanupState.state = 'completed'; + }, + () => { + if (this.cleanupState) this.cleanupState.state = 'failed'; if (this.terminationPromise === termination) this.terminationPromise = null; } ); diff --git a/apps/cli/src/session/terminal-manager.ts b/apps/cli/src/session/terminal-manager.ts index 1b07f84b0..b40361d2e 100644 --- a/apps/cli/src/session/terminal-manager.ts +++ b/apps/cli/src/session/terminal-manager.ts @@ -31,6 +31,7 @@ export interface TerminalManager { waitForTerminalExit(acpSessionId: string, terminalId: string): Promise; killTerminal(acpSessionId: string, terminalId: string): Promise; disposeAll?(acpSessionId: string): Promise; + hasRunningTerminals?(): boolean; } interface TerminalState { @@ -134,6 +135,14 @@ abstract class BaseTerminalManager implements TerminalManager { } } + hasRunningTerminals(): boolean { + if (this.pendingStarts.size > 0) return true; + for (const terminal of this.terminals.values()) { + if (!terminal.disposed && terminal.exitStatus === null) return true; + } + return false; + } + async terminalOutput(acpSessionId: string, terminalId: string) { const state = this.getTerminal(acpSessionId, terminalId); return { diff --git a/apps/cli/tests/loro-doc-unload-data-plane-integration.test.ts b/apps/cli/tests/loro-doc-unload-data-plane-integration.test.ts index 4f4f65626..086e73fb4 100644 --- a/apps/cli/tests/loro-doc-unload-data-plane-integration.test.ts +++ b/apps/cli/tests/loro-doc-unload-data-plane-integration.test.ts @@ -57,7 +57,7 @@ import { ensureImplicitLocalWorkspace, loadOrCreateLocalIdentity, } from '../src/lib/cli-platform'; -import { LoroDocumentManager } from '../src/lib/loro/doc'; +import { LoroDocumentManager, SessionDocument } from '../src/lib/loro/doc'; import { makeLocalWorkspaceCatalog } from '../src/lib/local-workspace-catalog'; import type { Logger } from '../src/utils/logger'; @@ -482,6 +482,158 @@ describe('session GC unloads the repo doc and invalidates its local data-plane r } }); + it('joins the original in-flight destruction before reopening the document', async () => { + const harness = await createHarness(); + const { manager, sessionId } = harness; + const stale = await manager.getOrCreateSessionDoc(sessionId); + const entered = createDeferred(); + const release = createDeferred(); + const originalUnload = manager.repo.unloadDoc.bind(manager.repo); + const unload = vi.spyOn(manager.repo, 'unloadDoc').mockImplementationOnce(async (docId) => { + entered.resolve(); + await release.promise; + await originalUnload(docId); + }); + try { + const cleanup = manager.cleanSessionDoc(sessionId); + await entered.promise; + expect(stale.isDestroyed).toBe(true); + const opened = vi.fn(); + const reopening = manager.getOrCreateSessionDoc(sessionId).then((doc) => { + opened(); + return doc; + }); + // Drain queued ownership work while the original unload is still blocked. + await new Promise((resolve) => setImmediate(resolve)); + expect(unload).toHaveBeenCalledOnce(); + expect(opened).not.toHaveBeenCalled(); + expect(manager.sessions.get(sessionId)).toBe(stale); + release.resolve(); + await cleanup; + const fresh = await reopening; + expect(fresh).not.toBe(stale); + expect(fresh.isDestroyed).toBe(false); + expect(fresh.mirror).not.toBeNull(); + expect(manager.sessions.get(sessionId)).toBe(fresh); + expect(unload).toHaveBeenCalledOnce(); + expect((await fresh.getHistory()).map((entry) => entry.id)).toContain(harness.cliEntryId); + } finally { + release.resolve(); + unload.mockRestore(); + await harness.dispose(); + } + }); + it('keeps a failed destroyed wrapper private until coalesced teardown retry opens a fresh doc', async () => { + const harness = await createHarness(); + const { manager, sessionId } = harness; + const stale = await manager.getOrCreateSessionDoc(sessionId); + const unload = vi + .spyOn(manager.repo, 'unloadDoc') + .mockRejectedValueOnce(new Error('first unload failed')) + .mockRejectedValueOnce(new Error('retry unload failed')); + try { + await expect(manager.cleanSessionDoc(sessionId)).rejects.toThrow('first unload failed'); + expect(stale.isDestroyed).toBe(true); + expect(manager.sessions.get(sessionId)).toBe(stale); + await expect(manager.getOrCreateSessionDoc(sessionId)).rejects.toThrow('retry unload failed'); + expect(manager.sessions.get(sessionId)).toBe(stale); + const retry = vi.spyOn(stale, 'destroy'); + const [fresh, same] = await Promise.all([ + manager.getOrCreateSessionDoc(sessionId), + manager.getOrCreateSessionDoc(sessionId), + ]); + expect(fresh).not.toBe(stale); + expect(same).toBe(fresh); + expect(fresh.isDestroyed).toBe(false); + expect(fresh.mirror).not.toBeNull(); + expect(manager.sessions.get(sessionId)).toBe(fresh); + expect(stale.mirror).toBeNull(); + expect(retry).toHaveBeenCalledExactlyOnceWith({ preserveStatus: true }); + expect(unload).toHaveBeenCalledTimes(3); + expect((await fresh.getHistory()).map((entry) => entry.id)).toContain(harness.cliEntryId); + } finally { + unload.mockRestore(); + await harness.dispose(); + } + }); + + it.each(['reopen', 'cleanup'] as const)( + 'preserves a replacement installed during %s teardown', + async (operation) => { + const harness = await createHarness(); + const { manager, sessionId } = harness; + const stale = await manager.getOrCreateSessionDoc(sessionId); + try { + if (operation === 'reopen') { + const unload = vi + .spyOn(manager.repo, 'unloadDoc') + .mockRejectedValueOnce(new Error('unload failed')); + await expect(manager.cleanSessionDoc(sessionId)).rejects.toThrow('unload failed'); + unload.mockRestore(); + } + const originalDestroy = stale.destroy.bind(stale); + const replacement = new SessionDocument( + manager.repo, + sessionId, + (docId) => manager.unloadDocRoom(docId), + createSilentLogger() + ); + vi.spyOn(stale, 'destroy').mockImplementationOnce(async (options) => { + await originalDestroy(options); + await replacement.init(); + manager.sessions.set(sessionId, replacement); + }); + if (operation === 'reopen') { + expect(await manager.getOrCreateSessionDoc(sessionId)).toBe(replacement); + } else { + await manager.cleanSessionDoc(sessionId); + } + expect(manager.sessions.get(sessionId)).toBe(replacement); + expect(replacement.isDestroyed).toBe(false); + expect(replacement.mirror).not.toBeNull(); + } finally { + await harness.dispose(); + } + } + ); + + it('propagates destruction failure after awaiting a pending initialization', async () => { + const harness = await createHarness(); + const { manager } = harness; + const sessionId = 'pending-cleanup-failure' as SessionId; + const entered = createDeferred(); + const release = createDeferred(); + const originalInit = SessionDocument.prototype.init; + const init = vi.spyOn(SessionDocument.prototype, 'init').mockImplementationOnce(async function ( + this: SessionDocument + ) { + await originalInit.call(this); + entered.resolve(); + await release.promise; + }); + const unload = vi + .spyOn(manager.repo, 'unloadDoc') + .mockRejectedValueOnce(new Error('pending unload failed')); + try { + const opening = manager.getOrCreateSessionDoc(sessionId); + await entered.promise; + const cleanup = manager.cleanSessionDoc(sessionId); + const rejected = expect(cleanup).rejects.toThrow('pending unload failed'); + release.resolve(); + const stale = await opening; + await rejected; + expect(stale.isDestroyed).toBe(true); + expect(manager.sessions.get(sessionId)).toBe(stale); + const fresh = await manager.getOrCreateSessionDoc(sessionId); + expect(fresh).not.toBe(stale); + expect(fresh.isDestroyed).toBe(false); + } finally { + release.resolve(); + init.mockRestore(); + unload.mockRestore(); + await harness.dispose(); + } + }); it('lets a renderer update authored after session GC reach the CLI', async () => { const harness = await createHarness(); try { diff --git a/apps/cli/tests/machine-resource-history-rpc.test.ts b/apps/cli/tests/machine-resource-history-rpc.test.ts new file mode 100644 index 000000000..2122304c7 --- /dev/null +++ b/apps/cli/tests/machine-resource-history-rpc.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it, vi } from 'vitest'; +import { LocalMachineRpcRequestSchema, LocalMachineRpcResponseSchema } from '@lody/shared'; +import { MachineRuntime } from '../src/lib/machine-runtime'; + +describe('local resource history RPC', () => { + it('serves existing observations without starting work and rejects session-scoped access', async () => { + const history = { + type: 'machine/resource-history' as const, + machineId: 'machine', + instanceId: 'instance', + collectedWhileObserved: true as const, + samples: [], + }; + const sample = vi.fn(() => { + throw new Error('must not probe'); + }); + const runtime: MachineRuntime = Object.assign(Object.create(MachineRuntime.prototype), { + resourceMonitor: { getHistory: () => history, sample }, + }); + const request = LocalMachineRpcRequestSchema.parse({ + method: 'machine/get-resource-history', + machineId: 'machine', + workspaceId: 'workspace', + params: {}, + }); + const response = await runtime.dispatchLocalMachineRpc(request); + expect(LocalMachineRpcResponseSchema.parse(response)).toEqual({ ok: true, result: history }); + expect( + await runtime.dispatchLocalMachineRpc({ ...request, ownerSessionId: 'session' }) + ).toEqual({ ok: false, error: 'Machine resource history requires workspace-level access' }); + expect(sample).not.toHaveBeenCalled(); + Object.assign(runtime, { resourceMonitor: null }); + expect(await runtime.dispatchLocalMachineRpc(request)).toEqual({ + ok: false, + error: 'Resource monitor stopped', + }); + }); +}); diff --git a/apps/cli/tests/message-handler-machine-registration.test.ts b/apps/cli/tests/message-handler-machine-registration.test.ts index 6b86e6830..a899737b2 100644 --- a/apps/cli/tests/message-handler-machine-registration.test.ts +++ b/apps/cli/tests/message-handler-machine-registration.test.ts @@ -186,6 +186,7 @@ describe('MessageHandler machine registration', () => { localProjectRemoval: 1, providerSetup: 1, acpProtocolAuthentication: 2, + resourceHistory: 1, }); await handler.cleanup(); diff --git a/apps/cli/tests/message-handler-protected-work.test.ts b/apps/cli/tests/message-handler-protected-work.test.ts new file mode 100644 index 000000000..7017b97e3 --- /dev/null +++ b/apps/cli/tests/message-handler-protected-work.test.ts @@ -0,0 +1,305 @@ +import { expect, it, vi } from 'vitest'; +import { SessionIdSchema } from '@lody/shared'; +import { MessageHandler } from '../src/lib/message-handler'; + +const sessionId = SessionIdSchema.parse('protected-work'); +const pendingTask = { type: 'subagent_task', taskId: 'task-1', status: 'pending' }; +const goal = (status: string) => ({ + type: 'goal', + threadId: 'goal-1', + objective: 'Finish work', + status, +}); + +function fixture(items: unknown[] = []) { + const hasRunningTerminals = vi.fn(() => false); + const initialRuntime = { terminalManager: { hasRunningTerminals } }; + let runtime: typeof initialRuntime | null = initialRuntime; + const getHistory = vi.fn(async () => [{ items }]); + const getMetaState = vi.fn(async (): Promise => null); + const receiver = { + sessionManager: { getSession: () => runtime }, + workspaceDocument: { getOrCreateSessionDoc: async () => ({ getHistory, getMetaState }) }, + }; + return { + read: () => MessageHandler.prototype.hasProtectedWork.call(receiver, sessionId), + getHistory, + getMetaState, + hasRunningTerminals, + removeRuntime: () => { + runtime = null; + }, + replaceRuntime: () => { + runtime = { terminalManager: { hasRunningTerminals } }; + }, + }; +} + +it('reads history once for both goals and background tasks', async () => { + const f = fixture([goal('paused'), pendingTask]); + await expect(f.read()).resolves.toBe(true); + expect(f.getHistory).toHaveBeenCalledOnce(); +}); + +it('preserves active goals even without a runtime and lets latest history override legacy goal', async () => { + const active = fixture([goal('active')]); + active.removeRuntime(); + await expect(active.read()).resolves.toBe(true); + const paused = fixture([goal('paused')]); + paused.getMetaState.mockResolvedValue({ latestGoal: goal('active') }); + paused.removeRuntime(); + await expect(paused.read()).resolves.toBe(false); + const legacy = fixture(); + legacy.getMetaState.mockResolvedValue({ latestGoal: goal('active') }); + legacy.removeRuntime(); + await expect(legacy.read()).resolves.toBe(true); +}); + +it('does not let stale task snapshots pin state after the runtime has exited', async () => { + const f = fixture([pendingTask]); + f.removeRuntime(); + await expect(f.read()).resolves.toBe(false); +}); + +it('rechecks runtime ownership after awaiting history', async () => { + const f = fixture([pendingTask]); + f.getHistory.mockImplementation(async () => { + f.removeRuntime(); + return [{ items: [pendingTask] }]; + }); + await expect(f.read()).resolves.toBe(false); +}); + +it('does not apply an old history snapshot to a replacement runtime', async () => { + const f = fixture(); + f.getHistory.mockImplementation(async () => { + f.replaceRuntime(); + return [{ items: [] }]; + }); + await expect(f.read()).resolves.toBe(true); +}); + +it('checks live terminals before history and again after awaited metadata', async () => { + const live = fixture(); + live.hasRunningTerminals.mockReturnValue(true); + await expect(live.read()).resolves.toBe(true); + expect(live.getHistory).not.toHaveBeenCalled(); + const raced = fixture(); + raced.getMetaState.mockImplementation(async () => { + raced.hasRunningTerminals.mockReturnValue(true); + return null; + }); + await expect(raced.read()).resolves.toBe(true); +}); + +it('surfaces history errors for the per-session GC guard', async () => { + const f = fixture(); + f.getHistory.mockRejectedValue(new Error('history unavailable')); + await expect(f.read()).rejects.toThrow('history unavailable'); +}); + +it.each(['turn', 'terminal', 'history', 'replacement', 'metadata', 'rpc'] as const)( + 'refuses cleanup when %s changes during preview close without clearing presence', + async (change) => { + let active = false; + let terminal = false; + let state = { history: [] as unknown[] }; + let metadataClock = 0; + let pendingDispatch = false; + const metadata = { + version: () => ({ peer: { physicalTime: metadataClock, logicalCounter: 0 } }), + }; + let runtime = { terminalManager: { hasRunningTerminals: () => terminal } }; + const doc = { mirror: { getState: () => state } }; + const terminateSession = vi.fn(async () => {}); + const clearPresence = vi.fn(); + const cleanSessionDoc = vi.fn(async () => {}); + const deleteSession = vi.fn(); + const receiver = { + logger: { debug: () => {} }, + sessionManager: { getSession: () => runtime, terminateSession }, + workspaceDocument: { + sessions: new Map([[sessionId, doc]]), + cleanSessionDoc, + repo: { getMeta: () => metadata }, + }, + sessionDispatchWatcher: { hasPendingDispatch: () => pendingDispatch }, + hasActiveTurn: () => active, + hasPendingUpdates: () => false, + isArchiveInFlight: () => false, + clearSessionActivePresence: clearPresence, + store: { deleteSession }, + previewService: { + closeSessionPreviewForCleanup: async () => { + if (change === 'turn') active = true; + if (change === 'terminal') terminal = true; + if (change === 'history') state = { history: [goal('active')] }; + if (change === 'metadata') metadataClock++; + if (change === 'rpc') pendingDispatch = true; + if (change === 'replacement') + runtime = { terminalManager: { hasRunningTerminals: () => false } }; + }, + }, + }; + const guard = MessageHandler.prototype.captureGCCleanupGuard.call(receiver, sessionId); + expect(guard()).toBe(true); + await expect( + MessageHandler.prototype.cleanSessionForGC.call(receiver, sessionId, guard) + ).resolves.toBe(false); + expect(terminateSession).not.toHaveBeenCalled(); + expect(clearPresence).not.toHaveBeenCalled(); + expect(cleanSessionDoc).not.toHaveBeenCalled(); + expect(deleteSession).not.toHaveBeenCalled(); + } +); + +it('starts termination synchronously after the final cleanup guard', async () => { + const events: string[] = []; + const receiver = { + logger: { debug: () => {} }, + previewService: { + closeSessionPreviewForCleanup: async () => { + events.push('preview'); + }, + }, + sessionManager: { + tryAcquireGCCleanupLease: () => () => events.push('release-manager'), + terminateSession: () => { + events.push('terminate'); + return Promise.resolve(); + }, + }, + sessionDispatchWatcher: { + tryAcquireGCCleanupLease: () => () => events.push('release-dispatch'), + }, + executionService: { tryAcquireGCCleanupLease: () => () => events.push('release-execution') }, + clearSessionActivePresence: () => events.push('presence'), + workspaceDocument: { + cleanSessionDoc: async () => { + events.push('doc'); + }, + }, + store: { deleteSession: () => events.push('store') }, + }; + const guard = () => { + events.push('guard'); + return true; + }; + await expect( + MessageHandler.prototype.cleanSessionForGC.call(receiver, sessionId, guard) + ).resolves.toBe(true); + expect(events).toEqual([ + 'guard', + 'preview', + 'guard', + 'guard', + 'terminate', + 'presence', + 'doc', + 'store', + 'release-manager', + 'release-execution', + 'release-dispatch', + ]); +}); + +it.each(['success', 'termination-failure', 'document-failure'] as const)( + 'holds all GC leases across termination and document awaits, releasing on %s', + async (outcome) => { + let finishTermination!: () => void; + let finishDocument!: () => void; + let documentStarted!: () => void; + const termination = new Promise((resolve) => { + finishTermination = resolve; + }); + const document = new Promise((resolve) => { + finishDocument = resolve; + }); + const documentStart = new Promise((resolve) => { + documentStarted = resolve; + }); + const held = new Set(); + const acquire = (name: string) => () => { + held.add(name); + return () => held.delete(name); + }; + const deleteSession = vi.fn(() => + expect([...held].sort()).toEqual(['dispatch', 'execution', 'manager']) + ); + const cleanSessionDoc = vi.fn(async () => { + documentStarted(); + await document; + if (outcome === 'document-failure') throw new Error('document failure'); + }); + const receiver = { + logger: { debug: () => {} }, + previewService: { closeSessionPreviewForCleanup: async () => {} }, + sessionManager: { + tryAcquireGCCleanupLease: acquire('manager'), + terminateSession: async () => { + await termination; + if (outcome === 'termination-failure') throw new Error('termination failure'); + }, + }, + sessionDispatchWatcher: { tryAcquireGCCleanupLease: acquire('dispatch') }, + executionService: { tryAcquireGCCleanupLease: acquire('execution') }, + clearSessionActivePresence: () => {}, + workspaceDocument: { cleanSessionDoc }, + store: { deleteSession }, + }; + const cleanup = MessageHandler.prototype.cleanSessionForGC.call( + receiver, + sessionId, + () => true + ); + const result = + outcome === 'success' + ? expect(cleanup).resolves.toBe(true) + : expect(cleanup).rejects.toThrow( + outcome === 'termination-failure' ? 'termination failure' : 'document failure' + ); + await Promise.resolve(); + expect([...held].sort()).toEqual(['dispatch', 'execution', 'manager']); + finishTermination(); + if (outcome !== 'termination-failure') { + await documentStart; + expect([...held].sort()).toEqual(['dispatch', 'execution', 'manager']); + expect(deleteSession).not.toHaveBeenCalled(); + finishDocument(); + } + await result; + expect(held.size).toBe(0); + expect(deleteSession).toHaveBeenCalledTimes(outcome === 'success' ? 1 : 0); + } +); + +it.each(['dispatch', 'execution', 'manager'] as const)( + 'defers GC and releases earlier leases when %s admission is busy', + async (busy) => { + const held = new Set(); + const acquire = (name: string) => () => { + if (name === busy) return null; + held.add(name); + return () => held.delete(name); + }; + const terminateSession = vi.fn(); + const clearSessionActivePresence = vi.fn(); + const deleteSession = vi.fn(); + const receiver = { + logger: { debug: () => {} }, + previewService: { closeSessionPreviewForCleanup: async () => {} }, + sessionManager: { tryAcquireGCCleanupLease: acquire('manager'), terminateSession }, + sessionDispatchWatcher: { tryAcquireGCCleanupLease: acquire('dispatch') }, + executionService: { tryAcquireGCCleanupLease: acquire('execution') }, + clearSessionActivePresence, + store: { deleteSession }, + }; + await expect( + MessageHandler.prototype.cleanSessionForGC.call(receiver, sessionId, () => true) + ).resolves.toBe(false); + expect(held.size).toBe(0); + expect(terminateSession).not.toHaveBeenCalled(); + expect(clearSessionActivePresence).not.toHaveBeenCalled(); + expect(deleteSession).not.toHaveBeenCalled(); + } +); diff --git a/apps/cli/tests/session-dispatch-watcher-gc.test.ts b/apps/cli/tests/session-dispatch-watcher-gc.test.ts new file mode 100644 index 000000000..efe39bf59 --- /dev/null +++ b/apps/cli/tests/session-dispatch-watcher-gc.test.ts @@ -0,0 +1,261 @@ +import { describe, expect, it, vi } from 'vitest'; +import { SessionIdSchema, type SessionMeta, type WorkspaceId } from '@lody/shared'; +import { SessionDispatchWatcher } from '../src/session/session-dispatch-watcher'; +import type { LoroDocumentManager } from '../src/lib/loro/doc'; +import type { SessionExecutionService } from '../src/session/session-execution-service'; +import type { Logger } from '../src/utils/logger'; + +const sessionId = SessionIdSchema.parse('gc-lease-session'); +const logger: Logger = { + info() {}, + warn() {}, + error() {}, + success() {}, + debug() {}, + setLevel() {}, + child: () => logger, + close: async () => {}, +}; +function deferred() { + let resolve = () => {}; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} +function fixture() { + let meta: SessionMeta = { + id: sessionId, + machineId: 'machine-1', + userId: 'user-1', + createdAt: new Date().toISOString(), + cliType: 'builtin', + agentType: 'codex', + status: { type: 'idle' }, + }; + let notifyMetadata = (_event: { kind: string; docId: string }) => {}; + const makeDoc = () => ({ + mirror: { subscribe: vi.fn(() => vi.fn()) }, + getMetaState: vi.fn(async () => meta), + getHistory: vi.fn(async () => []), + updateHistory: vi.fn(async () => {}), + setStatus: vi.fn(async () => {}), + waitForRemoteSync: vi.fn(async () => {}), + }); + let doc = makeDoc(); + const getDocMeta = vi.fn(async () => ({ meta })); + const getOrCreateSessionDoc = vi.fn(async () => doc); + const cancel = vi.fn(async () => ({ success: true })); + const dispatch = vi.fn( + async (options: { + onAccessAllowed: () => void | Promise; + accessPromise: Promise; + requestPromise: Promise; + }) => { + await options.accessPromise; + await options.requestPromise; + await options.onAccessAllowed(); + } + ); + const watcher = new SessionDispatchWatcher({ + logger, + machineId: 'machine-1', + workspaceId: 'workspace-1' as WorkspaceId, + userResolver: { + resolve: async (id) => ({ id, name: 'User', email: 'user@example.com' }), + clear() {}, + }, + canUseMachine: async () => ({ outcome: 'allowed' }), + workspaceDocument: { + repo: { + getDocMeta, + upsertDocMeta: vi.fn(async () => {}), + getMeta: () => ({ scan: async () => [] }), + watch: (callback: typeof notifyMetadata) => { + notifyMetadata = callback; + return { unsubscribe() {} }; + }, + }, + getOrCreateSessionDoc, + onMetaRoomSynced: () => () => {}, + } as unknown as LoroDocumentManager, + executionService: { + getExecutionSnapshot: () => ({ + hasActiveTurn: false, + hasBlockingPendingCreate: false, + hasReusableSession: false, + }), + tryAcquireSessionRewriteConflictLease: () => () => {}, + dispatchPreparedSessionTurn: dispatch, + cancelSession: cancel, + } as unknown as SessionExecutionService, + }); + return { + watcher, + dispatch, + cancel, + getOrCreateSessionDoc, + getDocMeta, + metadataChanged: () => notifyMetadata({ kind: 'doc-metadata', docId: `session-${sessionId}` }), + replaceDoc: () => { + doc = makeDoc(); + return doc; + }, + setMeta: (next: SessionMeta) => { + meta = next; + }, + offer: () => + watcher.offerRpcTurn({ + sessionId, + userTurnId: 'rpc-1', + userId: 'user-1', + timestamp: new Date().toISOString(), + inputConfig: { prompt: 'Do the work' }, + }), + }; +} + +describe('watcher GC cleanup lease', () => { + it('resumes a metadata-only cancellation against the fresh document', async () => { + const f = fixture(); + await f.watcher.start(); + const release = f.watcher.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + try { + const record = await f.getDocMeta(); + f.setMeta({ ...record.meta, lastCanceledTurn: 'cancel-1' }); + f.metadataChanged(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(f.getOrCreateSessionDoc).not.toHaveBeenCalled(); + expect(f.cancel).not.toHaveBeenCalled(); + const fresh = f.replaceDoc(); + release?.(); + await vi.waitFor(() => expect(f.cancel).toHaveBeenCalledOnce()); + expect(fresh.mirror.subscribe).toHaveBeenCalledOnce(); + } finally { + f.watcher.stop(); + } + }); + + it('stashes RPC and metadata work until release, then opens the fresh document', async () => { + const f = fixture(); + await f.watcher.start(); + const oldUnsubscribe = vi.fn(); + const watched = f.watcher as unknown as { + watchedSessions: Map void }>; + }; + watched.watchedSessions.set(sessionId, { unsubscribe: oldUnsubscribe }); + const release = f.watcher.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + try { + await expect(f.offer()).resolves.toBe('accepted'); + f.metadataChanged(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(f.dispatch).not.toHaveBeenCalled(); + expect(f.getOrCreateSessionDoc).not.toHaveBeenCalled(); + expect(f.watcher.hasPendingDispatch(sessionId)).toBe(true); + const fresh = f.replaceDoc(); + release?.(); + release?.(); + expect(oldUnsubscribe).toHaveBeenCalledOnce(); + await vi.waitFor(() => expect(f.dispatch).toHaveBeenCalledTimes(1)); + expect(fresh.mirror.subscribe).toHaveBeenCalledOnce(); + } finally { + f.watcher.stop(); + } + }); + + it('does not reopen an idle document when releasing without pending work', async () => { + const f = fixture(); + await f.watcher.start(); + const release = f.watcher.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + release?.(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(f.getOrCreateSessionDoc).not.toHaveBeenCalled(); + f.watcher.stop(); + }); + + it('refuses queued metadata work and an active reconcile', async () => { + const f = fixture(); + await f.watcher.start(); + const read = deferred(); + const record = await f.getDocMeta(); + f.getDocMeta.mockClear(); + f.getDocMeta.mockImplementation(async () => { + await read.promise; + return record; + }); + f.metadataChanged(); + expect(f.watcher.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + await vi.waitFor(() => expect(f.getDocMeta).toHaveBeenCalled()); + expect(f.watcher.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + read.resolve(); + f.watcher.stop(); + }); + + it('refuses an already accepted RPC payload even without a running check', async () => { + const f = fixture(); + vi.spyOn(f.watcher, 'enqueueSessionCheck').mockResolvedValue(undefined); + await expect(f.offer()).resolves.toBe('accepted'); + expect(f.watcher.hasPendingDispatch(sessionId)).toBe(true); + expect(f.watcher.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + f.watcher.stop(); + }); + + it.each(['dispatch', 'cancel'] as const)('refuses an existing %s check', async (kind) => { + const f = fixture(); + const gate = deferred(); + const internals = f.watcher as unknown as { + maybeHandleSession: () => Promise; + maybeHandleCancelRequest: () => Promise; + enqueueCancelCheck: (id: typeof sessionId) => Promise; + }; + if (kind === 'dispatch') + vi.spyOn(internals, 'maybeHandleSession').mockReturnValue(gate.promise); + else vi.spyOn(internals, 'maybeHandleCancelRequest').mockReturnValue(gate.promise); + const pending = + kind === 'dispatch' + ? f.watcher.enqueueSessionCheck(sessionId) + : internals.enqueueCancelCheck(sessionId); + expect(f.watcher.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + gate.resolve(); + await pending; + const release = f.watcher.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + release?.(); + f.watcher.stop(); + }); + + it('defers cancellation checks until release', async () => { + const f = fixture(); + const internals = f.watcher as unknown as { + maybeHandleCancelRequest: () => Promise; + enqueueCancelCheck: (id: typeof sessionId) => Promise; + }; + const cancel = vi.spyOn(internals, 'maybeHandleCancelRequest').mockResolvedValue(undefined); + const release = f.watcher.tryAcquireGCCleanupLease(sessionId); + await internals.enqueueCancelCheck(sessionId); + expect(cancel).not.toHaveBeenCalled(); + release?.(); + await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce()); + f.watcher.stop(); + }); + + it('does not resume deferred work after stop or let an old release clear a new lease', async () => { + const f = fixture(); + await f.watcher.start(); + const oldRelease = f.watcher.tryAcquireGCCleanupLease(sessionId); + await f.offer(); + f.watcher.stop(); + oldRelease?.(); + expect(f.dispatch).not.toHaveBeenCalled(); + await f.watcher.start(); + const newRelease = f.watcher.tryAcquireGCCleanupLease(sessionId); + expect(newRelease).not.toBeNull(); + oldRelease?.(); + expect(f.watcher.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + newRelease?.(); + f.watcher.stop(); + }); +}); diff --git a/apps/cli/tests/session-execution-service.test.ts b/apps/cli/tests/session-execution-service.test.ts index 20170acd4..85f714951 100644 --- a/apps/cli/tests/session-execution-service.test.ts +++ b/apps/cli/tests/session-execution-service.test.ts @@ -210,6 +210,57 @@ const createBaseDeps = ( }; describe('SessionExecutionService', () => { + it.each(['startSession', 'continueSession'] as const)( + '%s waits for GC before document access and excludes GC until its failure unwinds', + async (method) => { + const documentGate = createDeferred(); + const documentOpened = createDeferred(); + const getOrCreateSessionDoc = vi.fn(async () => { + documentOpened.resolve(); + await documentGate.promise; + throw new Error('document unavailable'); + }); + const deps = createBaseDeps({ + workspaceDocument: { getOrCreateSessionDoc } as unknown as LoroDocumentManager, + }); + const service = new SessionExecutionService(deps); + const sessionId = 'gc-direct-operation' as SessionId; + const release = service.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + const operation = service[method]({ sessionId, acpSessionConfig: {} } as Parameters< + typeof service.startSession + >[0] & + Parameters[0]); + const result = expect(operation).rejects.toThrow('document unavailable'); + await Promise.resolve(); + expect(getOrCreateSessionDoc).not.toHaveBeenCalled(); + release!(); + release!(); + await documentOpened.promise; + expect(service.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + documentGate.resolve(); + await result; + const next = service.tryAcquireGCCleanupLease(sessionId); + expect(next).not.toBeNull(); + next!(); + } + ); + + it('reserves direct operation ownership synchronously before its first continuation', async () => { + const deps = createBaseDeps({}); + vi.mocked(deps.workspaceDocument.getOrCreateSessionDoc).mockRejectedValue(new Error('no doc')); + const service = new SessionExecutionService(deps); + const sessionId = 'gc-before-continuation' as SessionId; + const operation = service.continueSession({ sessionId } as Parameters< + typeof service.continueSession + >[0]); + expect(service.tryAcquireGCCleanupLease(sessionId)).toBeNull(); + await expect(operation).rejects.toThrow('no doc'); + const release = service.tryAcquireGCCleanupLease(sessionId); + expect(release).not.toBeNull(); + release!(); + }); + it('advances one session owner through consecutive prompt handoffs', async () => { const steerPrompt = vi.fn(() => ({ completion: new Promise(() => {}), diff --git a/apps/cli/tests/session-terminate-cleanup.test.ts b/apps/cli/tests/session-terminate-cleanup.test.ts index e5aef3c0c..82f3f3ec3 100644 --- a/apps/cli/tests/session-terminate-cleanup.test.ts +++ b/apps/cli/tests/session-terminate-cleanup.test.ts @@ -76,6 +76,26 @@ function createProcessHandle(terminate: SessionProcessHandle['terminate']): Sess } describe('Session terminate cleanup', () => { + it('reports failed cleanup and a new retry attempt without inferring success from status', async () => { + const session = createSession(); + session.acpSessionId = 'acp-session-1' as ACPSessionId; + session.terminalManager = createTerminalManager({ + disposeAll: vi + .fn() + .mockRejectedValueOnce(new Error('synthetic disposal failure')) + .mockResolvedValue(undefined), + }); + expect((await session.getMonitorRuntimeInfo()).cleanup).toBeNull(); + const failed = session.terminate(true); + expect((await session.getMonitorRuntimeInfo()).cleanup?.state).toBe('running'); + await expect(failed).rejects.toThrow('Session process termination failed'); + const receipt = (await session.getMonitorRuntimeInfo()).cleanup; + expect(receipt?.state).toBe('failed'); + await session.terminate(true); + expect((await session.getMonitorRuntimeInfo()).cleanup?.state).toBe('completed'); + expect(receipt?.state).toBe('failed'); + }); + it('shares pending termination and upgrades force without waiting for terminal disposal', async () => { const session = createSession(); let finishDisposal = () => {}; diff --git a/apps/cli/tests/terminal-manager.test.ts b/apps/cli/tests/terminal-manager.test.ts index 681a3cc40..51c40fa44 100644 --- a/apps/cli/tests/terminal-manager.test.ts +++ b/apps/cli/tests/terminal-manager.test.ts @@ -106,16 +106,18 @@ describe('ShellTerminalManager', () => { }); }); -function releaseFixture(handles: SessionProcessHandle[]) { +function releaseFixture(handles: SessionProcessHandle[], spawn?: SessionSandbox['spawn']) { const sandbox: SessionSandbox = { enabled: false, description: 'test', applyLimits: async () => {}, - spawn: vi.fn(async () => { - const handle = handles.shift(); - if (!handle) throw new Error('no handle'); - return handle; - }), + spawn: + spawn ?? + vi.fn(async () => { + const handle = handles.shift(); + if (!handle) throw new Error('no handle'); + return handle; + }), terminate: async () => {}, cleanup: async () => {}, }; @@ -344,3 +346,96 @@ it('finishes disposal when a pending terminal launch rejects', async () => { failSpawn(new Error('launch failed')); await Promise.all([start, disposal]); }); + +it('protects a pending terminal start and live watch until its observed exit', async () => { + const owned = observedHandle(); + let finishStart: (handle: SessionProcessHandle) => void = () => { + throw new Error('Start promise not initialized'); + }; + const starting = new Promise((resolve) => { + finishStart = resolve; + }); + const manager = releaseFixture([], () => starting); + expect(manager.hasRunningTerminals()).toBe(false); + + const creation = manager.createTerminal('acp-1', 'node', ['--watch', 'server.js']); + expect(manager.hasRunningTerminals()).toBe(true); + finishStart(owned.handle); + const id = await creation; + expect(manager.hasRunningTerminals()).toBe(true); + expect((await manager.terminalOutput('acp-1', id)).exitStatus).toBeNull(); + + owned.exit(0); + expect(manager.hasRunningTerminals()).toBe(false); + // The terminal remains addressable for output. Its mere presence cannot pin GC. + expect((await manager.terminalOutput('acp-1', id)).exitStatus).toEqual({ + exitCode: 0, + signal: undefined, + }); + expect(owned.unsubscribe).not.toHaveBeenCalled(); +}); + +it('clears the background guard when terminal startup rejects', async () => { + let rejectStart: (error: Error) => void = () => { + throw new Error('Start promise not initialized'); + }; + const starting = new Promise((_, reject) => { + rejectStart = reject; + }); + const manager = releaseFixture([], () => starting); + const creation = manager.createTerminal('acp-1', 'watch'); + const rejected = expect(creation).rejects.toThrow('spawn rejected'); + expect(manager.hasRunningTerminals()).toBe(true); + rejectStart(new Error('spawn rejected')); + await rejected; + expect(manager.hasRunningTerminals()).toBe(false); +}); + +it('keeps another pending start protected when one terminal startup fails', async () => { + const owned = observedHandle(); + let finishStart: (handle: SessionProcessHandle) => void = () => { + throw new Error('Start promise not initialized'); + }; + const starting = new Promise((resolve) => { + finishStart = resolve; + }); + const manager = releaseFixture([], async (command) => { + if (command === 'bad-watch') throw new Error('spawn rejected'); + return starting; + }); + const creation = manager.createTerminal('acp-1', 'good-watch'); + await expect(manager.createTerminal('acp-1', 'bad-watch')).rejects.toThrow('spawn rejected'); + expect(manager.hasRunningTerminals()).toBe(true); + finishStart(owned.handle); + await creation; + expect(manager.hasRunningTerminals()).toBe(true); + owned.exit(0); + expect(manager.hasRunningTerminals()).toBe(false); +}); + +it('protects live terminal work through MessageHandler before consulting empty history', async () => { + const { MessageHandler } = await import('../src/lib/message-handler'); + const { SessionIdSchema } = await import('@lody/shared'); + const owned = observedHandle(); + const terminalManager = releaseFixture([owned.handle]); + const id = await terminalManager.createTerminal('acp-1', 'watch'); + const getHistory = vi.fn(async () => []); + const runtime = { terminalManager }; + const receiver = { + sessionManager: { getSession: () => runtime }, + workspaceDocument: { + getOrCreateSessionDoc: async () => ({ getHistory, getMetaState: async () => null }), + }, + }; + const sessionId = SessionIdSchema.parse('watch-session'); + await expect(MessageHandler.prototype.hasProtectedWork.call(receiver, sessionId)).resolves.toBe( + true + ); + expect(getHistory).not.toHaveBeenCalled(); + owned.exit(0); + await expect(MessageHandler.prototype.hasProtectedWork.call(receiver, sessionId)).resolves.toBe( + false + ); + expect(getHistory).toHaveBeenCalledOnce(); + expect((await terminalManager.terminalOutput('acp-1', id)).exitStatus?.exitCode).toBe(0); +}); diff --git a/packages/shared/src/local-machine-rpc.ts b/packages/shared/src/local-machine-rpc.ts index 98264f1b8..fd388ab9c 100644 --- a/packages/shared/src/local-machine-rpc.ts +++ b/packages/shared/src/local-machine-rpc.ts @@ -1,4 +1,5 @@ import { z } from 'zod'; +import { MachineResourceHistorySchema } from './machine-monitor'; import { CodeCollabV2ErrorSchema, CodeCollabV2FileIndexRequestSchema, @@ -74,6 +75,10 @@ export type SessionActiveInvocationContextResult = z.infer< >; export const LocalMachineRpcRequestSchema = z.discriminatedUnion('method', [ + BaseLocalMachineRpcRequestSchema.extend({ + method: z.literal('machine/get-resource-history'), + params: z.object({}).strict(), + }).strict(), BaseLocalMachineRpcRequestSchema.extend({ method: z.literal('session/get-active-invocation-context'), params: z @@ -235,6 +240,7 @@ export type LocalMachineRpcRequest = z.infer; export type MachineMonitorResourceUsage = { memoryBytes: number | null; @@ -118,20 +120,69 @@ const MachineMonitorResourceUsageSchema = z.object({ quality: z.enum(['exact-process', 'exact-cgroup', 'estimated-tree', 'unavailable']), }); +export const MachineResourceHistorySchema = z + .object({ + type: z.literal('machine/resource-history'), + machineId: z.string().min(1), + instanceId: z.string().min(1), + collectedWhileObserved: z.literal(true), + samples: z + .array( + z + .object({ + sampledAtMs: z.number().finite().nonnegative(), + source: z.enum(['available', 'unavailable', 'not-sampled']), + cliControlPlane: MachineMonitorResourceUsageSchema.strict().nullable(), + memoryKind: z.enum(['rss-sum', 'physical-footprint-sum', 'working-set-sum']), + processesTruncated: z.boolean(), + sessionsTruncated: z.boolean(), + processes: z + .array( + z + .object({ + pid: z.number().int().positive(), + startedAtMs: z.number().finite().nonnegative(), + sessionId: z.string().nullable(), + memoryBytes: z.number().finite().nonnegative(), + cpuTimeMicros: z.number().finite().nonnegative(), + }) + .strict() + ) + .max(256), + sessions: z + .array( + z + .object({ + sessionId: z.string(), + parentSessionId: z.string().nullable(), + status: MachineMonitorSessionStatusSchema, + cleanup: z + .object({ + state: z.enum(['running', 'failed', 'completed']), + attemptedAtMs: z.number().finite().nonnegative(), + }) + .strict() + .nullable(), + resource: MachineMonitorResourceUsageSchema.strict(), + }) + .strict() + ) + .max(100), + }) + .strict() + ) + .max(120), + }) + .strict(); + +export type MachineResourceHistory = z.infer; + const AcpSessionMonitorSnapshotSchema = z.object({ sessionId: SessionIdSchema, parentSessionId: SessionIdSchema.nullable(), agentCliType: z.string().nullable(), agentType: z.string().nullable(), - status: z.enum([ - 'initializing', - 'running', - 'waiting_permission', - 'finalizing', - 'idle', - 'stopping', - 'failed', - ]), + status: MachineMonitorSessionStatusSchema, lastActivityAtMs: nullableFiniteNonNegative, startedAtMs: nullableFiniteNonNegative, resource: MachineMonitorResourceUsageSchema, diff --git a/packages/shared/src/machine-protocol-capabilities.ts b/packages/shared/src/machine-protocol-capabilities.ts index 288d45ab6..1b36cec2a 100644 --- a/packages/shared/src/machine-protocol-capabilities.ts +++ b/packages/shared/src/machine-protocol-capabilities.ts @@ -12,12 +12,14 @@ export const MACHINE_PROTOCOL_CAPABILITIES = { localProjectRemoval: 'localProjectRemoval', providerSetup: 'providerSetup', acpProtocolAuthentication: 'acpProtocolAuthentication', + resourceHistory: 'resourceHistory', } as const; export const ACP_AUTHENTICATION_INTERACTIONS_PROTOCOL_VERSION = 2; export const LOCAL_PROJECT_REMOVAL_PROTOCOL_VERSION = 1; export const PROVIDER_SETUP_PROTOCOL_VERSION = 1; export const ACP_PROTOCOL_AUTHENTICATION_VERSION = 2; +export const RESOURCE_HISTORY_PROTOCOL_VERSION = 1; type MachineProtocolCapabilityCarrier = { protocolCapabilities?: MachineProtocolCapabilities; @@ -52,8 +54,20 @@ export const CURRENT_MACHINE_PROTOCOL_CAPABILITIES: MachineProtocolCapabilities [MACHINE_PROTOCOL_CAPABILITIES.localProjectRemoval]: LOCAL_PROJECT_REMOVAL_PROTOCOL_VERSION, [MACHINE_PROTOCOL_CAPABILITIES.providerSetup]: PROVIDER_SETUP_PROTOCOL_VERSION, [MACHINE_PROTOCOL_CAPABILITIES.acpProtocolAuthentication]: ACP_PROTOCOL_AUTHENTICATION_VERSION, + [MACHINE_PROTOCOL_CAPABILITIES.resourceHistory]: RESOURCE_HISTORY_PROTOCOL_VERSION, }; +/** Whether the daemon exposes recent observed resource samples over local Machine RPC. */ +export function machineSupportsResourceHistory( + machine: MachineProtocolCapabilityCarrier | null | undefined +): boolean { + return machineSupportsProtocolCapability( + machine, + MACHINE_PROTOCOL_CAPABILITIES.resourceHistory, + RESOURCE_HISTORY_PROTOCOL_VERSION + ); +} + /** Whether the target daemon supports interactive Custom/Registry ACP authentication. */ export function machineSupportsAcpAuthenticationInteractionsProtocol( machine: MachineProtocolCapabilityCarrier | null | undefined diff --git a/packages/shared/tests/machine-resource-history.test.ts b/packages/shared/tests/machine-resource-history.test.ts new file mode 100644 index 000000000..9d53482fa --- /dev/null +++ b/packages/shared/tests/machine-resource-history.test.ts @@ -0,0 +1,94 @@ +import { expect, it } from 'vitest'; +import { MachineResourceHistorySchema } from '../src/machine-monitor'; + +const sample = { + sampledAtMs: 1, + source: 'unavailable', + cliControlPlane: null, + memoryKind: 'rss-sum', + processesTruncated: false, + sessionsTruncated: false, + processes: [], + sessions: [], +}; +const receipt = { + type: 'machine/resource-history', + machineId: 'machine', + instanceId: 'instance', + collectedWhileObserved: true, + samples: [sample], +}; + +const resource = { + memoryBytes: 1, + cpuCores: 0, + cpuPercentOfMachine: 0, + processCount: 1, + memoryKind: 'rss-sum', + quality: 'estimated-tree', +}; +const session = { + sessionId: 'session', + parentSessionId: null, + status: 'idle', + cleanup: null, + resource, +}; + +it('accepts bounded observation receipts and rejects oversized arrays or raw diagnostics', () => { + expect(MachineResourceHistorySchema.safeParse(receipt).success).toBe(true); + expect( + MachineResourceHistorySchema.safeParse({ ...receipt, samples: Array(121).fill(sample) }).success + ).toBe(false); + expect( + MachineResourceHistorySchema.safeParse({ + ...receipt, + samples: [{ ...sample, commandLine: 'private' }], + }).success + ).toBe(false); + expect( + MachineResourceHistorySchema.safeParse({ ...receipt, collectedWhileObserved: false }).success + ).toBe(false); +}); + +it('rejects private fields nested in either resource object', () => { + const populated = { ...sample, cliControlPlane: resource, sessions: [session] }; + expect(MachineResourceHistorySchema.safeParse({ ...receipt, samples: [populated] }).success).toBe( + true + ); + for (const modified of [ + { ...populated, cliControlPlane: { ...resource, commandLine: 'private' } }, + { ...populated, sessions: [{ ...session, resource: { ...resource, environment: 'private' } }] }, + ]) { + expect( + MachineResourceHistorySchema.safeParse({ ...receipt, samples: [modified] }).success + ).toBe(false); + } +}); + +it('accepts only the monitor protocol session statuses', () => { + for (const status of [ + 'initializing', + 'running', + 'waiting_permission', + 'finalizing', + 'idle', + 'stopping', + 'failed', + ]) { + expect( + MachineResourceHistorySchema.safeParse({ + ...receipt, + samples: [{ ...sample, sessions: [{ ...session, status }] }], + }).success + ).toBe(true); + } + for (const status of ['', 'unknown', 'completed']) { + expect( + MachineResourceHistorySchema.safeParse({ + ...receipt, + samples: [{ ...sample, sessions: [{ ...session, status }] }], + }).success + ).toBe(false); + } +});