From 4dffbedbceca60d83b018503cfcc75350fcd2eb7 Mon Sep 17 00:00:00 2001 From: Ver-zhzh Date: Sun, 11 Oct 2026 07:29:35 +0000 Subject: [PATCH 1/2] =?UTF-8?q?fix(server):=20sidecar=20watched-idle=20?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E4=BA=8B=E4=BB=B6=E6=97=A5=E5=BF=97=E5=8F=AF?= =?UTF-8?q?=E6=B7=98=E6=B1=B0=EF=BC=88#445=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - evictLoadedBeyondCap：被 SSE 观察但静置超 idleAgentTtlMs 的会话允许淘汰 （此前一律跳过，长生命周期 SSE 标签页把完整内存环永久钉住）； 淘汰前 flushDeltaBuf/flushToolResultBuf，重连走 getEvents→ensureEvents 从磁盘重放，不丢事件；idleAgentTtlMs<=0 保持旧的被观察永不淘汰。 - evictLoadedBeyondCap(exceptId?)：ensureEvents 刚载入的会话不受同一次 淘汰扫描影响，避免刚读回的日志被同一调用立即丢弃。 - 新增回归测试 session-evict-watched-idle.test.ts（3 场景）。 注：本 PR 原含 maxEventBytes 64MB→8MB、effectiveAfterRestart 两项， 已由上游 abb7628（v3.30.0 republish）自行收编；rebase 到最新 main 后 diff 已收窄到只剩本项（按评审要求）。 Closes #445 --- .../session-evict-watched-idle.test.ts | 220 ++++++++++++++++++ src/server/session-manager.ts | 35 ++- 2 files changed, 248 insertions(+), 7 deletions(-) create mode 100644 src/server/__tests__/session-evict-watched-idle.test.ts diff --git a/src/server/__tests__/session-evict-watched-idle.test.ts b/src/server/__tests__/session-evict-watched-idle.test.ts new file mode 100644 index 000000000..8515bcb79 --- /dev/null +++ b/src/server/__tests__/session-evict-watched-idle.test.ts @@ -0,0 +1,220 @@ +import { test } from 'node:test' +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + RuntimeSessionManager, + type ManagedAgent, +} from '../session-manager.js' +import { FileSessionPersistence } from '../session-persistence.js' +import type { AgentCallbacks } from '../../agent/loop-types.js' +import type { Artifact } from '../../artifact/types.js' +import type { OaiMessage } from '../../api/oai-types.js' + +/** 照抄 src/server/__tests__/session-manager.test.ts 的 FakeAgent（同一契约： + * run 挂起到 finish()，callbacks 暴露给用例直接驱动事件）。 */ +class FakeAgent implements ManagedAgent { + callbacks?: AgentCallbacks + aborted = false + artifacts: Artifact[] = [] + /** Rewind: in-memory message store for testing. */ + messages: OaiMessage[] = [] + /** S — captures the mode this agent was built with + any live switches. */ + builtApprovalMode?: string + liveApprovalMode?: string + /** 每次 run 收到的 prompt，按序记录（含自动续跑注入的 'continue'）。 */ + prompts: string[] = [] + /** 累计用量快照——只有显式设值的用例才会让「中断补发」事件出现（默认 + * undefined，避免给既有用例的事件流凭空多一条 turn_complete）。 */ + usageSnapshot?: { + input_tokens?: number + output_tokens?: number + cache_read_input_tokens?: number + cache_creation_input_tokens?: number + } + usageTurns = 0 + getTotalUsage() { return this.usageSnapshot ?? {} } + getTurnCount() { return this.usageTurns } + /** 实时上下文占用（默认 0 = 不携带，避免扰动既有用例的事件形状）。 */ + estimatedTokens = 0 + getEstimatedTokens() { return this.estimatedTokens } + private resolveRun?: () => void + + run(prompt: string, cb: AgentCallbacks): Promise { + this.prompts.push(prompt) + this.callbacks = cb + return new Promise((res) => { this.resolveRun = res }) + } + finish(): void { this.resolveRun?.() } + abort(): void { + this.aborted = true + this.callbacks?.onAbort() + this.resolveRun?.() + } + /** 模拟 agent 内部 watchdog 自中止：带 reason 的 onAbort + run settle。 + * 与 abort()（用户中止，无 reason）区分——manager.abort 不走这条路。 */ + watchdogAbort(reason = 'watchdog:goal'): void { + this.callbacks?.onAbort(reason) + this.resolveRun?.() + } + /** 模拟流错误（超时/网络）终结本 run：onError + onStreamInterrupted + run settle。 + * 区别于 abort()（用户中止→aborted）与 finish()（成功→completed）。 */ + streamInterrupt(message = 'Request timed out: server did not respond within 45 seconds'): void { + const err = new Error(message) + this.callbacks?.onError(err) + this.callbacks?.onStreamInterrupted?.(err) + this.resolveRun?.() + } + setApprovalMode(mode: string): void { this.liveApprovalMode = mode } + listArtifacts(): Artifact[] { return this.artifacts } + readArtifact(id: string): Promise { + return Promise.resolve(this.artifacts.some((a) => a.id === id) ? `raw:${id}` : null) + } + getMessages(): OaiMessage[] { return this.messages } + replaceMessages(msgs: OaiMessage[]): void { this.messages = msgs } + rewindToMessages(msgs: OaiMessage[]): void { this.messages = msgs } + /** P0-2: optional — plan_task onToolResult reads this to emit todo_state */ + getTodos?: () => Array<{ id: string; content: string; status: string }> +} + +/** + * issue #445 / PR #447 返工：watched-idle 淘汰差集。 + * + * evictLoadedBeyondCap 在 LRU 超出 maxLoadedSessions 时淘汰日志，但此前对 + * 「被观察（listeners>0）且空闲」的会话一律跳过——长生命周期 SSE 标签页会把 + * 完整内存环永久钉住（内存泄漏）。修复后：被观察会话一旦静置超过 idleAgentTtlMs + * 就与普通未观察会话一样可淘汰（事件已落盘，重连走 getEvents→ensureEvents 从 + * 磁盘重放，不丢事件）；idleAgentTtlMs<=0 关闭该路径，保持旧的「被观察永不淘汰」。 + */ +function makeEvictManager(ttlMs: number) { + const root = mkdtempSync(join(tmpdir(), 'evict-watched-')) + let now = 1_000_000 + const agents: FakeAgent[] = [] + const persistence = new FileSessionPersistence(join(root, 'sessions')) + const manager = new RuntimeSessionManager({ + createAgent: () => { const a = new FakeAgent(); agents.push(a); return a as any }, + defaultCwd: join(root, 'work'), + persistence, + maxLoadedSessions: 2, + idleAgentTtlMs: ttlMs, + watchdogContinueDelayMs: 0, + now: () => now, + }) + return { + root, + manager, + agents, + persistence, + advance: (ms: number) => { now += ms }, + cleanup() { rmSync(root, { recursive: true, force: true }) }, + } +} + +/** 内部状态探针：仓内既有用例同样用 as any 触达 InternalSession。 */ +const sessionOf = ( + manager: RuntimeSessionManager, + id: string, +): { eventsLoaded: boolean; events: Array<{ type: string; data: unknown }> } => + (manager as unknown as { sessions: Map }).sessions.get(id) as never + +test('超 TTL 的被观察会话被淘汰且事件可从磁盘重放', async () => { + const f = makeEvictManager(1000) + try { + // 1. 建一个跑过又静置的会话，且被观察（SSE listener）。 + const rec = f.manager.createSession({ prompt: 'task one' }) + const agent = f.agents[0]! + agent.callbacks!.onDelegationActivity!({ + workOrderId: 'w1', + status: 'completed', + parentToolId: 'tool', + progressLine: 'DATA-ONE', + }) + agent.finish() + await f.manager.waitForRunSettled(rec.id) + const unsub = f.manager.subscribe(rec.id, () => {}) + assert.ok(unsub, 'subscribe 必须返回 unsubscribe') + + // 2. 事件已落到磁盘（淘汰后的重放源）。事件写链是异步排空的， + // 同步 flushSync 覆盖不了在途写链——用 flushThrough 等到磁盘高水位 + // ≥ 内存最新 seq（pr353 同款模式），再断言落盘。 + await f.persistence.flushThrough(rec.id, f.manager.getReplayWindow(rec.id)!.diskLastSeq) + assert.ok( + f.persistence.loadEvents(rec.id).some( + (e) => e.type === 'delegation' && (e.data as any).workerId === 'w1', + ), + 'delegation 事件必须先落盘', + ) + + // 3. 静置超过 TTL，sweep 释放已建 agent：会话变为无 agent、未运行、被观察、超 TTL。 + f.advance(1001) + f.manager.sweepIdleAgents() + + // 4. 制造 LRU 溢出：loadedOrder=[s1,s2,s3] 长度 3 > maxLoadedSessions 2。 + f.manager.createSession({}) + const s3 = f.manager.createSession({}) + f.manager.getEvents(s3.id) // 触发 ensureEvents → touchLoaded → evictLoadedBeyondCap + + // 5. 超 TTL 的被观察会话必须被淘汰。 + assert.equal(sessionOf(f.manager, rec.id).eventsLoaded, false, 'watched-idle 超 TTL 必须淘汰') + assert.equal(sessionOf(f.manager, rec.id).events.length, 0) + + // 6. 淘汰后从磁盘重放，事件无丢失。 + const replayed = f.manager.getEvents(rec.id)! + assert.ok( + replayed.events.some((e) => e.type === 'delegation' && (e.data as any).workerId === 'w1'), + '淘汰后重放不得丢失 delegation 事件', + ) + assert.equal(sessionOf(f.manager, rec.id).eventsLoaded, true, '重放后回到内存常驻') + + unsub!() + } finally { + f.cleanup() + } +}) + +test('未超 TTL 的被观察会话不淘汰', async () => { + const f = makeEvictManager(60_000) + try { + const s1 = f.manager.createSession({}) + const unsub = f.manager.subscribe(s1.id, () => {}) + + const s2 = f.manager.createSession({}) + const s3 = f.manager.createSession({}) + f.manager.getEvents(s3.id) + + // 被观察且未超 TTL：必须跳过、保留在内存。 + assert.equal(sessionOf(f.manager, s1.id).eventsLoaded, true, '被观察且未超 TTL 不得淘汰') + // 对照：淘汰逻辑确实跑了——淘汰的是无观察的 LRU 会话。 + assert.equal(sessionOf(f.manager, s2.id).eventsLoaded, false, '无观察的 LRU 会话应被淘汰') + + unsub!() + } finally { + f.cleanup() + } +}) + +test('idleAgentTtlMs<=0 时被观察会话永不淘汰', async () => { + const f = makeEvictManager(0) + try { + const s1 = f.manager.createSession({}) + const unsub = f.manager.subscribe(s1.id, () => {}) + + // 远超任何 TTL，模拟长静置。 + f.advance(3_600_000) + + const s2 = f.manager.createSession({}) + const s3 = f.manager.createSession({}) + f.manager.getEvents(s3.id) + + // ttl<=0 时 watched-idle 淘汰路径关闭，永不淘汰被观察会话。 + assert.equal(sessionOf(f.manager, s1.id).eventsLoaded, true, 'ttl<=0 时被观察会话永不淘汰') + // 对照：淘汰逻辑仍在跑（淘汰无观察会话）。 + assert.equal(sessionOf(f.manager, s2.id).eventsLoaded, false, '无观察的 LRU 会话应被淘汰') + + unsub!() + } finally { + f.cleanup() + } +}) diff --git a/src/server/session-manager.ts b/src/server/session-manager.ts index 999ce2900..93eab8425 100644 --- a/src/server/session-manager.ts +++ b/src/server/session-manager.ts @@ -2009,7 +2009,7 @@ export class RuntimeSessionManager { this.sweepStaleDelegationNodes(session, 'caller_aborted') } this.touchLoaded(session) - this.evictLoadedBeyondCap() + this.evictLoadedBeyondCap(session.record.id) } /** Fold a freshly-read on-disk log into the session (shared by the sync and @@ -2105,7 +2105,7 @@ export class RuntimeSessionManager { await session.eventsLoadPromise } this.touchLoaded(session) - this.evictLoadedBeyondCap() + this.evictLoadedBeyondCap(session.record.id) } /** Async twin of getEvents — reconnect/replay entry point for HTTP routes. */ @@ -2260,16 +2260,37 @@ export class RuntimeSessionManager { /** * Drop event logs of idle LRU sessions to bound resident memory. Never unloads - * a session that's live (agent built or running) or being watched (SSE - * listeners) — its in-memory log is the source of truth for in-flight appends - * and replay; those reload cleanly from disk once idle. + * a session that's live (agent built or running). Watched-but-idle sessions + * (SSE listeners, no agent/run) are the issue #445 leak: listeners only drop + * when the client disconnects, so a couple of long-lived tabs pin their full + * ring forever. Once such a log is quiescent past the idle TTL its events are + * provably on disk (every append is flushed through persistence.appendEvent), + * so it may be dropped like any unwatched session — a reconnect replays from + * disk via getEvents→ensureEvents (the same path an evicted unwatched log + * uses), so no event is silently lost. idleAgentTtlMs <= 0 disables the path + * (preserving the old "never evict a watched session" behavior). + * + * `exceptId` shields the session being opened by the current ensureEvents + * call: without it a just-loaded watched-idle log would be dropped by the + * same call that read it, and getEvents would hand back an empty log. */ - private evictLoadedBeyondCap(): void { + private evictLoadedBeyondCap(exceptId?: string): void { let i = 0 while (i < this.loadedOrder.length && this.loadedOrder.length > this.maxLoadedSessions) { const id = this.loadedOrder[i]! const s = this.sessions.get(id) - if (!s || s.agent || s.running || s.listeners.size > 0) { i++; continue } + if (!s || s.agent || s.running) { i++; continue } + if (id === exceptId) { i++; continue } + const watched = s.listeners.size > 0 + if (watched && (this.idleAgentTtlMs <= 0 || this.now() - s.record.updatedAt < this.idleAgentTtlMs)) { + i++ + continue + } + // Unflushed coalesced deltas live only in memory — flush before dropping + // so the replay they came from keeps every event (same guard unloadSession + // uses). + this.flushDeltaBuf(s) + this.flushToolResultBuf(s) s.events = [] s.knownArtifacts = new Set() s.eventsLoaded = false From 107a56d7f9612174b44a826b143499ebda721bd2 Mon Sep 17 00:00:00 2001 From: Ver-zhzh Date: Sun, 11 Oct 2026 22:59:44 +0000 Subject: [PATCH 2/2] =?UTF-8?q?fix(server):=20=E6=B7=98=E6=B1=B0=E7=9A=84?= =?UTF-8?q?=E8=A2=AB=E8=A7=82=E5=AF=9F=E4=BC=9A=E8=AF=9D=E4=BB=8D=E5=8F=AF?= =?UTF-8?q?=E7=BB=8F=E6=BC=82=E7=A7=BB=E5=90=88=E5=B9=B6=E6=8E=A5=E6=94=B6?= =?UTF-8?q?=E5=A4=96=E9=83=A8=E8=BF=9B=E5=BA=A6=EF=BC=88#445=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit evictLoadedBeyondCap 淘汰 watched-idle 会话后 eventsLoaded=false， 但 adoptExternalSessions 的漂移合并与 mergeExternalEvents 的 await 后守卫都只看 eventsLoaded，导致活着的 SSE listener 永远 收不到外部新事件（session-rehydrate LRU 压力测试回归）。 放宽两处资格判定为 eventsLoaded || listeners.size > 0： 无 listener 的已淘汰会话仍不参与合并（内存继续释放）； 只合并 seq > session.seq 的尾部新事件，不全量常驻； getEvents→ensureEvents 磁盘重放路径未动。 --- src/server/session-manager.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/server/session-manager.ts b/src/server/session-manager.ts index 93eab8425..0dda3cf1c 100644 --- a/src/server/session-manager.ts +++ b/src/server/session-manager.ts @@ -1663,7 +1663,13 @@ export class RuntimeSessionManager { continue } if (this.syncExternalRecord(existing, rawRecord)) recordMerged = true - if (!existing.running && existing.eventsLoaded && !existing.tombstoned) { + // 漂移合并资格：已加载的会话（原有），或**有活 listener 的已淘汰会话**。 + // issue #445 rework 后 watched-idle 超 TTL 会被淘汰（eventsLoaded=false + // 但 listeners 保留），若此处仍只看 eventsLoaded，活着的 SSE 连接永远 + // 收不到外部新事件（回归）。无 listener 的已淘汰会话保持原样：内存已 + // 释放，无外部进度时不做任何重读。 + const mergeEligible = existing.eventsLoaded || existing.listeners.size > 0 + if (!existing.running && mergeEligible && !existing.tombstoned) { const merge = this.maybeSyncExternalEvents(existing) if (merge) eventMerges.push(merge) } @@ -1752,8 +1758,12 @@ export class RuntimeSessionManager { // 水位只在读成功后前进——读失败下轮重试 session.externalSyncSize = stat.size session.externalSyncMtimeMs = stat.mtimeMs - // 等待期间会话被逐出/卸载/替换 → 丢弃这份陈旧磁盘快照（同 ensureEventsAsync 的守卫） - if (this.sessions.get(id) !== session || !session.eventsLoaded || session.tombstoned) return + // 等待期间会话被逐出/卸载/替换 → 丢弃这份陈旧磁盘快照（同 ensureEventsAsync 的守卫）。 + // 资格判定与 adoptExternalSessions 对齐：已加载，或有活 listener 的已淘汰会话 + // （evicted-watched：eventsLoaded=false 但 listeners 保留）都仍是有效合并目标； + // 二者皆无才是真正失效的快照。novel 过滤（seq > session.seq）保证只合并尾部 + // 新事件，不把全量历史重新常驻内存。 + if (this.sessions.get(id) !== session || (!session.eventsLoaded && session.listeners.size === 0) || session.tombstoned) return if (artifactIds) for (const a of artifactIds) session.knownArtifacts.add(a) if (diskFirst !== undefined && diskFirst > (session.diskFirstSeq ?? 1)) session.diskFirstSeq = diskFirst const novel = candidates.filter((e) => e.seq > session.seq).sort((a, b) => a.seq - b.seq)