From aa874e66f72c218b62ab83b978af4d98507b6bc0 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 19:05:36 +0800 Subject: [PATCH 01/21] feat(channels): add lazy session route recovery --- .../channels/base/src/SessionRouter.test.ts | 211 ++++++++++++++ packages/channels/base/src/SessionRouter.ts | 272 +++++++++++++++--- 2 files changed, 447 insertions(+), 36 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index de04d003e6d..368aebcdb2c 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -214,6 +214,12 @@ describe('SessionRouter', () => { expect(bridge.newSession).toHaveBeenCalledWith('/default'); }); + it('uses defaultCwd when cwd is empty', async () => { + const router = new SessionRouter(bridge, '/default'); + await router.resolve('ch', 'alice', 'chat1', undefined, ''); + expect(bridge.newSession).toHaveBeenCalledWith('/default'); + }); + it('deduplicates concurrent session creation for the same route', async () => { let resolveNewSession!: (sessionId: string) => void; const newSession = vi.fn( @@ -240,6 +246,31 @@ describe('SessionRouter', () => { expect(newSession).toHaveBeenCalledTimes(1); }); + it('reserves a route before synchronously entering the bridge', async () => { + let calls = 0; + let reentered = false; + let nested!: Promise; + const router = new SessionRouter(mockBridge(), '/default'); + const newSession = vi.fn(() => { + const sessionId = `session-${++calls}`; + if (!reentered) { + reentered = true; + nested = router.resolve('ch', 'alice', 'chat1'); + } + return sessionId; + }); + router.setBridge({ ...mockBridge(), newSession }); + + const first = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + + await expect(Promise.all([first, nested])).resolves.toEqual([ + 'session-1', + 'session-1', + ]); + expect(newSession).toHaveBeenCalledTimes(1); + }); + it('retries if a new session dies before the route is stored', async () => { let calls = 0; const router = new SessionRouter(mockBridge(), '/default'); @@ -966,4 +997,184 @@ describe('SessionRouter', () => { expect(bridge.newSession).not.toHaveBeenCalled(); }); }); + + describe('lazy recovery', () => { + function createLazyRouter(persistPath: string, customBridge = bridge) { + return new SessionRouter(customBridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + } + + it('restores route metadata without loading daemon sessions', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const router = createLazyRouter(persistPath); + + expect(router.restoreRoutes()).toEqual({ restored: 1, dropped: 0 }); + expect(bridge.loadSession).not.toHaveBeenCalled(); + expect(router.getSession('ch', 'alice', 'chat1')).toBe('old-session'); + }); + + it('loads a dormant route once and then reuses the live binding', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const router = createLazyRouter(persistPath); + router.restoreRoutes(); + + await expect(router.resolve('ch', 'alice', 'chat1')).resolves.toBe( + 'old-session', + ); + await expect(router.resolve('ch', 'alice', 'chat1')).resolves.toBe( + 'old-session', + ); + expect(bridge.loadSession).toHaveBeenCalledTimes(1); + expect(bridge.newSession).not.toHaveBeenCalled(); + }); + + it('coalesces concurrent loads for one dormant route', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + let finishLoad!: (value: string) => void; + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn( + () => + new Promise((resolve) => { + finishLoad = resolve; + }), + ), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + const first = router.resolve('ch', 'alice', 'chat1'); + const second = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + finishLoad('old-session'); + + await expect(Promise.all([first, second])).resolves.toEqual([ + 'old-session', + 'old-session', + ]); + expect(lazyBridge.loadSession).toHaveBeenCalledTimes(1); + }); + + it('replaces a route only after fallback creation succeeds', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn().mockRejectedValue(new Error('gone')), + newSession: vi.fn().mockResolvedValue('replacement-session'), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + await expect(router.resolve('ch', 'alice', 'chat1')).resolves.toBe( + 'replacement-session', + ); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({ + 'ch:alice:chat1': expect.objectContaining({ + sessionId: 'replacement-session', + }), + }); + }); + + it('retains the dormant route when load and fallback creation both fail', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn().mockRejectedValue(new Error('temporarily gone')), + newSession: vi.fn().mockRejectedValue(new Error('at capacity')), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + await expect(router.resolve('ch', 'alice', 'chat1')).rejects.toThrow( + 'at capacity', + ); + expect(router.getSession('ch', 'alice', 'chat1')).toBe('old-session'); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({ + 'ch:alice:chat1': expect.objectContaining({ sessionId: 'old-session' }), + }); + }); + + it('marks a dead lazy session dormant and reloads it on next resolve', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const router = createLazyRouter(persistPath); + router.restoreRoutes(); + await router.resolve('ch', 'alice', 'chat1'); + + expect(router.handleSessionDied('old-session')).toBe(true); + expect(router.hasSession('ch', 'alice', 'chat1')).toBe(true); + await router.resolve('ch', 'alice', 'chat1'); + + expect(bridge.loadSession).toHaveBeenCalledTimes(2); + }); + + it('does not eagerly load route counts above the daemon live-session cap', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const entries = Object.fromEntries( + Array.from({ length: 25 }, (_, index) => [ + `ch:user-${index}:chat-${index}`, + { + sessionId: `old-${index}`, + target: { + channelName: 'ch', + senderId: `user-${index}`, + chatId: `chat-${index}`, + }, + cwd: '/tmp', + }, + ]), + ); + writeFileSync(persistPath, JSON.stringify(entries)); + const router = createLazyRouter(persistPath); + + expect(router.restoreRoutes()).toEqual({ restored: 25, dropped: 0 }); + expect(bridge.loadSession).not.toHaveBeenCalled(); + }); + + it('clears a dormant route destructively', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const router = createLazyRouter(persistPath); + router.restoreRoutes(); + + expect(router.removeSession('ch', 'alice', 'chat1')).toEqual([ + 'old-session', + ]); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({}); + await expect(router.resolve('ch', 'alice', 'chat1')).resolves.toBe( + 'session-1', + ); + expect(bridge.loadSession).not.toHaveBeenCalled(); + }); + + it('keeps eager session-death behavior as the default', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const sessionId = await router.resolve('ch', 'alice', 'chat1'); + + expect(router.handleSessionDied(sessionId)).toBe(true); + expect(router.hasSession('ch', 'alice', 'chat1')).toBe(false); + }); + }); }); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 249a0745417..690c773c457 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -21,12 +21,19 @@ interface ResolveOptions { routingThreadId?: string; } +export type SessionRecoveryMode = 'eager' | 'lazy'; + +export interface SessionRouterOptions { + recoveryMode?: SessionRecoveryMode; +} + export class SessionRouter { private toSession: Map = new Map(); // routing key → session ID private toTarget: Map = new Map(); // session ID → target private toCwd: Map = new Map(); // session ID → cwd private creatingSessions: Map> = new Map(); private sessionLoadWindows: Set = new Set(); + private readonly liveSessionIds = new Set(); private bridge: ChannelAgentBridge; private defaultCwd: string; @@ -34,17 +41,20 @@ export class SessionRouter { private channelScopes: Map = new Map(); private channelApprovalModes: Map = new Map(); private persistPath: string | undefined; + private readonly recoveryMode: SessionRecoveryMode; constructor( bridge: ChannelAgentBridge, defaultCwd: string, scope: SessionScope = 'user', persistPath?: string, + options: SessionRouterOptions = {}, ) { this.bridge = bridge; this.defaultCwd = defaultCwd; this.defaultScope = scope; this.persistPath = persistPath; + this.recoveryMode = options.recoveryMode ?? 'eager'; } /** Replace the bridge instance (used after crash recovery restart). */ @@ -108,10 +118,18 @@ export class SessionRouter { chatId, options?.routingThreadId ?? threadId, ); - let failedCreateWaits = 0; + const input = { + channelName, + senderId, + chatId, + threadId, + cwd: cwd || this.defaultCwd, + isGroup, + }; + let failedWaits = 0; for (;;) { const existing = this.toSession.get(key); - if (existing) { + if (existing && this.isLive(existing)) { this.promoteTargetToGroup(existing, isGroup); return existing; } @@ -122,53 +140,145 @@ export class SessionRouter { const sessionId = await creating; this.promoteTargetToGroup(sessionId, isGroup); return sessionId; - } catch (err) { + } catch (error) { if (this.creatingSessions.get(key) === creating) { this.creatingSessions.delete(key); } - failedCreateWaits++; - if (failedCreateWaits > 3) { - throw err; - } + failedWaits++; + if (failedWaits > 3) throw error; continue; } } - // Register the in-flight route before starting newSession(), because a - // bridge can emit sessionDied synchronously while creating the session. - const created = Promise.resolve().then(async () => { - const sessionCwd = cwd || this.defaultCwd; - const loadWindow = this.beginSessionLoad(); + const operation = Promise.resolve().then(() => + existing + ? this.loadOrReplaceSession(key, existing, input) + : this.createAndStoreSession(key, input), + ); + this.creatingSessions.set(key, operation); + try { + const sessionId = await operation; + this.promoteTargetToGroup(sessionId, isGroup); + return sessionId; + } finally { + if (this.creatingSessions.get(key) === operation) { + this.creatingSessions.delete(key); + } + } + } + } + + private isLive(sessionId: string): boolean { + return this.recoveryMode === 'eager' || this.liveSessionIds.has(sessionId); + } + + private async createAndStoreSession( + key: string, + input: { + channelName: string; + senderId: string; + chatId: string; + threadId?: string; + cwd: string; + isGroup?: boolean; + }, + ): Promise { + const loadWindow = this.beginSessionLoad(); + try { + const sessionId = await this.createLiveSession( + input.cwd, + loadWindow, + key, + this.sessionOptions(input.channelName), + ); + this.toSession.set(key, sessionId); + this.toTarget.set(sessionId, { + channelName: input.channelName, + senderId: input.senderId, + chatId: input.chatId, + threadId: input.threadId, + isGroup: input.isGroup, + }); + this.toCwd.set(sessionId, input.cwd); + this.liveSessionIds.add(sessionId); + this.persist(); + return sessionId; + } finally { + this.endSessionLoad(loadWindow); + } + } + + private async loadOrReplaceSession( + key: string, + savedSessionId: string, + input: { + channelName: string; + senderId: string; + chatId: string; + threadId?: string; + cwd: string; + isGroup?: boolean; + }, + ): Promise { + const savedCwd = this.toCwd.get(savedSessionId) ?? input.cwd; + const loadWindow = this.beginSessionLoad(); + try { + try { + const loadedSessionId = await this.bridge.loadSession( + savedSessionId, + savedCwd, + this.sessionOptions(input.channelName), + ); + if ( + typeof loadedSessionId !== 'string' || + loadedSessionId.length === 0 || + loadWindow.delete(loadedSessionId) + ) { + throw new Error('Invalid or dead restored session ID'); + } + if (loadedSessionId !== savedSessionId) { + const target = this.toTarget.get(savedSessionId); + this.deleteByKey(key); + this.toSession.set(key, loadedSessionId); + if (target) this.toTarget.set(loadedSessionId, target); + this.toCwd.set(loadedSessionId, savedCwd); + this.persist(); + } + this.liveSessionIds.add(loadedSessionId); + return loadedSessionId; + } catch (loadError) { try { - const sessionId = await this.createLiveSession( - sessionCwd, + const replacement = await this.createLiveSession( + input.cwd, loadWindow, key, - this.sessionOptions(channelName), + this.sessionOptions(input.channelName), ); - this.toSession.set(key, sessionId); - this.toTarget.set(sessionId, { - channelName, - senderId, - chatId, - threadId, - isGroup, + this.deleteByKey(key); + this.toSession.set(key, replacement); + this.toTarget.set(replacement, { + channelName: input.channelName, + senderId: input.senderId, + chatId: input.chatId, + threadId: input.threadId, + isGroup: input.isGroup, }); - this.toCwd.set(sessionId, sessionCwd); + this.toCwd.set(replacement, input.cwd); + this.liveSessionIds.add(replacement); this.persist(); - return sessionId; - } finally { - this.endSessionLoad(loadWindow); - } - }); - this.creatingSessions.set(key, created); - try { - return await created; - } finally { - if (this.creatingSessions.get(key) === created) { - this.creatingSessions.delete(key); + process.stderr.write( + `[SessionRouter] Replaced unavailable session ${sanitizeLogText(savedSessionId, 128)} for key ${sanitizeLogText(key, 256)} after load failed: ${sanitizeLogText(loadError instanceof Error ? loadError.message : String(loadError), 512)}\n`, + ); + return replacement; + } catch (createError) { + process.stderr.write( + `[SessionRouter] Failed to load session ${sanitizeLogText(savedSessionId, 128)} for key ${sanitizeLogText(key, 256)} (${sanitizeLogText(loadError instanceof Error ? loadError.message : String(loadError), 512)}) and failed to create a replacement (${sanitizeLogText(createError instanceof Error ? createError.message : String(createError), 512)})\n`, + ); + throw createError; } } + } finally { + this.endSessionLoad(loadWindow); } } @@ -263,6 +373,7 @@ export class SessionRouter { if (this.toCwd.delete(sessionId)) { removed = true; } + this.liveSessionIds.delete(sessionId); if (!removed && this.sessionLoadWindows.size > 0) { for (const loadWindow of this.sessionLoadWindows) { loadWindow.add(sessionId); @@ -274,12 +385,25 @@ export class SessionRouter { return removed; } + handleSessionDied(sessionId: string): boolean { + if (this.recoveryMode === 'eager') { + return this.removeSessionId(sessionId); + } + const known = this.toTarget.has(sessionId); + this.liveSessionIds.delete(sessionId); + for (const loadWindow of this.sessionLoadWindows) { + loadWindow.add(sessionId); + } + return known; + } + private deleteByKey(key: string): string | null { const sessionId = this.toSession.get(key); if (!sessionId) return null; this.toSession.delete(key); this.toTarget.delete(sessionId); this.toCwd.delete(sessionId); + this.liveSessionIds.delete(sessionId); return sessionId; } @@ -310,6 +434,21 @@ export class SessionRouter { return entries; } + restoreRoutes(): { restored: number; dropped: number } { + const persisted = this.readPersistedEntries(); + if (!persisted) return { restored: 0, dropped: 0 }; + this.dispose(); + let restored = 0; + for (const [key, entry] of Object.entries(persisted.entries)) { + this.toSession.set(key, entry.sessionId); + this.toTarget.set(entry.sessionId, entry.target); + this.toCwd.set(entry.sessionId, entry.cwd); + restored++; + } + if (persisted.dropped > 0) this.persist(); + return { restored, dropped: persisted.dropped }; + } + /** * Restore session mappings from a previous bridge. * Called after bridge restart — attempts loadSession for each saved mapping. @@ -403,13 +542,18 @@ export class SessionRouter { return { restored, failed }; } - /** Clear in-memory state and delete persist file. Used on clean shutdown. */ - clearAll(): void { + dispose(): void { this.toSession.clear(); this.toTarget.clear(); this.toCwd.clear(); this.creatingSessions.clear(); this.sessionLoadWindows.clear(); + this.liveSessionIds.clear(); + } + + /** Clear in-memory state and delete persist file. Used on clean shutdown. */ + clearAll(): void { + this.dispose(); if (this.persistPath && existsSync(this.persistPath)) { try { unlinkSync(this.persistPath); @@ -419,6 +563,62 @@ export class SessionRouter { } } + private readPersistedEntries(): + | { entries: Record; dropped: number } + | undefined { + const persistPath = this.persistPath; + if (!persistPath || !existsSync(persistPath)) return undefined; + + let parsed: unknown; + try { + parsed = JSON.parse(readFileSync(persistPath, 'utf-8')); + } catch (error) { + process.stderr.write( + `[SessionRouter] Corrupted persist file at ${sanitizeLogText(persistPath, 1024)}: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 512)}\n`, + ); + return undefined; + } + if ( + typeof parsed !== 'object' || + parsed === null || + Array.isArray(parsed) + ) { + process.stderr.write( + `[SessionRouter] Invalid route store at ${sanitizeLogText(persistPath, 1024)}: expected an object\n`, + ); + return undefined; + } + + const entries: Record = {}; + let dropped = 0; + for (const [key, value] of Object.entries(parsed)) { + if (this.isPersistedEntry(value)) entries[key] = value; + else dropped++; + } + return { entries, dropped }; + } + + private isPersistedEntry(value: unknown): value is PersistedEntry { + if (typeof value !== 'object' || value === null) return false; + const entry = value as Record; + const target = entry['target']; + if (typeof target !== 'object' || target === null) return false; + const typedTarget = target as Record; + return ( + typeof entry['sessionId'] === 'string' && + entry['sessionId'].length > 0 && + typeof entry['cwd'] === 'string' && + entry['cwd'].length > 0 && + typeof typedTarget['channelName'] === 'string' && + typeof typedTarget['senderId'] === 'string' && + typeof typedTarget['chatId'] === 'string' && + (typedTarget['threadId'] === undefined || + typeof typedTarget['threadId'] === 'string') && + (typedTarget['isGroup'] === undefined || + typeof typedTarget['isGroup'] === 'boolean') + ); + } + private persist(): void { if (!this.persistPath) return; From 21dfa307d50a374b48298d678574ea6d96db80d4 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 19:16:38 +0800 Subject: [PATCH 02/21] fix(channels): harden session route persistence --- .../channels/base/src/SessionRouter.test.ts | 92 +++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 91 ++++++++++++------ 2 files changed, 156 insertions(+), 27 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 368aebcdb2c..454f51a96fc 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -2,7 +2,9 @@ import { existsSync, mkdtempSync, readFileSync, + readdirSync, rmSync, + statSync, writeFileSync, } from 'node:fs'; import { tmpdir } from 'node:os'; @@ -11,6 +13,19 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { SessionRouter } from './SessionRouter.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; +const mockRenameSync = vi.hoisted(() => vi.fn()); + +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + renameSync: (from: string, to: string) => { + mockRenameSync(from, to); + return actual.renameSync(from, to); + }, + }; +}); + let sessionCounter = 0; function mockBridge(): ChannelAgentBridge { @@ -48,6 +63,7 @@ describe('SessionRouter', () => { beforeEach(() => { sessionCounter = 0; + mockRenameSync.mockClear(); bridge = mockBridge(); tempDirs = []; }); @@ -958,6 +974,82 @@ describe('SessionRouter', () => { }); }); + describe('persistence safety', () => { + it('quarantines invalid JSON and starts with no routes', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync(persistPath, '{bad'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + + expect(router.restoreRoutes()).toEqual({ restored: 0, dropped: 0 }); + expect(existsSync(persistPath)).toBe(false); + expect( + readdirSync(dir).some((name) => + name.startsWith('routes.json.corrupt-'), + ), + ).toBe(true); + }); + + it('drops malformed entries but keeps valid siblings', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'valid-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + }, + broken: { sessionId: 42 }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + + expect(router.restoreRoutes()).toEqual({ restored: 1, dropped: 1 }); + expect(router.getSession('ch', 'alice', 'chat1')).toBe('valid-session'); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({ + 'ch:alice:chat1': expect.objectContaining({ + sessionId: 'valid-session', + }), + }); + }); + + it('persists through a same-directory temporary file and rename', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + + await router.resolve('ch', 'alice', 'chat1'); + + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({ + 'ch:alice:chat1': expect.objectContaining({ sessionId: 'session-1' }), + }); + expect(mockRenameSync).toHaveBeenCalledWith( + expect.stringMatching(/\.tmp$/), + persistPath, + ); + expect(readdirSync(dir).filter((name) => name.endsWith('.tmp'))).toEqual( + [], + ); + if (process.platform !== 'win32') { + expect(statSync(dir).mode & 0o777).toBe(0o700); + expect(statSync(persistPath).mode & 0o777).toBe(0o600); + } + }); + }); + describe('getAll', () => { it('returns all session entries', async () => { const router = new SessionRouter(bridge, '/tmp'); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 690c773c457..39d6c38a108 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -1,4 +1,14 @@ -import { existsSync, readFileSync, writeFileSync, unlinkSync } from 'node:fs'; +import { + chmodSync, + existsSync, + mkdirSync, + readFileSync, + renameSync, + rmSync, + unlinkSync, + writeFileSync, +} from 'node:fs'; +import { dirname, join } from 'node:path'; import process from 'node:process'; import type { SessionScope, SessionTarget } from './types.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; @@ -458,25 +468,13 @@ export class SessionRouter { restored: number; failed: number; }> { - const persistPath = this.persistPath; - if (!persistPath || !existsSync(persistPath)) { - return { restored: 0, failed: 0 }; - } - - let entries: Record; - try { - entries = JSON.parse(readFileSync(persistPath, 'utf-8')); - } catch (err) { - const reason = err instanceof Error ? err.message : String(err); - process.stderr.write( - `[SessionRouter] Corrupted persist file at ${sanitizeLogText(persistPath, 1024)}: ${sanitizeLogText(reason, 512)}\n`, - ); - return { restored: 0, failed: 0 }; - } + const persisted = this.readPersistedEntries(); + if (!persisted) return { restored: 0, failed: 0 }; + const entries = persisted.entries; let restored = 0; let failed = 0; - let changed = false; + let changed = persisted.dropped > 0; const reservations = new Map(); // Reserve every persisted key up front so inbound messages during restart @@ -573,6 +571,12 @@ export class SessionRouter { try { parsed = JSON.parse(readFileSync(persistPath, 'utf-8')); } catch (error) { + const quarantinePath = `${persistPath}.corrupt-${Date.now()}`; + try { + renameSync(persistPath, quarantinePath); + } catch { + // Keep startup available even if quarantine itself fails. + } process.stderr.write( `[SessionRouter] Corrupted persist file at ${sanitizeLogText(persistPath, 1024)}: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 512)}\n`, ); @@ -583,6 +587,12 @@ export class SessionRouter { parsed === null || Array.isArray(parsed) ) { + const quarantinePath = `${persistPath}.corrupt-${Date.now()}`; + try { + renameSync(persistPath, quarantinePath); + } catch { + // Keep startup available even if quarantine itself fails. + } process.stderr.write( `[SessionRouter] Invalid route store at ${sanitizeLogText(persistPath, 1024)}: expected an object\n`, ); @@ -625,19 +635,46 @@ export class SessionRouter { const data: Record = {}; for (const [key, sessionId] of this.toSession) { const target = this.toTarget.get(sessionId); - if (target) { - data[key] = { - sessionId, - target, - cwd: this.toCwd.get(sessionId) || this.defaultCwd, - }; - } + if (!target) continue; + data[key] = { + sessionId, + target, + cwd: this.toCwd.get(sessionId) ?? this.defaultCwd, + }; } + const dir = dirname(this.persistPath); + const tempPath = join( + dir, + `${Date.now()}-${process.pid}-${Math.random().toString(16).slice(2)}.tmp`, + ); try { - writeFileSync(this.persistPath, JSON.stringify(data, null, 2), 'utf-8'); - } catch { - // best-effort — don't break message flow for persistence failure + mkdirSync(dir, { recursive: true, mode: 0o700 }); + try { + chmodSync(dir, 0o700); + } catch { + // Windows and some filesystems do not implement POSIX modes. + } + writeFileSync(tempPath, JSON.stringify(data, null, 2), { + encoding: 'utf-8', + mode: 0o600, + }); + renameSync(tempPath, this.persistPath); + try { + chmodSync(this.persistPath, 0o600); + } catch { + // Windows and some filesystems do not implement POSIX modes. + } + } catch (error) { + process.stderr.write( + `[SessionRouter] Failed to persist routes at ${sanitizeLogText(this.persistPath, 1024)}: ${sanitizeLogText(error instanceof Error ? error.message : String(error), 512)}\n`, + ); + } finally { + try { + rmSync(tempPath, { force: true }); + } catch { + // best-effort temp cleanup + } } } From 437dffbdb871b7c2be68ef46dac655930e8aba9d Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 19:27:14 +0800 Subject: [PATCH 03/21] fix(channels): drop malformed eager routes --- .../channels/base/src/SessionRouter.test.ts | 43 +++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 16 +++++-- 2 files changed, 55 insertions(+), 4 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 454f51a96fc..ed327640bb6 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -767,6 +767,49 @@ describe('SessionRouter', () => { expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({}); }); + it('drops malformed persisted routes from existing eager state', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'sessions.json'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + const aliceSession = await router.resolve('ch', 'alice', 'chat1'); + await router.resolve('ch', 'bob', 'chat2'); + const persisted = JSON.parse( + readFileSync(persistPath, 'utf-8'), + ) as Record; + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': persisted['ch:alice:chat1'], + 'ch:bob:chat2': { sessionId: 42 }, + }), + ); + const restartedBridge = { + ...mockBridge(), + loadSession: vi + .fn() + .mockImplementation((sessionId: string) => + Promise.resolve(sessionId), + ), + } as unknown as ChannelAgentBridge; + + router.setBridge(restartedBridge); + + await expect(router.restoreSessions()).resolves.toEqual({ + restored: 1, + failed: 0, + }); + expect(restartedBridge.loadSession).toHaveBeenCalledWith( + aliceSession, + '/tmp', + ); + expect(router.getSession('ch', 'alice', 'chat1')).toBe(aliceSession); + expect(router.getSession('ch', 'bob', 'chat2')).toBeUndefined(); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({ + 'ch:alice:chat1': expect.objectContaining({ sessionId: aliceSession }), + }); + }); + it('persists replacement ids returned by loadSession', async () => { const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); tempDirs.push(dir); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 39d6c38a108..ebfb0b155b9 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -477,6 +477,10 @@ export class SessionRouter { let changed = persisted.dropped > 0; const reservations = new Map(); + for (const key of persisted.droppedKeys) { + this.deleteByKey(key); + } + // Reserve every persisted key up front so inbound messages during restart // wait for restore instead of returning stale IDs or creating duplicates. for (const key of Object.keys(entries)) { @@ -562,7 +566,11 @@ export class SessionRouter { } private readPersistedEntries(): - | { entries: Record; dropped: number } + | { + entries: Record; + dropped: number; + droppedKeys: string[]; + } | undefined { const persistPath = this.persistPath; if (!persistPath || !existsSync(persistPath)) return undefined; @@ -600,12 +608,12 @@ export class SessionRouter { } const entries: Record = {}; - let dropped = 0; + const droppedKeys: string[] = []; for (const [key, value] of Object.entries(parsed)) { if (this.isPersistedEntry(value)) entries[key] = value; - else dropped++; + else droppedKeys.push(key); } - return { entries, dropped }; + return { entries, dropped: droppedKeys.length, droppedKeys }; } private isPersistedEntry(value: unknown): value is PersistedEntry { From 997640d70af0295b405c98b15cf395e555fd058f Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 19:35:33 +0800 Subject: [PATCH 04/21] fix(channels): preserve durable routes on session death --- .../channels/base/src/ChannelBase.test.ts | 58 ++++++++++++++++--- packages/channels/base/src/ChannelBase.ts | 2 +- .../cli/src/commands/channel/runtime.test.ts | 22 ++++++- packages/cli/src/commands/channel/runtime.ts | 4 +- 4 files changed, 74 insertions(+), 12 deletions(-) diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 27ac5119adb..c785fd7fd4e 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -24,6 +24,7 @@ import type { ChannelWebhookConfig, ChannelWebhookTask, } from './ChannelWebhookTask.js'; +import { SessionRouter } from './SessionRouter.js'; // Concrete test implementation class TestChannel extends ChannelBase { @@ -4320,10 +4321,51 @@ describe('ChannelBase', () => { expect(secondPrompt).toContain('Be concise.'); }); + it('forgets instructions when policy-aware session death preserves a route', async () => { + const router = new SessionRouter(bridge, '/tmp', 'user', undefined, { + recoveryMode: 'lazy', + }); + const ch = createChannel( + { instructions: 'Be concise.' }, + { router, registerBridgeEvents: true }, + ); + await ch.handleInbound(envelope({ text: 'first' })); + const sessionId = router.getSession('test-chan', 'user1', 'chat1'); + expect(sessionId).toBeDefined(); + + (bridge as unknown as EventEmitter).emit('sessionDied', { sessionId }); + + expect(router.hasSession('test-chan', 'user1', 'chat1')).toBe(true); + (bridge.loadSession as ReturnType).mockResolvedValueOnce( + sessionId, + ); + await ch.handleInbound(envelope({ text: 'second' })); + + const secondPrompt = (bridge.prompt as ReturnType).mock + .calls[1]![1] as string; + expect(secondPrompt).toContain('Be concise.'); + }); + + it('/status reports a dormant durable route as active', async () => { + const router = new SessionRouter(bridge, '/tmp', 'user', undefined, { + recoveryMode: 'lazy', + }); + const ch = createChannel({}, { router, registerBridgeEvents: true }); + await ch.handleInbound(envelope({ text: 'first' })); + (bridge as unknown as EventEmitter).emit('sessionDied', { + sessionId: 's-1', + }); + ch.sent = []; + + await ch.handleInbound(envelope({ text: '/status' })); + + expect(ch.sent[0]!.text).toContain('Session: active'); + }); + it('can register bridge events when a supplied router is channel-owned', () => { const router = { getTarget: vi.fn().mockReturnValue({ chatId: 'chat1' }), - removeSessionId: vi.fn(), + handleSessionDied: vi.fn(), setBridge: vi.fn(), }; const ch = createChannel({}, { @@ -4344,13 +4386,13 @@ describe('ChannelBase', () => { }); expect(ch.toolCalls).toEqual([{ chatId: 'chat1', event: toolCall }]); - expect(router.removeSessionId).toHaveBeenCalledWith('s-1'); + expect(router.handleSessionDied).toHaveBeenCalledWith('s-1'); }); it('leaves supplied router bridge events to the gateway by default', () => { const router = { getTarget: vi.fn(), - removeSessionId: vi.fn(), + handleSessionDied: vi.fn(), setBridge: vi.fn(), }; const ch = createChannel({}, { router } as unknown as ChannelBaseOptions); @@ -4367,13 +4409,13 @@ describe('ChannelBase', () => { }); expect(ch.toolCalls).toEqual([]); - expect(router.removeSessionId).not.toHaveBeenCalled(); + expect(router.handleSessionDied).not.toHaveBeenCalled(); }); it('updates a supplied router bridge even when events are gateway-owned', () => { const router = { getTarget: vi.fn(), - removeSessionId: vi.fn(), + handleSessionDied: vi.fn(), setBridge: vi.fn(), }; const ch = createChannel({}, { router } as unknown as ChannelBaseOptions); @@ -4414,7 +4456,7 @@ describe('ChannelBase', () => { const newBridge = createBridge(); const router = { getTarget: vi.fn().mockReturnValue({ chatId: 'chat1' }), - removeSessionId: vi.fn(), + handleSessionDied: vi.fn(), setBridge: vi.fn(), }; const ch = createChannel({}, { @@ -4444,8 +4486,8 @@ describe('ChannelBase', () => { (newBridge as unknown as EventEmitter).emit('toolCall', toolCall); expect(router.setBridge).toHaveBeenCalledWith(newBridge); - expect(router.removeSessionId).toHaveBeenCalledTimes(1); - expect(router.removeSessionId).toHaveBeenCalledWith('new-session'); + expect(router.handleSessionDied).toHaveBeenCalledTimes(1); + expect(router.handleSessionDied).toHaveBeenCalledWith('new-session'); expect(ch.toolCalls).toEqual([{ chatId: 'chat1', event: toolCall }]); }); diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 80ba87d8c39..d02742e17a4 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -1430,7 +1430,7 @@ export abstract class ChannelBase { onToolCall(_chatId: string, _event: ToolCallEvent): void {} onSessionDied(sessionId: string): void { - this.router.removeSessionId(sessionId); + this.router.handleSessionDied(sessionId); this.instructedSessions.delete(sessionId); this.removePendingPermissionsForSession(sessionId); } diff --git a/packages/cli/src/commands/channel/runtime.test.ts b/packages/cli/src/commands/channel/runtime.test.ts index 271b2abab18..4a14fa8162a 100644 --- a/packages/cli/src/commands/channel/runtime.test.ts +++ b/packages/cli/src/commands/channel/runtime.test.ts @@ -1,6 +1,10 @@ import { EventEmitter } from 'node:events'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { parseConfiguredChannels, registerPermissionRelay } from './runtime.js'; +import { + parseConfiguredChannels, + registerPermissionRelay, + registerSessionCleanup, +} from './runtime.js'; vi.mock('@qwen-code/qwen-code-core', () => ({ Storage: { getGlobalQwenDir: () => '/tmp/qwen' }, @@ -278,3 +282,19 @@ describe('registerPermissionRelay', () => { }); }); }); + +describe('registerSessionCleanup', () => { + it('updates routing state when no channel matches the dead session', () => { + const bridge = new EventEmitter(); + const router = { + getTarget: vi.fn(), + handleSessionDied: vi.fn(), + }; + + registerSessionCleanup(bridge as never, router as never, new Map()); + bridge.emit('sessionDied', { sessionId: 'session-1' }); + + expect(router.handleSessionDied).toHaveBeenCalledTimes(1); + expect(router.handleSessionDied).toHaveBeenCalledWith('session-1'); + }); +}); diff --git a/packages/cli/src/commands/channel/runtime.ts b/packages/cli/src/commands/channel/runtime.ts index 70a8799fa88..4363b78ad8d 100644 --- a/packages/cli/src/commands/channel/runtime.ts +++ b/packages/cli/src/commands/channel/runtime.ts @@ -228,14 +228,14 @@ export function registerSessionCleanup( const safeId = sanitizeLogText(event.sessionId, 128); const safeReason = event.reason ? sanitizeLogText(event.reason, 512) : ''; writeStderrLine( - `[Channel] Session ${safeId} died${safeReason ? ` (${safeReason})` : ''}, removing routing state`, + `[Channel] Session ${safeId} died${safeReason ? ` (${safeReason})` : ''}, updating routing state`, ); const target = router.getTarget(event.sessionId); const channel = target ? channels.get(target.channelName) : undefined; if (channel) { channel.onSessionDied(event.sessionId); } else { - router.removeSessionId(event.sessionId); + router.handleSessionDied(event.sessionId); } }); } From 8d4d0a843b3dc6e4973705d0ccb1636897367754 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 19:46:33 +0800 Subject: [PATCH 05/21] feat(cli): restore daemon channel routes lazily --- .../commands/channel/daemon-worker.test.ts | 47 +++++++++++++++---- .../cli/src/commands/channel/daemon-worker.ts | 15 ++++-- .../cli/src/commands/channel/runtime.test.ts | 14 ++++++ packages/cli/src/commands/channel/runtime.ts | 12 ++++- 4 files changed, 76 insertions(+), 12 deletions(-) diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index fb6192a8813..0fe85837a62 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -12,6 +12,9 @@ const mockRegisterToolCallDispatch = vi.hoisted(() => vi.fn()); const mockRegisterPermissionRelay = vi.hoisted(() => vi.fn()); const mockRegisterSessionCleanup = vi.hoisted(() => vi.fn()); const mockSessionsPath = vi.hoisted(() => vi.fn(() => '/tmp/sessions.json')); +const mockDaemonSessionRoutesPath = vi.hoisted(() => + vi.fn(() => '/tmp/qwen/channels/daemon/workspace-hash/routes.json'), +); const mockLoadSettings = vi.hoisted(() => vi.fn((_cwd?: string, _opts?: unknown) => ({ merged: { proxy: 'http://settings-proxy:8080' as string | undefined }, @@ -99,6 +102,10 @@ const mockDaemonChannelBridge = vi.hoisted(() => const mockRouterSetChannelScope = vi.hoisted(() => vi.fn()); const mockRouterSetChannelApprovalMode = vi.hoisted(() => vi.fn()); const mockRouterClearAll = vi.hoisted(() => vi.fn()); +const mockRouterRestoreRoutes = vi.hoisted(() => + vi.fn(() => ({ restored: 1, dropped: 0 })), +); +const mockRouterDispose = vi.hoisted(() => vi.fn()); const mockSessionRouter = vi.hoisted(() => vi.fn( ( @@ -110,6 +117,8 @@ const mockSessionRouter = vi.hoisted(() => setChannelScope: mockRouterSetChannelScope, setChannelApprovalMode: mockRouterSetChannelApprovalMode, clearAll: mockRouterClearAll, + restoreRoutes: mockRouterRestoreRoutes, + dispose: mockRouterDispose, }), ), ); @@ -139,6 +148,7 @@ vi.mock('./proxy.js', () => ({ vi.mock('./runtime.js', () => ({ createChannel: mockCreateChannel, + daemonSessionRoutesPath: mockDaemonSessionRoutesPath, loadChannelsConfig: mockLoadChannelsConfig, loadChannelsFromExtensions: mockLoadChannelsFromExtensions, parseConfiguredChannels: mockParseConfiguredChannels, @@ -612,8 +622,23 @@ describe('runChannelDaemonWorker', () => { skipLoadEnvironment: true, }); expect(mockLoadChannelsConfig).toHaveBeenCalledWith('/workspace', settings); + expect(mockDaemonSessionRoutesPath).toHaveBeenCalledWith('/workspace'); + expect(mockSessionRouter).toHaveBeenCalledWith( + expect.any(Object), + '/workspace', + 'user', + '/tmp/qwen/channels/daemon/workspace-hash/routes.json', + { recoveryMode: 'lazy' }, + ); + expect(mockRouterRestoreRoutes).toHaveBeenCalledTimes(1); + expect(mockBridgeLoadSession).not.toHaveBeenCalled(); + expect(mockRouterSetChannelScope.mock.invocationCallOrder[0]).toBeLessThan( + mockRouterRestoreRoutes.mock.invocationCallOrder[0], + ); + expect(mockRouterRestoreRoutes.mock.invocationCallOrder[0]).toBeLessThan( + mockCreateChannel.mock.invocationCallOrder[0], + ); expect(mockSessionsPath).not.toHaveBeenCalled(); - expect(mockSessionRouter.mock.calls[0]![3]).toBeUndefined(); expect(ready).toHaveBeenCalledWith({ channels: ['telegram'], requestedChannels: ['telegram'], @@ -623,8 +648,9 @@ describe('runChannelDaemonWorker', () => { await handle.close(); expect(mockBridgeStop).toHaveBeenCalled(); expect(mockBridgeStop.mock.invocationCallOrder[0]!).toBeLessThan( - mockRouterClearAll.mock.invocationCallOrder[0]!, + mockRouterDispose.mock.invocationCallOrder[0]!, ); + expect(mockRouterClearAll).not.toHaveBeenCalled(); }); it('selects all configured channels in one shared router', async () => { @@ -819,7 +845,7 @@ describe('runChannelDaemonWorker', () => { expect(mockBridgeStop).toHaveBeenCalled(); }); - it('clears router state when startup rollback bridge stop fails', async () => { + it('disposes router state when startup rollback bridge stop fails', async () => { const sdk = createSdk(); mockCreateChannel.mockRejectedValueOnce(new Error('adapter boom')); mockBridgeStop.mockImplementationOnce(() => { @@ -836,7 +862,8 @@ describe('runChannelDaemonWorker', () => { ).rejects.toThrow('adapter boom'); expect(mockBridgeStop).toHaveBeenCalled(); - expect(mockRouterClearAll).toHaveBeenCalled(); + expect(mockRouterDispose).toHaveBeenCalled(); + expect(mockRouterClearAll).not.toHaveBeenCalled(); }); it('does not repopulate daemon-private env from worker settings loads', async () => { @@ -966,7 +993,8 @@ describe('runChannelDaemonWorker', () => { await expect(started).rejects.toThrow('Daemon worker startup aborted.'); expect(disconnect).toHaveBeenCalled(); expect(mockBridgeStop).toHaveBeenCalled(); - expect(mockRouterClearAll).toHaveBeenCalled(); + expect(mockRouterDispose).toHaveBeenCalled(); + expect(mockRouterClearAll).not.toHaveBeenCalled(); }); it('fails fast when a channel cwd does not match the daemon workspace', async () => { @@ -988,7 +1016,7 @@ describe('runChannelDaemonWorker', () => { ).rejects.toThrow('must use daemon workspace "/workspace"'); }); - it('clears router state even when bridge stop fails during close', async () => { + it('disposes router state even when bridge stop fails during close', async () => { const sdk = createSdk(); mockBridgeStop.mockImplementationOnce(() => { throw new Error('stop boom'); @@ -1002,7 +1030,8 @@ describe('runChannelDaemonWorker', () => { }); await expect(handle.close()).rejects.toThrow('stop boom'); - expect(mockRouterClearAll).toHaveBeenCalled(); + expect(mockRouterDispose).toHaveBeenCalled(); + expect(mockRouterClearAll).not.toHaveBeenCalled(); }); it('runs webhook tasks on the matching channel handle', async () => { @@ -1421,6 +1450,7 @@ describe('daemonWorkerCommand', () => { expect(exit).not.toHaveBeenCalled(); expect(disconnect).not.toHaveBeenCalled(); expect(mockBridgeStop).not.toHaveBeenCalled(); + expect(mockRouterDispose).not.toHaveBeenCalled(); expect(mockRouterClearAll).not.toHaveBeenCalled(); await handler; @@ -1429,7 +1459,8 @@ describe('daemonWorkerCommand', () => { expect(send).not.toHaveBeenCalled(); expect(disconnect).toHaveBeenCalled(); expect(mockBridgeStop).toHaveBeenCalled(); - expect(mockRouterClearAll).toHaveBeenCalled(); + expect(mockRouterDispose).toHaveBeenCalled(); + expect(mockRouterClearAll).not.toHaveBeenCalled(); } finally { restoreSend(); } diff --git a/packages/cli/src/commands/channel/daemon-worker.ts b/packages/cli/src/commands/channel/daemon-worker.ts index dc646bc6c20..d324f507f8e 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -39,6 +39,7 @@ import { writeStderrLine, writeStdoutLine } from '../../utils/stdioHelpers.js'; import { resolveProxyUrl } from './proxy.js'; import { createChannel, + daemonSessionRoutesPath, loadChannelsConfig, loadChannelsFromExtensions, parseConfiguredChannels, @@ -349,7 +350,8 @@ export async function runChannelDaemonWorker( bridgeFacade, daemonWorkspace, 'user', - undefined, + daemonSessionRoutesPath(daemonWorkspace), + { recoveryMode: 'lazy' }, ); router = createdRouter; for (const { name, config } of parsed) { @@ -358,6 +360,13 @@ export async function runChannelDaemonWorker( createdRouter.setChannelApprovalMode(name, config.approvalMode); } } + const restoredRoutes = createdRouter.restoreRoutes(); + writeStdoutLine( + `[Channel] Restored ${restoredRoutes.restored} dormant route(s)` + + (restoredRoutes.dropped > 0 + ? `; dropped ${restoredRoutes.dropped} invalid route(s)` + : ''), + ); for (const { name, config } of parsed) { throwIfStartupAborted(startupSignal); @@ -450,7 +459,7 @@ export async function runChannelDaemonWorker( try { bridge.stop(); } finally { - createdRouter.clearAll(); + createdRouter.dispose(); } }, }; @@ -461,7 +470,7 @@ export async function runChannelDaemonWorker( } catch { // best-effort during startup rollback } finally { - router?.clearAll(); + router?.dispose(); } throw err; } diff --git a/packages/cli/src/commands/channel/runtime.test.ts b/packages/cli/src/commands/channel/runtime.test.ts index 4a14fa8162a..14addb943ea 100644 --- a/packages/cli/src/commands/channel/runtime.test.ts +++ b/packages/cli/src/commands/channel/runtime.test.ts @@ -1,13 +1,17 @@ import { EventEmitter } from 'node:events'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { + daemonSessionRoutesPath, parseConfiguredChannels, registerPermissionRelay, registerSessionCleanup, + sessionsPath, } from './runtime.js'; vi.mock('@qwen-code/qwen-code-core', () => ({ Storage: { getGlobalQwenDir: () => '/tmp/qwen' }, + hashDaemonWorkspace: (workspace: string) => + workspace === '/workspace' ? 'workspace-hash' : 'other-hash', })); vi.mock('../../config/settings.js', () => ({ @@ -28,6 +32,16 @@ vi.mock('./channel-registry.js', () => ({ supportedTypes: async () => ['telegram'], })); +it('isolates daemon route stores by workspace hash', () => { + expect(daemonSessionRoutesPath('/workspace')).toBe( + '/tmp/qwen/channels/daemon/workspace-hash/routes.json', + ); + expect(daemonSessionRoutesPath('/other')).toBe( + '/tmp/qwen/channels/daemon/other-hash/routes.json', + ); + expect(daemonSessionRoutesPath('/workspace')).not.toBe(sessionsPath()); +}); + describe('parseConfiguredChannels', () => { beforeEach(() => { delete process.env['TOKEN_LITERAL_VALUE']; diff --git a/packages/cli/src/commands/channel/runtime.ts b/packages/cli/src/commands/channel/runtime.ts index 4363b78ad8d..6b4d692dfe9 100644 --- a/packages/cli/src/commands/channel/runtime.ts +++ b/packages/cli/src/commands/channel/runtime.ts @@ -1,6 +1,6 @@ import * as path from 'node:path'; import { pathToFileURL } from 'node:url'; -import { Storage } from '@qwen-code/qwen-code-core'; +import { hashDaemonWorkspace, Storage } from '@qwen-code/qwen-code-core'; import type { SessionRouter, ChannelAgentBridge, @@ -31,6 +31,16 @@ export function sessionsPath(): string { return path.join(Storage.getGlobalQwenDir(), 'channels', 'sessions.json'); } +export function daemonSessionRoutesPath(workspaceCwd: string): string { + return path.join( + Storage.getGlobalQwenDir(), + 'channels', + 'daemon', + hashDaemonWorkspace(workspaceCwd), + 'routes.json', + ); +} + export function channelLoopPath(): string { return path.join(Storage.getGlobalQwenDir(), 'channels', 'cron.json'); } From 281ecfe0450b9daf7d54d02ad9a834a538584ffa Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 20:17:35 +0800 Subject: [PATCH 06/21] fix(channels): invalidate stale route operations --- .../base/src/DaemonChannelBridge.test.ts | 92 ++++++++++ .../channels/base/src/DaemonChannelBridge.ts | 19 ++ .../channels/base/src/SessionRouter.test.ts | 163 ++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 139 +++++++++++++-- 4 files changed, 396 insertions(+), 17 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index f98f1f9468d..d00e109654f 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1542,6 +1542,98 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); + it('rejects and cancels a new session factory result that arrives after stop', async () => { + const events = new EventQueue(); + const session = createFakeSession(events); + let finishFactory!: (session: FakeSession) => void; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + + await bridge.start(); + const creating = bridge.newSession('/repo'); + await Promise.resolve(); + bridge.stop(); + finishFactory(session); + + await expect(creating).rejects.toThrow('stopped'); + expect(session.cancel).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([]); + }); + + it('rejects and cancels a load factory result that arrives after stop', async () => { + const events = new EventQueue(); + const session = createFakeSession(events, 'existing-session'); + let finishFactory!: (session: FakeSession) => void; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + + await bridge.start(); + const loading = bridge.loadSession('existing-session', '/repo'); + await Promise.resolve(); + bridge.stop(); + finishFactory(session); + + await expect(loading).rejects.toThrow('stopped'); + expect(session.cancel).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([]); + }); + + it('keeps a pre-stop factory result stale after restart', async () => { + const staleEvents = new EventQueue(); + const staleSession = createFakeSession(staleEvents, 'stale-session'); + const currentEvents = new EventQueue(); + const currentSession = createFakeSession(currentEvents, 'current-session'); + let finishStaleFactory!: (session: FakeSession) => void; + const factory = vi + .fn() + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishStaleFactory = resolve; + }), + ) + .mockResolvedValueOnce(currentSession); + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: factory, + }); + + await bridge.start(); + const staleCreation = bridge.newSession('/repo'); + await Promise.resolve(); + bridge.stop(); + await bridge.start(); + finishStaleFactory(staleSession); + + await expect(staleCreation).rejects.toThrow('stopped'); + await expect(bridge.newSession('/repo')).resolves.toBe('current-session'); + expect(staleSession.cancel).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([ + { + sessionId: 'current-session', + workspaceCwd: '/repo', + hasActivePrompt: false, + }, + ]); + + currentEvents.close(); + bridge.stop(); + }); + it('rejects mismatched daemon session ids while loading', async () => { const events = new EventQueue(); const session = createFakeSession(events, 'different-session'); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index be296568589..da062bca82b 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -202,6 +202,7 @@ export class DaemonChannelBridge >(); private readonly turnBarriers = new Map void>(); private connected = false; + private lifecycleGeneration = 0; private latestAvailableCommandsSessionId: string | undefined; private lastError: unknown; @@ -252,12 +253,14 @@ export class DaemonChannelBridge cwd: string, options?: { approvalMode?: string }, ): Promise { + const lifecycleGeneration = this.lifecycleGeneration; const session = await this.options.sessionFactory({ workspaceCwd: cwd || this.options.cwd, modelServiceId: this.options.modelServiceId, sessionScope: this.options.sessionScope ?? 'thread', ...(options?.approvalMode ? { approvalMode: options.approvalMode } : {}), }); + await this.rejectStaleSession(session, lifecycleGeneration); this.attachSession(session); return session.sessionId; } @@ -267,6 +270,7 @@ export class DaemonChannelBridge cwd: string, options?: { approvalMode?: string }, ): Promise { + const lifecycleGeneration = this.lifecycleGeneration; const session = await this.options.sessionFactory({ workspaceCwd: cwd || this.options.cwd, modelServiceId: this.options.modelServiceId, @@ -274,6 +278,7 @@ export class DaemonChannelBridge sessionScope: this.options.sessionScope ?? 'thread', ...(options?.approvalMode ? { approvalMode: options.approvalMode } : {}), }); + await this.rejectStaleSession(session, lifecycleGeneration); if (session.sessionId !== sessionId) { throw new Error( `Daemon returned session ${session.sessionId} while loading ${sessionId}`, @@ -425,6 +430,7 @@ export class DaemonChannelBridge } stop(): void { + this.lifecycleGeneration++; for (const sessionId of Array.from(this.sessions.keys())) { const session = this.sessions.get(sessionId); if (session) { @@ -453,6 +459,19 @@ export class DaemonChannelBridge void this.pumpEvents(session, controller.signal); } + private async rejectStaleSession( + session: DaemonChannelSessionClient, + lifecycleGeneration: number, + ): Promise { + if (lifecycleGeneration === this.lifecycleGeneration) return; + try { + await session.cancel(); + } catch (error) { + this.lastError = error; + } + throw new Error('Daemon channel bridge stopped during session creation'); + } + private ensureSession(sessionId: string): DaemonChannelSessionClient { const session = this.sessions.get(sessionId); if (!session) { diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index ed327640bb6..96aa481dfa5 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -1200,6 +1200,169 @@ describe('SessionRouter', () => { expect(lazyBridge.loadSession).toHaveBeenCalledTimes(1); }); + it.each(['removeSession', 'removeSessionId'] as const)( + 'rejects a dormant load invalidated by %s', + async (removal) => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + let finishLoad!: (value: string) => void; + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn( + () => + new Promise((resolve) => { + finishLoad = resolve; + }), + ), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + if (removal === 'removeSession') { + router.removeSession('ch', 'alice', 'chat1'); + } else { + router.removeSessionId('old-session'); + } + finishLoad('old-session'); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({}); + }, + ); + + it('does not install a replacement created after route removal', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + let finishCreation!: (value: string) => void; + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn().mockRejectedValue(new Error('gone')), + newSession: vi.fn( + () => + new Promise((resolve) => { + finishCreation = resolve; + }), + ), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await vi.waitFor(() => expect(lazyBridge.newSession).toHaveBeenCalled()); + router.removeSession('ch', 'alice', 'chat1'); + finishCreation('replacement-session'); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + expect(router.getTarget('replacement-session')).toBeUndefined(); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({}); + }); + + it('does not retry an invalidated shared recovery operation', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + let failLoad!: (error: Error) => void; + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn( + () => + new Promise((_resolve, reject) => { + failLoad = reject; + }), + ), + newSession: vi.fn().mockResolvedValue('replacement-session'), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + router.restoreRoutes(); + + const first = router.resolve('ch', 'alice', 'chat1'); + const second = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + router.removeSession('ch', 'alice', 'chat1'); + failLoad(new Error('gone')); + + await expect(first).rejects.toThrow('invalidated'); + await expect(second).rejects.toThrow('invalidated'); + expect(lazyBridge.loadSession).toHaveBeenCalledTimes(1); + expect(lazyBridge.newSession).not.toHaveBeenCalled(); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + }); + + it('does not install an absent route created after its removal', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + let finishCreation!: (value: string) => void; + const lazyBridge = { + ...mockBridge(), + newSession: vi.fn( + () => + new Promise((resolve) => { + finishCreation = resolve; + }), + ), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + expect(router.removeSession('ch', 'alice', 'chat1')).toEqual([]); + finishCreation('late-session'); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + expect(router.getTarget('late-session')).toBeUndefined(); + expect(existsSync(persistPath)).toBe(false); + }); + + it.each(['dormant load', 'absent creation'] as const)( + 'rejects a late %s after disposal', + async (operation) => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + let finish!: (value: string) => void; + const lazyBridge = { + ...mockBridge(), + loadSession: vi.fn( + () => + new Promise((resolve) => { + finish = resolve; + }), + ), + newSession: vi.fn( + () => + new Promise((resolve) => { + finish = resolve; + }), + ), + } satisfies ChannelAgentBridge; + const router = createLazyRouter(persistPath, lazyBridge); + if (operation === 'dormant load') { + writePersistedSession(persistPath, 'ch:alice:chat1'); + router.restoreRoutes(); + } + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + router.dispose(); + finish(operation === 'dormant load' ? 'old-session' : 'late-session'); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + expect(router.getAll()).toEqual([]); + }, + ); + it('replaces a route only after fallback creation succeeds', async () => { const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); tempDirs.push(dir); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index ebfb0b155b9..4f609b92583 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -26,6 +26,13 @@ interface SessionReservation { reject: (error: unknown) => void; } +interface SessionOperation { + promise: Promise; + target: SessionTarget; + lifecycleGeneration: number; + invalidationError?: Error; +} + type SessionLoadWindow = Set; interface ResolveOptions { routingThreadId?: string; @@ -41,9 +48,10 @@ export class SessionRouter { private toSession: Map = new Map(); // routing key → session ID private toTarget: Map = new Map(); // session ID → target private toCwd: Map = new Map(); // session ID → cwd - private creatingSessions: Map> = new Map(); + private creatingSessions: Map = new Map(); private sessionLoadWindows: Set = new Set(); private readonly liveSessionIds = new Set(); + private lifecycleGeneration = 0; private bridge: ChannelAgentBridge; private defaultCwd: string; @@ -147,10 +155,14 @@ export class SessionRouter { const creating = this.creatingSessions.get(key); if (creating) { try { - const sessionId = await creating; + const sessionId = await creating.promise; + this.assertOperationCurrent(creating); this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } catch (error) { + if (creating.invalidationError) { + throw creating.invalidationError; + } if (this.creatingSessions.get(key) === creating) { this.creatingSessions.delete(key); } @@ -160,14 +172,23 @@ export class SessionRouter { } } - const operation = Promise.resolve().then(() => - existing - ? this.loadOrReplaceSession(key, existing, input) - : this.createAndStoreSession(key, input), + const operation = this.createSessionOperation( + { + channelName: input.channelName, + senderId: input.senderId, + chatId: input.chatId, + threadId: input.threadId, + isGroup: input.isGroup, + }, + (currentOperation) => + existing + ? this.loadOrReplaceSession(key, existing, input, currentOperation) + : this.createAndStoreSession(key, input, currentOperation), ); this.creatingSessions.set(key, operation); try { - const sessionId = await operation; + const sessionId = await operation.promise; + this.assertOperationCurrent(operation); this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } finally { @@ -192,6 +213,7 @@ export class SessionRouter { cwd: string; isGroup?: boolean; }, + operation: SessionOperation, ): Promise { const loadWindow = this.beginSessionLoad(); try { @@ -200,7 +222,9 @@ export class SessionRouter { loadWindow, key, this.sessionOptions(input.channelName), + operation, ); + this.assertOperationCurrent(operation); this.toSession.set(key, sessionId); this.toTarget.set(sessionId, { channelName: input.channelName, @@ -229,6 +253,7 @@ export class SessionRouter { cwd: string; isGroup?: boolean; }, + operation: SessionOperation, ): Promise { const savedCwd = this.toCwd.get(savedSessionId) ?? input.cwd; const loadWindow = this.beginSessionLoad(); @@ -239,6 +264,11 @@ export class SessionRouter { savedCwd, this.sessionOptions(input.channelName), ); + this.assertOperationCurrent(operation); + if (this.toSession.get(key) !== savedSessionId) { + this.invalidateOperation(operation); + this.assertOperationCurrent(operation); + } if ( typeof loadedSessionId !== 'string' || loadedSessionId.length === 0 || @@ -257,13 +287,16 @@ export class SessionRouter { this.liveSessionIds.add(loadedSessionId); return loadedSessionId; } catch (loadError) { + this.assertOperationCurrent(operation); try { const replacement = await this.createLiveSession( input.cwd, loadWindow, key, this.sessionOptions(input.channelName), + operation, ); + this.assertOperationCurrent(operation); this.deleteByKey(key); this.toSession.set(key, replacement); this.toTarget.set(replacement, { @@ -281,6 +314,7 @@ export class SessionRouter { ); return replacement; } catch (createError) { + this.assertOperationCurrent(operation); process.stderr.write( `[SessionRouter] Failed to load session ${sanitizeLogText(savedSessionId, 128)} for key ${sanitizeLogText(key, 256)} (${sanitizeLogText(loadError instanceof Error ? loadError.message : String(loadError), 512)}) and failed to create a replacement (${sanitizeLogText(createError instanceof Error ? createError.message : String(createError), 512)})\n`, ); @@ -347,6 +381,7 @@ export class SessionRouter { const scope = this.channelScopes.get(channelName) || this.defaultScope; if (chatId) { const key = this.routingKey(channelName, senderId, chatId, threadId); + this.invalidateRouteOperation(key); const sessionId = this.deleteByKey(key); if (sessionId) removedIds.push(sessionId); } else if (scope === 'single') { @@ -359,10 +394,19 @@ export class SessionRouter { target?.channelName === channelName && target.senderId === senderId ) { + this.invalidateRouteOperation(k); const sessionId = this.deleteByKey(k); if (sessionId) removedIds.push(sessionId); } } + for (const [key, operation] of [...this.creatingSessions]) { + if ( + operation.target.channelName === channelName && + operation.target.senderId === senderId + ) { + this.invalidateRouteOperation(key); + } + } } if (removedIds.length > 0) this.persist(); return removedIds; @@ -373,6 +417,7 @@ export class SessionRouter { let removed = false; for (const [key, mappedSessionId] of [...this.toSession.entries()]) { if (mappedSessionId === sessionId) { + this.invalidateRouteOperation(key); this.toSession.delete(key); removed = true; } @@ -471,11 +516,15 @@ export class SessionRouter { const persisted = this.readPersistedEntries(); if (!persisted) return { restored: 0, failed: 0 }; const entries = persisted.entries; + const restoreGeneration = this.lifecycleGeneration; let restored = 0; let failed = 0; let changed = persisted.dropped > 0; - const reservations = new Map(); + const reservations = new Map< + string, + { reservation: SessionReservation; operation: SessionOperation } + >(); for (const key of persisted.droppedKeys) { this.deleteByKey(key); @@ -487,20 +536,30 @@ export class SessionRouter { this.deleteByKey(key); const reservation = this.createSessionReservation(); reservation.promise.catch(() => undefined); - this.creatingSessions.set(key, reservation.promise); - reservations.set(key, reservation); + const operation = this.createSessionOperation( + entries[key]!.target, + () => reservation.promise, + ); + operation.promise.catch(() => undefined); + this.creatingSessions.set(key, operation); + reservations.set(key, { reservation, operation }); } const loadWindow = this.beginSessionLoad(); try { for (const [key, entry] of Object.entries(entries)) { - const reservation = reservations.get(key); - if (!reservation) continue; + const reserved = reservations.get(key); + if (!reserved) continue; + const { reservation, operation } = reserved; try { + this.assertOperationCurrent(operation); const options = this.sessionOptions(entry.target.channelName); - const sessionId = options - ? await this.bridge.loadSession(entry.sessionId, entry.cwd, options) - : await this.bridge.loadSession(entry.sessionId, entry.cwd); + const sessionId = await this.bridge.loadSession( + entry.sessionId, + entry.cwd, + options, + ); + this.assertOperationCurrent(operation); if (typeof sessionId !== 'string' || sessionId.length === 0) { throw new Error('Invalid restored session ID'); } @@ -527,7 +586,7 @@ export class SessionRouter { failed++; changed = true; } finally { - if (this.creatingSessions.get(key) === reservation.promise) { + if (this.creatingSessions.get(key) === operation) { this.creatingSessions.delete(key); } } @@ -537,7 +596,7 @@ export class SessionRouter { } // Update persist file to only include successfully restored sessions - if (changed) { + if (changed && restoreGeneration === this.lifecycleGeneration) { this.persist(); } @@ -545,6 +604,10 @@ export class SessionRouter { } dispose(): void { + this.lifecycleGeneration++; + for (const operation of this.creatingSessions.values()) { + this.invalidateOperation(operation); + } this.toSession.clear(); this.toTarget.clear(); this.toCwd.clear(); @@ -691,6 +754,7 @@ export class SessionRouter { loadWindow: SessionLoadWindow, routingKey: string, options: { approvalMode?: string } | undefined, + operation: SessionOperation, ): Promise { const maxAttempts = 2; let lastDeadSessionId: string | undefined; @@ -698,6 +762,7 @@ export class SessionRouter { const sessionId = options ? await this.bridge.newSession(cwd, options) : await this.bridge.newSession(cwd); + this.assertOperationCurrent(operation); if (typeof sessionId !== 'string' || sessionId.length === 0) { throw new Error('Invalid session ID from bridge'); } @@ -717,6 +782,46 @@ export class SessionRouter { return loadWindow; } + private createSessionOperation( + target: SessionTarget, + run: (operation: SessionOperation) => Promise, + ): SessionOperation { + const operation: SessionOperation = { + promise: Promise.resolve(''), + target, + lifecycleGeneration: this.lifecycleGeneration, + }; + operation.promise = Promise.resolve() + .then(() => run(operation)) + .catch((error: unknown) => { + this.assertOperationCurrent(operation); + throw error; + }); + return operation; + } + + private invalidateRouteOperation(key: string): void { + const operation = this.creatingSessions.get(key); + if (!operation) return; + this.invalidateOperation(operation); + this.creatingSessions.delete(key); + } + + private invalidateOperation(operation: SessionOperation): void { + operation.invalidationError ??= new Error( + 'Session route operation was invalidated', + ); + } + + private assertOperationCurrent(operation: SessionOperation): void { + if (operation.lifecycleGeneration !== this.lifecycleGeneration) { + this.invalidateOperation(operation); + } + if (operation.invalidationError) { + throw operation.invalidationError; + } + } + private createSessionReservation(): SessionReservation { let resolveReservation!: (sessionId: string) => void; let rejectReservation!: (error: unknown) => void; From 4abe61cdb0a499239f30391c0b219ea550dd4121 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 20:28:31 +0800 Subject: [PATCH 07/21] fix(channels): close route lifecycle microtask gaps --- .../base/src/DaemonChannelBridge.test.ts | 33 ++++++++++++++++ .../channels/base/src/DaemonChannelBridge.ts | 10 +++-- .../channels/base/src/SessionRouter.test.ts | 39 +++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 26 ++++++++++++- 4 files changed, 102 insertions(+), 6 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index d00e109654f..60dc1856996 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1592,6 +1592,39 @@ describe('DaemonChannelBridge', () => { expect(bridge.listSessions()).toEqual([]); }); + it.each(['new', 'load'] as const)( + 'does not attach a %s factory result after a queued stop', + async (operation) => { + const events = new EventQueue(); + const sessionId = + operation === 'new' ? 'new-session' : 'existing-session'; + const session = createFakeSession(events, sessionId); + let finishFactory!: (session: FakeSession) => void; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + + await bridge.start(); + const creating = + operation === 'new' + ? bridge.newSession('/repo') + : bridge.loadSession(sessionId, '/repo'); + await Promise.resolve(); + finishFactory(session); + queueMicrotask(() => bridge.stop()); + + await expect(creating).resolves.toBe(sessionId); + expect(session.cancel).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([]); + }, + ); + it('keeps a pre-stop factory result stale after restart', async () => { const staleEvents = new EventQueue(); const staleSession = createFakeSession(staleEvents, 'stale-session'); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index da062bca82b..f8af6b96a22 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -260,7 +260,9 @@ export class DaemonChannelBridge sessionScope: this.options.sessionScope ?? 'thread', ...(options?.approvalMode ? { approvalMode: options.approvalMode } : {}), }); - await this.rejectStaleSession(session, lifecycleGeneration); + if (lifecycleGeneration !== this.lifecycleGeneration) { + await this.rejectStaleSession(session); + } this.attachSession(session); return session.sessionId; } @@ -278,7 +280,9 @@ export class DaemonChannelBridge sessionScope: this.options.sessionScope ?? 'thread', ...(options?.approvalMode ? { approvalMode: options.approvalMode } : {}), }); - await this.rejectStaleSession(session, lifecycleGeneration); + if (lifecycleGeneration !== this.lifecycleGeneration) { + await this.rejectStaleSession(session); + } if (session.sessionId !== sessionId) { throw new Error( `Daemon returned session ${session.sessionId} while loading ${sessionId}`, @@ -461,9 +465,7 @@ export class DaemonChannelBridge private async rejectStaleSession( session: DaemonChannelSessionClient, - lifecycleGeneration: number, ): Promise { - if (lifecycleGeneration === this.lifecycleGeneration) return; try { await session.cancel(); } catch (error) { diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 96aa481dfa5..586689230cf 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -896,6 +896,45 @@ describe('SessionRouter', () => { expect(router.getAll()).toHaveLength(1); }); + it.each(['removeSession', 'removeSessionId'] as const)( + 'invalidates a restore waiter when %s runs after reservation resolution', + async (removal) => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'sessions.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + let resolveLoadSession!: (sessionId: string) => void; + bridge = { + ...mockBridge(), + loadSession: vi.fn( + () => + new Promise((resolve) => { + resolveLoadSession = resolve; + }), + ), + }; + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + + const restore = router.restoreSessions(); + await Promise.resolve(); + const resolved = router.resolve('ch', 'alice', 'chat1'); + resolveLoadSession('restored-session'); + queueMicrotask(() => { + if (removal === 'removeSession') { + router.removeSession('ch', 'alice', 'chat1'); + } else { + router.removeSessionId('restored-session'); + } + }); + + await expect(resolved).rejects.toThrow('invalidated'); + await expect(restore).resolves.toEqual({ restored: 1, failed: 0 }); + expect(bridge.newSession).not.toHaveBeenCalled(); + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + expect(JSON.parse(readFileSync(persistPath, 'utf-8'))).toEqual({}); + }, + ); + it('creates a fresh session for concurrent resolve when restore fails', async () => { const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); tempDirs.push(dir); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 4f609b92583..cb39e6b0f99 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -30,6 +30,7 @@ interface SessionOperation { promise: Promise; target: SessionTarget; lifecycleGeneration: number; + routeGeneration: number; invalidationError?: Error; } @@ -51,6 +52,7 @@ export class SessionRouter { private creatingSessions: Map = new Map(); private sessionLoadWindows: Set = new Set(); private readonly liveSessionIds = new Set(); + private readonly routeGenerations = new Map(); private lifecycleGeneration = 0; private bridge: ChannelAgentBridge; @@ -156,7 +158,7 @@ export class SessionRouter { if (creating) { try { const sessionId = await creating.promise; - this.assertOperationCurrent(creating); + this.assertOperationResultCurrent(key, sessionId, creating); this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } catch (error) { @@ -173,6 +175,7 @@ export class SessionRouter { } const operation = this.createSessionOperation( + key, { channelName: input.channelName, senderId: input.senderId, @@ -188,7 +191,7 @@ export class SessionRouter { this.creatingSessions.set(key, operation); try { const sessionId = await operation.promise; - this.assertOperationCurrent(operation); + this.assertOperationResultCurrent(key, sessionId, operation); this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } finally { @@ -537,6 +540,7 @@ export class SessionRouter { const reservation = this.createSessionReservation(); reservation.promise.catch(() => undefined); const operation = this.createSessionOperation( + key, entries[key]!.target, () => reservation.promise, ); @@ -614,6 +618,7 @@ export class SessionRouter { this.creatingSessions.clear(); this.sessionLoadWindows.clear(); this.liveSessionIds.clear(); + this.routeGenerations.clear(); } /** Clear in-memory state and delete persist file. Used on clean shutdown. */ @@ -783,6 +788,7 @@ export class SessionRouter { } private createSessionOperation( + key: string, target: SessionTarget, run: (operation: SessionOperation) => Promise, ): SessionOperation { @@ -790,6 +796,7 @@ export class SessionRouter { promise: Promise.resolve(''), target, lifecycleGeneration: this.lifecycleGeneration, + routeGeneration: this.routeGenerations.get(key) ?? 0, }; operation.promise = Promise.resolve() .then(() => run(operation)) @@ -801,6 +808,7 @@ export class SessionRouter { } private invalidateRouteOperation(key: string): void { + this.routeGenerations.set(key, (this.routeGenerations.get(key) ?? 0) + 1); const operation = this.creatingSessions.get(key); if (!operation) return; this.invalidateOperation(operation); @@ -822,6 +830,20 @@ export class SessionRouter { } } + private assertOperationResultCurrent( + key: string, + sessionId: string, + operation: SessionOperation, + ): void { + if (operation.routeGeneration !== (this.routeGenerations.get(key) ?? 0)) { + this.invalidateOperation(operation); + } + if (this.toSession.get(key) !== sessionId) { + this.invalidateOperation(operation); + } + this.assertOperationCurrent(operation); + } + private createSessionReservation(): SessionReservation { let resolveReservation!: (sessionId: string) => void; let rejectReservation!: (error: unknown) => void; From 4d3dbc1b11f764fb93ad5192b4b38372e62679b5 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 20:36:33 +0800 Subject: [PATCH 08/21] fix(channels): release route invalidation metadata --- .../channels/base/src/SessionRouter.test.ts | 34 +++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 30 ++++++++++++---- 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 586689230cf..343c9d3e0e0 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -57,6 +57,14 @@ function writePersistedSession(persistPath: string, key = 'key1'): void { ); } +function invalidationMetadataSize(router: SessionRouter): number { + const state = router as unknown as { + routeGenerations?: Map; + routeTokens?: Map; + }; + return (state.routeTokens ?? state.routeGenerations)?.size ?? 0; +} + describe('SessionRouter', () => { let bridge: ChannelAgentBridge; let tempDirs: string[] = []; @@ -550,6 +558,32 @@ describe('SessionRouter', () => { expect(router.hasSession('ch', 'bob', 'chat1', 'thread1')).toBe(true); expect(router.hasSession('ch', 'bob', 'chat1', 'thread2')).toBe(false); }); + + it('releases invalidation metadata for cleared and failed routes', async () => { + const router = new SessionRouter(bridge, '/tmp'); + + for (let index = 0; index < 20; index++) { + router.removeSession('ch', `missing-${index}`, `chat-${index}`); + } + + for (let index = 0; index < 20; index++) { + await router.resolve('ch', `complete-${index}`, `chat-${index}`); + router.removeSession('ch', `complete-${index}`, `chat-${index}`); + } + + router.setBridge({ + ...mockBridge(), + newSession: vi.fn().mockRejectedValue(new Error('unavailable')), + }); + for (let index = 0; index < 20; index++) { + await expect( + router.resolve('ch', `failed-${index}`, `chat-${index}`), + ).rejects.toThrow('unavailable'); + } + + expect(router.getAll()).toEqual([]); + expect(invalidationMetadataSize(router)).toBe(0); + }); }); describe('removeSessionId', () => { diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index cb39e6b0f99..fc0e5bf7ce6 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -30,7 +30,7 @@ interface SessionOperation { promise: Promise; target: SessionTarget; lifecycleGeneration: number; - routeGeneration: number; + routeToken: object; invalidationError?: Error; } @@ -52,7 +52,7 @@ export class SessionRouter { private creatingSessions: Map = new Map(); private sessionLoadWindows: Set = new Set(); private readonly liveSessionIds = new Set(); - private readonly routeGenerations = new Map(); + private readonly routeTokens = new Map(); private lifecycleGeneration = 0; private bridge: ChannelAgentBridge; @@ -168,6 +168,7 @@ export class SessionRouter { if (this.creatingSessions.get(key) === creating) { this.creatingSessions.delete(key); } + this.releaseRouteToken(key, creating); failedWaits++; if (failedWaits > 3) throw error; continue; @@ -198,6 +199,7 @@ export class SessionRouter { if (this.creatingSessions.get(key) === operation) { this.creatingSessions.delete(key); } + this.releaseRouteToken(key, operation); } } } @@ -593,6 +595,7 @@ export class SessionRouter { if (this.creatingSessions.get(key) === operation) { this.creatingSessions.delete(key); } + this.releaseRouteToken(key, operation); } } } finally { @@ -618,7 +621,7 @@ export class SessionRouter { this.creatingSessions.clear(); this.sessionLoadWindows.clear(); this.liveSessionIds.clear(); - this.routeGenerations.clear(); + this.routeTokens.clear(); } /** Clear in-memory state and delete persist file. Used on clean shutdown. */ @@ -792,11 +795,16 @@ export class SessionRouter { target: SessionTarget, run: (operation: SessionOperation) => Promise, ): SessionOperation { + let routeToken = this.routeTokens.get(key); + if (!routeToken) { + routeToken = {}; + this.routeTokens.set(key, routeToken); + } const operation: SessionOperation = { promise: Promise.resolve(''), target, lifecycleGeneration: this.lifecycleGeneration, - routeGeneration: this.routeGenerations.get(key) ?? 0, + routeToken, }; operation.promise = Promise.resolve() .then(() => run(operation)) @@ -808,7 +816,7 @@ export class SessionRouter { } private invalidateRouteOperation(key: string): void { - this.routeGenerations.set(key, (this.routeGenerations.get(key) ?? 0) + 1); + this.routeTokens.delete(key); const operation = this.creatingSessions.get(key); if (!operation) return; this.invalidateOperation(operation); @@ -835,7 +843,7 @@ export class SessionRouter { sessionId: string, operation: SessionOperation, ): void { - if (operation.routeGeneration !== (this.routeGenerations.get(key) ?? 0)) { + if (operation.routeToken !== this.routeTokens.get(key)) { this.invalidateOperation(operation); } if (this.toSession.get(key) !== sessionId) { @@ -844,6 +852,16 @@ export class SessionRouter { this.assertOperationCurrent(operation); } + private releaseRouteToken(key: string, operation: SessionOperation): void { + if ( + this.routeTokens.get(key) === operation.routeToken && + !this.toSession.has(key) && + !this.creatingSessions.has(key) + ) { + this.routeTokens.delete(key); + } + } + private createSessionReservation(): SessionReservation { let resolveReservation!: (sessionId: string) => void; let rejectReservation!: (error: unknown) => void; From 7677528a4e69fd608e1ed69aa7ee9443d73bb267 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 20:53:17 +0800 Subject: [PATCH 09/21] fix(channels): discard invalidated daemon sessions --- .../channels/base/src/ChannelAgentBridge.ts | 2 + .../channels/base/src/DaemonChannelBridge.ts | 29 +++- .../channels/base/src/SessionRouter.test.ts | 132 ++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 68 +++++++-- 4 files changed, 218 insertions(+), 13 deletions(-) diff --git a/packages/channels/base/src/ChannelAgentBridge.ts b/packages/channels/base/src/ChannelAgentBridge.ts index 8d699e0a203..b694600bcc7 100644 --- a/packages/channels/base/src/ChannelAgentBridge.ts +++ b/packages/channels/base/src/ChannelAgentBridge.ts @@ -108,6 +108,8 @@ export interface ChannelAgentBridge { options?: { imageBase64?: string; imageMimeType?: string }, ): Promise; cancelSession(sessionId: string): Promise; + /** Release a bridge-owned session that will not be routed to a caller. */ + discardSession?(sessionId: string): Promise; respondToPermission?( requestId: string, response: RequestPermissionResponse, diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index f8af6b96a22..9177e01be5c 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -37,6 +37,7 @@ export interface DaemonChannelSessionClient { lastEventId?: number; resume?: boolean; }): AsyncGenerator; + detach?(): Promise; cancel(): Promise; setModel(modelId: string): Promise>; respondToPermission( @@ -396,6 +397,20 @@ export class DaemonChannelBridge await session.cancel(); } + async discardSession(sessionId: string): Promise { + const session = this.removeSessionBinding(sessionId); + if (!session) return; + if (session.detach) { + try { + await session.detach(); + return; + } catch { + // Fall back to cancellation for clients that cannot detach cleanly. + } + } + await session.cancel(); + } + async setSessionModel( sessionId: string, modelId: string, @@ -760,9 +775,15 @@ export class DaemonChannelBridge } private dropSession(sessionId: string, reason: string): void { - if (!this.sessions.has(sessionId)) { - return; - } + if (!this.removeSessionBinding(sessionId)) return; + this.emit('sessionDied', { sessionId, reason }); + } + + private removeSessionBinding( + sessionId: string, + ): DaemonChannelSessionClient | undefined { + const session = this.sessions.get(sessionId); + if (!session) return undefined; this.resolveTurnBarrier(sessionId); this.eventControllers.get(sessionId)?.abort(); this.eventControllers.delete(sessionId); @@ -785,7 +806,7 @@ export class DaemonChannelBridge this.respondedRequestToSession.delete(requestId); } } - this.emit('sessionDied', { sessionId, reason }); + return session; } private getStringField( diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 343c9d3e0e0..0bd339cf252 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -11,6 +11,10 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { SessionRouter } from './SessionRouter.js'; +import { + DaemonChannelBridge, + type DaemonChannelSessionClient, +} from './DaemonChannelBridge.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; const mockRenameSync = vi.hoisted(() => vi.fn()); @@ -65,6 +69,33 @@ function invalidationMetadataSize(router: SessionRouter): number { return (state.routeTokens ?? state.routeGenerations)?.size ?? 0; } +function daemonSession( + sessionId: string, + detach?: () => Promise, +): DaemonChannelSessionClient & { detach?: () => Promise } { + return { + sessionId, + workspaceCwd: '/tmp', + prompt: vi.fn().mockResolvedValue({}), + events: vi.fn(async function* (options?: { signal?: AbortSignal }) { + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve(); + } else { + options?.signal?.addEventListener('abort', () => resolve(), { + once: true, + }); + } + }); + yield* []; + }), + cancel: vi.fn().mockResolvedValue(undefined), + setModel: vi.fn().mockResolvedValue({}), + respondToPermission: vi.fn().mockResolvedValue(true), + ...(detach ? { detach } : {}), + }; +} + describe('SessionRouter', () => { let bridge: ChannelAgentBridge; let tempDirs: string[] = []; @@ -1273,6 +1304,107 @@ describe('SessionRouter', () => { expect(lazyBridge.loadSession).toHaveBeenCalledTimes(1); }); + it('discards a daemon client created after an absent route is cleared', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const detach = vi.fn().mockResolvedValue(undefined); + const session = daemonSession('late-session', detach); + let finishFactory!: (session: DaemonChannelSessionClient) => void; + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + const sessionDied = vi.fn(); + daemonBridge.on('sessionDied', sessionDied); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + router.removeSession('ch', 'alice', 'chat1'); + finishFactory(session); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(daemonBridge.listSessions()).toEqual([]); + expect(detach).toHaveBeenCalledOnce(); + expect(session.cancel).not.toHaveBeenCalled(); + expect(sessionDied).not.toHaveBeenCalled(); + await daemonBridge.discardSession('late-session'); + expect(detach).toHaveBeenCalledOnce(); + }); + + it('discards a loaded daemon client after its dormant route is cleared', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const session = daemonSession('old-session'); + let finishFactory!: (session: DaemonChannelSessionClient) => void; + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + router.restoreRoutes(); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await Promise.resolve(); + router.removeSession('ch', 'alice', 'chat1'); + finishFactory(session); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(daemonBridge.listSessions()).toEqual([]); + expect(session.cancel).toHaveBeenCalledOnce(); + }); + + it('falls back to cancel when detach fails for an invalidated replacement', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const detach = vi.fn().mockRejectedValue(new Error('detach failed')); + const session = daemonSession('replacement-session', detach); + let finishFactory!: (session: DaemonChannelSessionClient) => void; + const factory = vi + .fn() + .mockRejectedValueOnce(new Error('gone')) + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ); + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: factory, + }); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + router.restoreRoutes(); + + const resolving = router.resolve('ch', 'alice', 'chat1'); + await vi.waitFor(() => expect(factory).toHaveBeenCalledTimes(2)); + router.removeSession('ch', 'alice', 'chat1'); + finishFactory(session); + + await expect(resolving).rejects.toThrow('invalidated'); + expect(daemonBridge.listSessions()).toEqual([]); + expect(detach).toHaveBeenCalledOnce(); + expect(session.cancel).toHaveBeenCalledOnce(); + }); + it.each(['removeSession', 'removeSessionId'] as const)( 'rejects a dormant load invalidated by %s', async (removal) => { diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index fc0e5bf7ce6..d885beaab4b 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -158,7 +158,12 @@ export class SessionRouter { if (creating) { try { const sessionId = await creating.promise; - this.assertOperationResultCurrent(key, sessionId, creating); + try { + this.assertOperationResultCurrent(key, sessionId, creating); + } catch (error) { + await this.discardInvalidatedSession(key, sessionId, creating); + throw error; + } this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } catch (error) { @@ -192,7 +197,12 @@ export class SessionRouter { this.creatingSessions.set(key, operation); try { const sessionId = await operation.promise; - this.assertOperationResultCurrent(key, sessionId, operation); + try { + this.assertOperationResultCurrent(key, sessionId, operation); + } catch (error) { + await this.discardInvalidatedSession(key, sessionId, operation); + throw error; + } this.promoteTargetToGroup(sessionId, isGroup); return sessionId; } finally { @@ -229,7 +239,12 @@ export class SessionRouter { this.sessionOptions(input.channelName), operation, ); - this.assertOperationCurrent(operation); + try { + this.assertOperationCurrent(operation); + } catch (error) { + await this.discardInvalidatedSession(key, sessionId, operation); + throw error; + } this.toSession.set(key, sessionId); this.toTarget.set(sessionId, { channelName: input.channelName, @@ -269,10 +284,15 @@ export class SessionRouter { savedCwd, this.sessionOptions(input.channelName), ); - this.assertOperationCurrent(operation); - if (this.toSession.get(key) !== savedSessionId) { - this.invalidateOperation(operation); + try { this.assertOperationCurrent(operation); + if (this.toSession.get(key) !== savedSessionId) { + this.invalidateOperation(operation); + this.assertOperationCurrent(operation); + } + } catch (error) { + await this.discardInvalidatedSession(key, loadedSessionId, operation); + throw error; } if ( typeof loadedSessionId !== 'string' || @@ -301,7 +321,12 @@ export class SessionRouter { this.sessionOptions(input.channelName), operation, ); - this.assertOperationCurrent(operation); + try { + this.assertOperationCurrent(operation); + } catch (error) { + await this.discardInvalidatedSession(key, replacement, operation); + throw error; + } this.deleteByKey(key); this.toSession.set(key, replacement); this.toTarget.set(replacement, { @@ -565,7 +590,12 @@ export class SessionRouter { entry.cwd, options, ); - this.assertOperationCurrent(operation); + try { + this.assertOperationCurrent(operation); + } catch (error) { + await this.discardInvalidatedSession(key, sessionId, operation); + throw error; + } if (typeof sessionId !== 'string' || sessionId.length === 0) { throw new Error('Invalid restored session ID'); } @@ -770,7 +800,12 @@ export class SessionRouter { const sessionId = options ? await this.bridge.newSession(cwd, options) : await this.bridge.newSession(cwd); - this.assertOperationCurrent(operation); + try { + this.assertOperationCurrent(operation); + } catch (error) { + await this.discardInvalidatedSession(routingKey, sessionId, operation); + throw error; + } if (typeof sessionId !== 'string' || sessionId.length === 0) { throw new Error('Invalid session ID from bridge'); } @@ -862,6 +897,21 @@ export class SessionRouter { } } + private async discardInvalidatedSession( + key: string, + sessionId: string, + operation: SessionOperation, + ): Promise { + const currentOperation = this.creatingSessions.get(key); + if (currentOperation && currentOperation !== operation) return; + if ([...this.toSession.values()].includes(sessionId)) return; + try { + await this.bridge.discardSession?.(sessionId); + } catch { + // Best-effort cleanup must not replace the terminal invalidation. + } + } + private createSessionReservation(): SessionReservation { let resolveReservation!: (sessionId: string) => void; let rejectReservation!: (error: unknown) => void; From e1de956f217f1c4a737f842ae915a222e0cbb2c3 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 21:03:16 +0800 Subject: [PATCH 10/21] fix(channels): defer orphan session cleanup --- .../channels/base/src/SessionRouter.test.ts | 98 +++++++++++++++++++ packages/channels/base/src/SessionRouter.ts | 39 +++++--- 2 files changed, 125 insertions(+), 12 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 0bd339cf252..0dff96bfd05 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -96,6 +96,12 @@ function daemonSession( }; } +async function drainMicrotasks(): Promise { + for (let index = 0; index < 20; index++) { + await Promise.resolve(); + } +} + describe('SessionRouter', () => { let bridge: ChannelAgentBridge; let tempDirs: string[] = []; @@ -1405,6 +1411,98 @@ describe('SessionRouter', () => { expect(session.cancel).toHaveBeenCalledOnce(); }); + it('does not discard a same-id binding owned by another in-flight route', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const firstSession = daemonSession('shared-session'); + const secondDetach = vi.fn().mockResolvedValue(undefined); + const secondSession = daemonSession('shared-session', secondDetach); + const finishFactories: Array< + (session: DaemonChannelSessionClient) => void + > = []; + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactories.push(resolve); + }), + ), + }); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + + const first = router.resolve('ch', 'alice', 'chat1'); + const second = router.resolve('ch', 'bob', 'chat2'); + await drainMicrotasks(); + expect(finishFactories).toHaveLength(2); + router.removeSession('ch', 'alice', 'chat1'); + finishFactories[0]!(firstSession); + finishFactories[1]!(secondSession); + + await expect(first).rejects.toThrow('invalidated'); + await expect(second).resolves.toBe('shared-session'); + expect(router.getSession('ch', 'bob', 'chat2')).toBe('shared-session'); + expect(daemonBridge.listSessions()).toEqual([ + { + sessionId: 'shared-session', + workspaceCwd: '/tmp', + hasActivePrompt: false, + }, + ]); + expect(secondDetach).not.toHaveBeenCalled(); + + daemonBridge.stop(); + }); + + it.each(['detach', 'cancel'] as const)( + 'does not wait for a hanging %s while rejecting invalidated creation', + async (cleanup) => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const neverSettles = vi.fn(() => new Promise(() => undefined)); + const session = daemonSession( + 'late-session', + cleanup === 'detach' ? neverSettles : undefined, + ); + if (cleanup === 'cancel') { + session.cancel = neverSettles; + } + let finishFactory!: (session: DaemonChannelSessionClient) => void; + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + + let rejection: unknown; + const resolving = router.resolve('ch', 'alice', 'chat1'); + void resolving.catch((error: unknown) => { + rejection = error; + }); + await drainMicrotasks(); + router.removeSession('ch', 'alice', 'chat1'); + finishFactory(session); + await drainMicrotasks(); + + expect(rejection).toEqual( + expect.objectContaining({ + message: 'Session route operation was invalidated', + }), + ); + expect(neverSettles).toHaveBeenCalledOnce(); + expect(daemonBridge.listSessions()).toEqual([]); + }, + ); + it.each(['removeSession', 'removeSessionId'] as const)( 'rejects a dormant load invalidated by %s', async (removal) => { diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index d885beaab4b..a866d8bd11c 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -161,7 +161,7 @@ export class SessionRouter { try { this.assertOperationResultCurrent(key, sessionId, creating); } catch (error) { - await this.discardInvalidatedSession(key, sessionId, creating); + this.scheduleDiscardInvalidatedSession(sessionId, creating); throw error; } this.promoteTargetToGroup(sessionId, isGroup); @@ -200,7 +200,7 @@ export class SessionRouter { try { this.assertOperationResultCurrent(key, sessionId, operation); } catch (error) { - await this.discardInvalidatedSession(key, sessionId, operation); + this.scheduleDiscardInvalidatedSession(sessionId, operation); throw error; } this.promoteTargetToGroup(sessionId, isGroup); @@ -242,7 +242,7 @@ export class SessionRouter { try { this.assertOperationCurrent(operation); } catch (error) { - await this.discardInvalidatedSession(key, sessionId, operation); + this.scheduleDiscardInvalidatedSession(sessionId, operation); throw error; } this.toSession.set(key, sessionId); @@ -291,7 +291,7 @@ export class SessionRouter { this.assertOperationCurrent(operation); } } catch (error) { - await this.discardInvalidatedSession(key, loadedSessionId, operation); + this.scheduleDiscardInvalidatedSession(loadedSessionId, operation); throw error; } if ( @@ -324,7 +324,7 @@ export class SessionRouter { try { this.assertOperationCurrent(operation); } catch (error) { - await this.discardInvalidatedSession(key, replacement, operation); + this.scheduleDiscardInvalidatedSession(replacement, operation); throw error; } this.deleteByKey(key); @@ -593,7 +593,7 @@ export class SessionRouter { try { this.assertOperationCurrent(operation); } catch (error) { - await this.discardInvalidatedSession(key, sessionId, operation); + this.scheduleDiscardInvalidatedSession(sessionId, operation); throw error; } if (typeof sessionId !== 'string' || sessionId.length === 0) { @@ -803,7 +803,7 @@ export class SessionRouter { try { this.assertOperationCurrent(operation); } catch (error) { - await this.discardInvalidatedSession(routingKey, sessionId, operation); + this.scheduleDiscardInvalidatedSession(sessionId, operation); throw error; } if (typeof sessionId !== 'string' || sessionId.length === 0) { @@ -897,16 +897,31 @@ export class SessionRouter { } } - private async discardInvalidatedSession( - key: string, + private scheduleDiscardInvalidatedSession( + sessionId: string, + operation: SessionOperation, + ): void { + void this.discardInvalidatedSessionWhenUnowned(sessionId, operation).catch( + () => undefined, + ); + } + + private async discardInvalidatedSessionWhenUnowned( sessionId: string, operation: SessionOperation, ): Promise { - const currentOperation = this.creatingSessions.get(key); - if (currentOperation && currentOperation !== operation) return; + for (;;) { + const possibleOwners = [...this.creatingSessions.values()].filter( + (current) => current !== operation, + ); + if (possibleOwners.length === 0) break; + await Promise.allSettled( + possibleOwners.map((current) => current.promise), + ); + } if ([...this.toSession.values()].includes(sessionId)) return; try { - await this.bridge.discardSession?.(sessionId); + void this.bridge.discardSession?.(sessionId).catch(() => undefined); } catch { // Best-effort cleanup must not replace the terminal invalidation. } From ebfa37a58bb0d671b453ab2659aa3fbd6c4e5a06 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 21:13:46 +0800 Subject: [PATCH 11/21] fix(channels): scope orphan cleanup to bindings --- .../channels/base/src/ChannelAgentBridge.ts | 7 ++- .../base/src/DaemonChannelBridge.test.ts | 49 +++++++++++++++ .../channels/base/src/DaemonChannelBridge.ts | 25 ++++++-- .../channels/base/src/SessionRouter.test.ts | 59 ++++++++++++++++++- packages/channels/base/src/SessionRouter.ts | 28 ++------- 5 files changed, 138 insertions(+), 30 deletions(-) diff --git a/packages/channels/base/src/ChannelAgentBridge.ts b/packages/channels/base/src/ChannelAgentBridge.ts index b694600bcc7..928e4a7df52 100644 --- a/packages/channels/base/src/ChannelAgentBridge.ts +++ b/packages/channels/base/src/ChannelAgentBridge.ts @@ -96,11 +96,13 @@ export interface ChannelAgentBridge { newSession( cwd: string, options?: ChannelAgentBridgeSessionOptions, + bindingToken?: object, ): Promise; loadSession( sessionId: string, cwd: string, options?: ChannelAgentBridgeSessionOptions, + bindingToken?: object, ): Promise; prompt( sessionId: string, @@ -109,7 +111,10 @@ export interface ChannelAgentBridge { ): Promise; cancelSession(sessionId: string): Promise; /** Release a bridge-owned session that will not be routed to a caller. */ - discardSession?(sessionId: string): Promise; + discardSession?( + sessionId: string, + expectedBindingToken?: object, + ): Promise; respondToPermission?( requestId: string, response: RequestPermissionResponse, diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index 60dc1856996..91be058b367 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1667,6 +1667,55 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); + it('conditionally discards only the binding owned by the expected token', async () => { + const firstEvents = new EventQueue(); + const secondEvents = new EventQueue(); + const firstSession = createFakeSession(firstEvents, 'shared-session'); + const secondSession = createFakeSession(secondEvents, 'shared-session'); + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi + .fn() + .mockResolvedValueOnce(firstSession) + .mockResolvedValueOnce(secondSession), + }); + const bindingBridge = bridge as unknown as { + newSession( + cwd: string, + options: undefined, + bindingToken: object, + ): Promise; + discardSession(sessionId: string, expectedToken: object): Promise; + }; + const firstToken = {}; + const secondToken = {}; + + await bridge.start(); + await bindingBridge.newSession('/repo', undefined, firstToken); + await bindingBridge.newSession('/repo', undefined, secondToken); + await bindingBridge.discardSession('shared-session', firstToken); + + expect(bridge.listSessions()).toEqual([ + { + sessionId: 'shared-session', + workspaceCwd: '/repo', + hasActivePrompt: false, + }, + ]); + expect(secondSession.cancel).not.toHaveBeenCalled(); + + await bindingBridge.discardSession('shared-session', secondToken); + expect(bridge.listSessions()).toEqual([]); + expect(secondSession.cancel).toHaveBeenCalledOnce(); + expect( + ( + bridge as unknown as { + sessionBindingTokens: Map; + } + ).sessionBindingTokens.size, + ).toBe(0); + }); + it('rejects mismatched daemon session ids while loading', async () => { const events = new EventQueue(); const session = createFakeSession(events, 'different-session'); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index 9177e01be5c..997417d266c 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -189,6 +189,7 @@ export class DaemonChannelBridge { private readonly options: DaemonChannelBridgeOptions; private readonly sessions = new Map(); + private readonly sessionBindingTokens = new Map(); private readonly eventControllers = new Map(); private readonly requestToSession = new Map(); private readonly respondedRequestToSession = new Map(); @@ -253,6 +254,7 @@ export class DaemonChannelBridge async newSession( cwd: string, options?: { approvalMode?: string }, + bindingToken?: object, ): Promise { const lifecycleGeneration = this.lifecycleGeneration; const session = await this.options.sessionFactory({ @@ -264,7 +266,7 @@ export class DaemonChannelBridge if (lifecycleGeneration !== this.lifecycleGeneration) { await this.rejectStaleSession(session); } - this.attachSession(session); + this.attachSession(session, bindingToken); return session.sessionId; } @@ -272,6 +274,7 @@ export class DaemonChannelBridge sessionId: string, cwd: string, options?: { approvalMode?: string }, + bindingToken?: object, ): Promise { const lifecycleGeneration = this.lifecycleGeneration; const session = await this.options.sessionFactory({ @@ -289,7 +292,7 @@ export class DaemonChannelBridge `Daemon returned session ${session.sessionId} while loading ${sessionId}`, ); } - this.attachSession(session); + this.attachSession(session, bindingToken); return session.sessionId; } @@ -397,7 +400,16 @@ export class DaemonChannelBridge await session.cancel(); } - async discardSession(sessionId: string): Promise { + async discardSession( + sessionId: string, + expectedBindingToken?: object, + ): Promise { + if ( + expectedBindingToken !== undefined && + this.sessionBindingTokens.get(sessionId) !== expectedBindingToken + ) { + return; + } const session = this.removeSessionBinding(sessionId); if (!session) return; if (session.detach) { @@ -467,12 +479,16 @@ export class DaemonChannelBridge return this.connected; } - private attachSession(session: DaemonChannelSessionClient): void { + private attachSession( + session: DaemonChannelSessionClient, + bindingToken?: object, + ): void { if (this.sessions.has(session.sessionId)) { this.dropSession(session.sessionId, 'session_replaced'); } this.sessions.set(session.sessionId, session); + this.sessionBindingTokens.set(session.sessionId, bindingToken); const controller = new AbortController(); this.eventControllers.set(session.sessionId, controller); void this.pumpEvents(session, controller.signal); @@ -788,6 +804,7 @@ export class DaemonChannelBridge this.eventControllers.get(sessionId)?.abort(); this.eventControllers.delete(sessionId); this.sessions.delete(sessionId); + this.sessionBindingTokens.delete(sessionId); this.abortActivePrompts(sessionId); this.activePrompts.delete(sessionId); this.availableCommandsBySession.delete(sessionId); diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 0dff96bfd05..10af2e8cff0 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -266,19 +266,28 @@ describe('SessionRouter', () => { it('passes cwd to bridge.newSession', async () => { const router = new SessionRouter(bridge, '/default'); await router.resolve('ch', 'alice', 'chat1', undefined, '/custom'); - expect(bridge.newSession).toHaveBeenCalledWith('/custom'); + expect(bridge.newSession).toHaveBeenCalledWith( + '/custom', + expect.any(Object), + ); }); it('uses defaultCwd when no cwd provided', async () => { const router = new SessionRouter(bridge, '/default'); await router.resolve('ch', 'alice', 'chat1'); - expect(bridge.newSession).toHaveBeenCalledWith('/default'); + expect(bridge.newSession).toHaveBeenCalledWith( + '/default', + expect.any(Object), + ); }); it('uses defaultCwd when cwd is empty', async () => { const router = new SessionRouter(bridge, '/default'); await router.resolve('ch', 'alice', 'chat1', undefined, ''); - expect(bridge.newSession).toHaveBeenCalledWith('/default'); + expect(bridge.newSession).toHaveBeenCalledWith( + '/default', + expect.any(Object), + ); }); it('deduplicates concurrent session creation for the same route', async () => { @@ -873,6 +882,7 @@ describe('SessionRouter', () => { expect(restartedBridge.loadSession).toHaveBeenCalledWith( aliceSession, '/tmp', + expect.any(Object), ); expect(router.getSession('ch', 'alice', 'chat1')).toBe(aliceSession); expect(router.getSession('ch', 'bob', 'chat2')).toBeUndefined(); @@ -1503,6 +1513,49 @@ describe('SessionRouter', () => { }, ); + it('discards invalidated bindings despite an unrelated hung operation', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const finishFactories: Array< + (session: DaemonChannelSessionClient) => void + > = []; + const factory = vi.fn(() => { + if (factory.mock.calls.length === 1) { + return new Promise(() => undefined); + } + return new Promise((resolve) => { + finishFactories.push(resolve); + }); + }); + const daemonBridge = new DaemonChannelBridge({ + cwd: '/tmp', + sessionFactory: factory, + }); + await daemonBridge.start(); + const router = createLazyRouter(persistPath, daemonBridge); + + const hung = router.resolve('ch', 'hung', 'hung-chat'); + void hung.catch(() => undefined); + const first = router.resolve('ch', 'alice', 'chat1'); + const second = router.resolve('ch', 'bob', 'chat2'); + await drainMicrotasks(); + expect(factory).toHaveBeenCalledTimes(3); + expect(finishFactories).toHaveLength(2); + router.removeSession('ch', 'alice', 'chat1'); + router.removeSession('ch', 'bob', 'chat2'); + finishFactories[0]!(daemonSession('late-alice')); + finishFactories[1]!(daemonSession('late-bob')); + + await expect(first).rejects.toThrow('invalidated'); + await expect(second).rejects.toThrow('invalidated'); + await drainMicrotasks(); + expect(daemonBridge.listSessions()).toEqual([]); + + router.dispose(); + expect(daemonBridge.listSessions()).toEqual([]); + }); + it.each(['removeSession', 'removeSessionId'] as const)( 'rejects a dormant load invalidated by %s', async (removal) => { diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index a866d8bd11c..f96024486ea 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -283,6 +283,7 @@ export class SessionRouter { savedSessionId, savedCwd, this.sessionOptions(input.channelName), + operation, ); try { this.assertOperationCurrent(operation); @@ -589,6 +590,7 @@ export class SessionRouter { entry.sessionId, entry.cwd, options, + operation, ); try { this.assertOperationCurrent(operation); @@ -797,9 +799,7 @@ export class SessionRouter { const maxAttempts = 2; let lastDeadSessionId: string | undefined; for (let attempt = 0; attempt < maxAttempts; attempt++) { - const sessionId = options - ? await this.bridge.newSession(cwd, options) - : await this.bridge.newSession(cwd); + const sessionId = await this.bridge.newSession(cwd, options, operation); try { this.assertOperationCurrent(operation); } catch (error) { @@ -901,27 +901,11 @@ export class SessionRouter { sessionId: string, operation: SessionOperation, ): void { - void this.discardInvalidatedSessionWhenUnowned(sessionId, operation).catch( - () => undefined, - ); - } - - private async discardInvalidatedSessionWhenUnowned( - sessionId: string, - operation: SessionOperation, - ): Promise { - for (;;) { - const possibleOwners = [...this.creatingSessions.values()].filter( - (current) => current !== operation, - ); - if (possibleOwners.length === 0) break; - await Promise.allSettled( - possibleOwners.map((current) => current.promise), - ); - } if ([...this.toSession.values()].includes(sessionId)) return; try { - void this.bridge.discardSession?.(sessionId).catch(() => undefined); + void this.bridge + .discardSession?.(sessionId, operation) + .catch(() => undefined); } catch { // Best-effort cleanup must not replace the terminal invalidation. } From d9b8a67cf4df9c607c31bd3173f35823ae3f8cbd Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 21:21:30 +0800 Subject: [PATCH 12/21] fix(cli): forward daemon session discard --- .../commands/channel/daemon-worker.test.ts | 33 +++++++++++++++++++ .../cli/src/commands/channel/daemon-worker.ts | 4 +++ 2 files changed, 37 insertions(+) diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index 0fe85837a62..f09d10473e2 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -78,6 +78,7 @@ const mockBridgeNewSession = vi.hoisted(() => vi.fn()); const mockBridgeLoadSession = vi.hoisted(() => vi.fn()); const mockBridgePrompt = vi.hoisted(() => vi.fn()); const mockBridgeCancelSession = vi.hoisted(() => vi.fn()); +const mockBridgeDiscardSession = vi.hoisted(() => vi.fn()); const mockBridgeRespondToPermission = vi.hoisted(() => vi.fn()); const mockBridgeShellCommand = vi.hoisted(() => vi.fn()); const mockBridgeGetAvailableCommands = vi.hoisted(() => vi.fn(() => [])); @@ -93,6 +94,7 @@ const mockDaemonChannelBridge = vi.hoisted(() => loadSession: mockBridgeLoadSession, prompt: mockBridgePrompt, cancelSession: mockBridgeCancelSession, + discardSession: mockBridgeDiscardSession, respondToPermission: mockBridgeRespondToPermission, shellCommand: mockBridgeShellCommand, start: mockBridgeStart, @@ -516,6 +518,7 @@ describe('createDaemonChannelBridgeFacade', () => { }); expect('respondToPermission' in facade).toBe(false); + expect('discardSession' in facade).toBe(false); }); it('omits listSessions when absent on bridge', () => { @@ -557,6 +560,36 @@ describe('createDaemonChannelBridgeFacade', () => { }); describe('runChannelDaemonWorker', () => { + it('forwards router discard through the daemon bridge facade', async () => { + const sdk = createSdk(); + const handle = await runChannelDaemonWorker({ + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + loadDaemonSdk: async () => sdk, + }); + const bridgeFacade = mockSessionRouter.mock.calls[0]![0] as { + discardSession?: ( + sessionId: string, + expectedBindingToken?: object, + ) => Promise; + }; + const bindingToken = {}; + + expect(bridgeFacade.discardSession).toBeTypeOf('function'); + await bridgeFacade.discardSession?.('orphan-session', bindingToken); + + expect(mockBridgeDiscardSession).toHaveBeenCalledWith( + 'orphan-session', + bindingToken, + ); + expect(mockBridgeDiscardSession.mock.instances[0]).toBe( + mockDaemonChannelBridge.mock.results[0]!.value, + ); + + await handle.close(); + }); + it('starts selected channels through a daemon-backed bridge facade', async () => { const sdk = createSdk(); const ready = vi.fn(); diff --git a/packages/cli/src/commands/channel/daemon-worker.ts b/packages/cli/src/commands/channel/daemon-worker.ts index d324f507f8e..53319875150 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -179,6 +179,10 @@ export function createDaemonChannelBridgeFacade( facade.respondToPermission = bridge.respondToPermission.bind(bridge); } + if (bridge.discardSession) { + facade.discardSession = bridge.discardSession.bind(bridge); + } + if (bridge.getAvailableCommands) { facade.getAvailableCommands = bridge.getAvailableCommands.bind(bridge); } From 33778de2dd899b0855a82265807560de78335930 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 21:31:08 +0800 Subject: [PATCH 13/21] fix(channels): detach stale daemon clients --- .../base/src/DaemonChannelBridge.test.ts | 50 +++++++++++++++++-- .../channels/base/src/DaemonChannelBridge.ts | 8 ++- 2 files changed, 52 insertions(+), 6 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index 91be058b367..ce8282ce83c 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1542,9 +1542,10 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); - it('rejects and cancels a new session factory result that arrives after stop', async () => { + it('rejects and detaches a new session factory result that arrives after stop', async () => { const events = new EventQueue(); const session = createFakeSession(events); + session.detach = vi.fn().mockResolvedValue(undefined); let finishFactory!: (session: FakeSession) => void; const bridge = new DaemonChannelBridge({ cwd: '/repo', @@ -1563,13 +1564,15 @@ describe('DaemonChannelBridge', () => { finishFactory(session); await expect(creating).rejects.toThrow('stopped'); - expect(session.cancel).toHaveBeenCalledOnce(); + expect(session.detach).toHaveBeenCalledOnce(); + expect(session.cancel).not.toHaveBeenCalled(); expect(bridge.listSessions()).toEqual([]); }); - it('rejects and cancels a load factory result that arrives after stop', async () => { + it('rejects and detaches a load factory result that arrives after stop', async () => { const events = new EventQueue(); const session = createFakeSession(events, 'existing-session'); + session.detach = vi.fn().mockResolvedValue(undefined); let finishFactory!: (session: FakeSession) => void; const bridge = new DaemonChannelBridge({ cwd: '/repo', @@ -1588,10 +1591,45 @@ describe('DaemonChannelBridge', () => { finishFactory(session); await expect(loading).rejects.toThrow('stopped'); - expect(session.cancel).toHaveBeenCalledOnce(); + expect(session.detach).toHaveBeenCalledOnce(); + expect(session.cancel).not.toHaveBeenCalled(); expect(bridge.listSessions()).toEqual([]); }); + it.each(['unavailable', 'rejected'] as const)( + 'falls back to cancel when detach is %s for a stale factory result', + async (detachState) => { + const events = new EventQueue(); + const session = createFakeSession(events, 'stale-session'); + if (detachState === 'rejected') { + session.detach = vi.fn().mockRejectedValue(new Error('detach failed')); + } + let finishFactory!: (session: FakeSession) => void; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + + await bridge.start(); + const creating = bridge.newSession('/repo'); + await Promise.resolve(); + bridge.stop(); + finishFactory(session); + + await expect(creating).rejects.toThrow('stopped'); + if (session.detach) { + expect(session.detach).toHaveBeenCalledOnce(); + } + expect(session.cancel).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([]); + }, + ); + it.each(['new', 'load'] as const)( 'does not attach a %s factory result after a queued stop', async (operation) => { @@ -1628,6 +1666,7 @@ describe('DaemonChannelBridge', () => { it('keeps a pre-stop factory result stale after restart', async () => { const staleEvents = new EventQueue(); const staleSession = createFakeSession(staleEvents, 'stale-session'); + staleSession.detach = vi.fn().mockResolvedValue(undefined); const currentEvents = new EventQueue(); const currentSession = createFakeSession(currentEvents, 'current-session'); let finishStaleFactory!: (session: FakeSession) => void; @@ -1654,7 +1693,8 @@ describe('DaemonChannelBridge', () => { await expect(staleCreation).rejects.toThrow('stopped'); await expect(bridge.newSession('/repo')).resolves.toBe('current-session'); - expect(staleSession.cancel).toHaveBeenCalledOnce(); + expect(staleSession.detach).toHaveBeenCalledOnce(); + expect(staleSession.cancel).not.toHaveBeenCalled(); expect(bridge.listSessions()).toEqual([ { sessionId: 'current-session', diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index 997417d266c..38944603a92 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -412,6 +412,12 @@ export class DaemonChannelBridge } const session = this.removeSessionBinding(sessionId); if (!session) return; + await this.releaseSessionClient(session); + } + + private async releaseSessionClient( + session: DaemonChannelSessionClient, + ): Promise { if (session.detach) { try { await session.detach(); @@ -498,7 +504,7 @@ export class DaemonChannelBridge session: DaemonChannelSessionClient, ): Promise { try { - await session.cancel(); + await this.releaseSessionClient(session); } catch (error) { this.lastError = error; } From 250a36847c2f649a1de35bb71f111fe80569ffa9 Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 21:39:07 +0800 Subject: [PATCH 14/21] fix(channels): reject stale sessions promptly --- .../base/src/DaemonChannelBridge.test.ts | 60 +++++++++++++++++++ .../channels/base/src/DaemonChannelBridge.ts | 6 +- 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index ce8282ce83c..5c0310083cd 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -115,6 +115,12 @@ async function waitFor(assertion: () => void): Promise { throw lastError; } +async function drainMicrotasks(): Promise { + for (let index = 0; index < 20; index++) { + await Promise.resolve(); + } +} + function turnCompleteEvent(sessionId = 'session-1'): DaemonChannelEvent { return { v: 1, @@ -1630,6 +1636,60 @@ describe('DaemonChannelBridge', () => { }, ); + it.each([ + ['new', 'detach'], + ['load', 'cancel'], + ] as const)( + 'rejects stale %s without waiting for hanging %s', + async (operation, cleanup) => { + const events = new EventQueue(); + const sessionId = + operation === 'new' ? 'new-session' : 'existing-session'; + const session = createFakeSession(events, sessionId); + const neverSettles = vi.fn(() => new Promise(() => undefined)); + if (cleanup === 'detach') { + session.detach = neverSettles; + } else { + session.cancel = neverSettles; + } + let finishFactory!: (session: FakeSession) => void; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn( + () => + new Promise((resolve) => { + finishFactory = resolve; + }), + ), + }); + const sessionDied = vi.fn(); + bridge.on('sessionDied', sessionDied); + + await bridge.start(); + const pending = + operation === 'new' + ? bridge.newSession('/repo') + : bridge.loadSession(sessionId, '/repo'); + let rejection: unknown; + void pending.catch((error: unknown) => { + rejection = error; + }); + await Promise.resolve(); + bridge.stop(); + finishFactory(session); + await drainMicrotasks(); + + expect(rejection).toEqual( + expect.objectContaining({ + message: 'Daemon channel bridge stopped during session creation', + }), + ); + expect(neverSettles).toHaveBeenCalledOnce(); + expect(bridge.listSessions()).toEqual([]); + expect(sessionDied).not.toHaveBeenCalled(); + }, + ); + it.each(['new', 'load'] as const)( 'does not attach a %s factory result after a queued stop', async (operation) => { diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index 38944603a92..ee6d24f1f59 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -503,11 +503,9 @@ export class DaemonChannelBridge private async rejectStaleSession( session: DaemonChannelSessionClient, ): Promise { - try { - await this.releaseSessionClient(session); - } catch (error) { + void this.releaseSessionClient(session).catch((error: unknown) => { this.lastError = error; - } + }); throw new Error('Daemon channel bridge stopped during session creation'); } From 64b54ff0cff7028171725a0ce31573361091a42f Mon Sep 17 00:00:00 2001 From: qqqys Date: Fri, 10 Jul 2026 23:31:23 +0800 Subject: [PATCH 15/21] test(channels): cover merged session arguments --- .../channels/base/src/SessionRouter.test.ts | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 10af2e8cff0..1c0422ae5bd 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -134,9 +134,11 @@ describe('SessionRouter', () => { await router.resolve('ch', 'alice', 'chat1'); - expect(bridge.newSession).toHaveBeenCalledWith('/tmp', { - approvalMode: 'yolo', - }); + expect(bridge.newSession).toHaveBeenCalledWith( + '/tmp', + { approvalMode: 'yolo' }, + expect.any(Object), + ); }); it('user scope: same sender+chat reuses session', async () => { @@ -268,6 +270,7 @@ describe('SessionRouter', () => { await router.resolve('ch', 'alice', 'chat1', undefined, '/custom'); expect(bridge.newSession).toHaveBeenCalledWith( '/custom', + undefined, expect.any(Object), ); }); @@ -277,6 +280,7 @@ describe('SessionRouter', () => { await router.resolve('ch', 'alice', 'chat1'); expect(bridge.newSession).toHaveBeenCalledWith( '/default', + undefined, expect.any(Object), ); }); @@ -286,6 +290,7 @@ describe('SessionRouter', () => { await router.resolve('ch', 'alice', 'chat1', undefined, ''); expect(bridge.newSession).toHaveBeenCalledWith( '/default', + undefined, expect.any(Object), ); }); @@ -713,9 +718,12 @@ describe('SessionRouter', () => { failed: 0, }); - expect(bridge.loadSession).toHaveBeenCalledWith('old-session', '/tmp', { - approvalMode: 'yolo', - }); + expect(bridge.loadSession).toHaveBeenCalledWith( + 'old-session', + '/tmp', + { approvalMode: 'yolo' }, + expect.any(Object), + ); }); it('logs malformed persisted session files', async () => { @@ -882,6 +890,7 @@ describe('SessionRouter', () => { expect(restartedBridge.loadSession).toHaveBeenCalledWith( aliceSession, '/tmp', + undefined, expect.any(Object), ); expect(router.getSession('ch', 'alice', 'chat1')).toBe(aliceSession); From 880f3ec6967df5a459d7b5e983c24212d9b20b7e Mon Sep 17 00:00:00 2001 From: qqqys Date: Sat, 11 Jul 2026 06:04:13 +0800 Subject: [PATCH 16/21] fix(channels): update session cleanup test mock --- packages/cli/src/commands/channel/start.test.ts | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/commands/channel/start.test.ts b/packages/cli/src/commands/channel/start.test.ts index 75fb16e73c0..8a9c8cc4017 100644 --- a/packages/cli/src/commands/channel/start.test.ts +++ b/packages/cli/src/commands/channel/start.test.ts @@ -60,7 +60,7 @@ const mockSanitizeLogText = vi.hoisted(() => ); const mockRouterClearAll = vi.hoisted(() => vi.fn()); const mockRouterGetTarget = vi.hoisted(() => vi.fn()); -const mockRouterRemoveSessionId = vi.hoisted(() => vi.fn()); +const mockRouterHandleSessionDied = vi.hoisted(() => vi.fn()); const mockRouterRestoreSessions = vi.hoisted(() => vi.fn()); const mockRouterSetBridge = vi.hoisted(() => vi.fn()); const mockRouterSetChannelScope = vi.hoisted(() => vi.fn()); @@ -88,7 +88,7 @@ const mockSessionRouter = vi.hoisted(() => vi.fn(() => ({ clearAll: mockRouterClearAll, getTarget: mockRouterGetTarget, - removeSessionId: mockRouterRemoveSessionId, + handleSessionDied: mockRouterHandleSessionDied, restoreSessions: mockRouterRestoreSessions, setBridge: mockRouterSetBridge, setChannelScope: mockRouterSetChannelScope, @@ -652,9 +652,9 @@ describe('startCommand.handler', () => { expect(mockSanitizeLogText).toHaveBeenCalledWith('dead\nsession', 128); expect(mockSanitizeLogText).toHaveBeenCalledWith('boom\nreason', 512); expect(mockWriteStderrLine).toHaveBeenCalledWith( - '[Channel] Session dead\\nsession died (boom\\nreason), removing routing state', + '[Channel] Session dead\\nsession died (boom\\nreason), updating routing state', ); - expect(mockRouterRemoveSessionId).toHaveBeenCalledWith('dead\nsession'); + expect(mockRouterHandleSessionDied).toHaveBeenCalledWith('dead\nsession'); expect(mockChannelOnSessionDied).not.toHaveBeenCalled(); }); @@ -734,7 +734,7 @@ describe('startCommand.handler', () => { sessionDiedListener!({ sessionId: 'dead-session' }); expect(mockChannelOnSessionDied).toHaveBeenCalledWith('dead-session'); - expect(mockRouterRemoveSessionId).not.toHaveBeenCalled(); + expect(mockRouterHandleSessionDied).not.toHaveBeenCalled(); }); it('registers session cleanup on the replacement bridge before restoring sessions', async () => { @@ -780,7 +780,7 @@ describe('startCommand.handler', () => { restartedSessionDiedListener({ sessionId: 'dead-after-restart' }); - expect(mockRouterRemoveSessionId).toHaveBeenCalledWith( + expect(mockRouterHandleSessionDied).toHaveBeenCalledWith( 'dead-after-restart', ); } finally { From 10483b629671af04c6f8c0990732f18577720220 Mon Sep 17 00:00:00 2001 From: qqqys Date: Sat, 11 Jul 2026 07:04:41 +0800 Subject: [PATCH 17/21] test(channels): update Telegram session cleanup mock --- packages/channels/telegram/src/TelegramAdapter.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/channels/telegram/src/TelegramAdapter.test.ts b/packages/channels/telegram/src/TelegramAdapter.test.ts index 284e7aa42f9..a305bffe99c 100644 --- a/packages/channels/telegram/src/TelegramAdapter.test.ts +++ b/packages/channels/telegram/src/TelegramAdapter.test.ts @@ -239,7 +239,8 @@ describe('TelegramChannel', () => { }); it('clears typing when a session dies without a terminal event', () => { - const channel = createChannel({}, { removeSessionId: vi.fn() }); + const handleSessionDied = vi.fn(); + const channel = createChannel({}, { handleSessionDied }); const bot = installFakeBot(channel); channel.emitLifecycle({ @@ -255,6 +256,8 @@ describe('TelegramChannel', () => { channel.onSessionDied('session-1'); + expect(handleSessionDied).toHaveBeenCalledWith('session-1'); + vi.advanceTimersByTime(4000); expect(bot.api.sendChatAction).toHaveBeenCalledTimes(1); }); From 6da5f3d9bcbaa5d916941c7f1c2e31a13302c54a Mon Sep 17 00:00:00 2001 From: qqqys Date: Sat, 11 Jul 2026 16:10:19 +0800 Subject: [PATCH 18/21] fix(channels): release mismatched daemon sessions --- packages/channels/base/src/DaemonChannelBridge.test.ts | 2 ++ packages/channels/base/src/DaemonChannelBridge.ts | 3 +++ 2 files changed, 5 insertions(+) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index ec2532ff0ee..c962267c80d 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1870,6 +1870,7 @@ describe('DaemonChannelBridge', () => { it('rejects mismatched daemon session ids while loading', async () => { const events = new EventQueue(); const session = createFakeSession(events, 'different-session'); + session.detach = vi.fn().mockResolvedValue(undefined); const bridge = new DaemonChannelBridge({ cwd: '/repo', sessionFactory: vi.fn().mockResolvedValue(session), @@ -1884,6 +1885,7 @@ describe('DaemonChannelBridge', () => { await expect(bridge.prompt('different-session', 'hello')).rejects.toThrow( 'No daemon session bound for different-session', ); + expect(session.detach).toHaveBeenCalledOnce(); events.close(); bridge.stop(); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index cae26190cf4..7023ab0732b 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -288,6 +288,9 @@ export class DaemonChannelBridge await this.rejectStaleSession(session); } if (session.sessionId !== sessionId) { + void this.releaseSessionClient(session).catch((error: unknown) => { + this.lastError = error; + }); throw new Error( `Daemon returned session ${session.sessionId} while loading ${sessionId}`, ); From 544052cd36f24d870ffb70d433479fda21720808 Mon Sep 17 00:00:00 2001 From: qqqys Date: Sat, 11 Jul 2026 18:05:49 +0800 Subject: [PATCH 19/21] fix(channels): release replaced daemon sessions --- .../channels/base/src/DaemonChannelBridge.test.ts | 1 + packages/channels/base/src/DaemonChannelBridge.ts | 13 +++++++++++-- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index c962267c80d..32cb7a79ddb 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1228,6 +1228,7 @@ describe('DaemonChannelBridge', () => { reason: 'session_replaced', }), ); + await waitFor(() => expect(firstSession.cancel).toHaveBeenCalledOnce()); await expect( bridge.respondToPermission('req-1', { outcome: { outcome: 'selected', optionId: 'proceed_once' }, diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index 7023ab0732b..b66aaf26233 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -492,8 +492,17 @@ export class DaemonChannelBridge session: DaemonChannelSessionClient, bindingToken?: object, ): void { - if (this.sessions.has(session.sessionId)) { - this.dropSession(session.sessionId, 'session_replaced'); + const replacedSession = this.removeSessionBinding(session.sessionId); + if (replacedSession) { + void this.releaseSessionClient(replacedSession).catch( + (error: unknown) => { + this.lastError = error; + }, + ); + this.emit('sessionDied', { + sessionId: session.sessionId, + reason: 'session_replaced', + }); } this.sessions.set(session.sessionId, session); From db2d803dd499476082e2afd15f7ae289bf1fc296 Mon Sep 17 00:00:00 2001 From: qqqys Date: Sat, 11 Jul 2026 23:11:20 +0800 Subject: [PATCH 20/21] fix(channels): release dropped daemon sessions --- .../base/src/DaemonChannelBridge.test.ts | 25 +++++++++++++++++++ .../channels/base/src/DaemonChannelBridge.ts | 16 +++++++++--- 2 files changed, 38 insertions(+), 3 deletions(-) diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index 32cb7a79ddb..c733b7e0566 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -2074,6 +2074,31 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); + it('releases a session client when the daemon reports it dead', async () => { + const events = new EventQueue(); + const session = createFakeSession(events); + const detach = vi.fn().mockResolvedValue(undefined); + session.detach = detach; + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn().mockResolvedValue(session), + }); + await bridge.start(); + + await bridge.newSession('/repo'); + events.push({ + id: 1, + v: 1, + type: 'session_died', + data: { reason: 'gone' }, + }); + + await waitFor(() => expect(detach).toHaveBeenCalledOnce()); + + events.close(); + bridge.stop(); + }); + it('listSessions shows hasActivePrompt false after cancelSession', async () => { const events = new EventQueue(); const session = createFakeSession(events); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index b66aaf26233..d553a835a29 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -478,7 +478,7 @@ export class DaemonChannelBridge this.lastError = error; }); } - this.dropSession(sessionId, 'bridge_stopped'); + this.dropSession(sessionId, 'bridge_stopped', false); } this.latestAvailableCommandsSessionId = undefined; this.connected = false; @@ -810,8 +810,18 @@ export class DaemonChannelBridge ); } - private dropSession(sessionId: string, reason: string): void { - if (!this.removeSessionBinding(sessionId)) return; + private dropSession( + sessionId: string, + reason: string, + releaseClient = true, + ): void { + const session = this.removeSessionBinding(sessionId); + if (!session) return; + if (releaseClient) { + void this.releaseSessionClient(session).catch((error: unknown) => { + this.lastError = error; + }); + } this.emit('sessionDied', { sessionId, reason }); } From 4f99a607be989325a0b6985022201076ed3e7e07 Mon Sep 17 00:00:00 2001 From: qqqys Date: Sun, 12 Jul 2026 01:05:38 +0800 Subject: [PATCH 21/21] fix(channels): guard lazy route recovery --- packages/channels/base/src/SessionRouter.test.ts | 12 ++++++++++++ packages/channels/base/src/SessionRouter.ts | 3 +++ 2 files changed, 15 insertions(+) diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 1c0422ae5bd..00fd0f0dfe8 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -1263,6 +1263,18 @@ describe('SessionRouter', () => { }); describe('lazy recovery', () => { + it('rejects route restoration outside lazy recovery mode', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + + expect(() => router.restoreRoutes()).toThrow( + 'restoreRoutes requires lazy recovery mode', + ); + }); + function createLazyRouter(persistPath: string, customBridge = bridge) { return new SessionRouter(customBridge, '/tmp', 'user', persistPath, { recoveryMode: 'lazy', diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index f96024486ea..617c251068a 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -521,6 +521,9 @@ export class SessionRouter { } restoreRoutes(): { restored: number; dropped: number } { + if (this.recoveryMode !== 'lazy') { + throw new Error('restoreRoutes requires lazy recovery mode'); + } const persisted = this.readPersistedEntries(); if (!persisted) return { restored: 0, dropped: 0 }; this.dispose();