From 49481123f5b6f69f53a07ec5a5b2f4087f6098ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?= =?UTF-8?q?=29?= Date: Sun, 24 May 2026 15:51:36 +0800 Subject: [PATCH] fix: recover stale IM session bindings IM adapters could keep in-process bridge state after desktop session deletion removed the persisted mapping or after the server no longer had the mapped session. The restore path now validates the stored mapping against bridge memory and the server session before reusing it, then clears stale transient state before creating a replacement session. Constraint: Issue #574 is intermittent and tied to adapter-created sessions; live reporter Telegram credentials are not available locally. Rejected: Keep separate recovery checks in each IM adapter | duplicates the same stale-state logic across Telegram, Feishu, WeChat, and DingTalk. Confidence: high Scope-risk: moderate Directive: Do not use bridge.hasSession() as the sole restore check; verify adapter-sessions state and server session existence first. Tested: bun run check:adapters (364 pass, 0 fail) Tested: cd adapters && bunx tsc --noEmit Tested: bun test src/server/__tests__/sessions.test.ts src/server/__tests__/websocket-handler.test.ts --timeout 30000 (110 pass, 0 fail) Tested: git diff --check Not-tested: Live Telegram bot/provider reproduction on the reporter machine --- adapters/common/__tests__/http-client.test.ts | 14 ++ .../common/__tests__/session-recovery.test.ts | 163 ++++++++++++++++++ adapters/common/__tests__/ws-bridge.test.ts | 2 + adapters/common/http-client.ts | 17 ++ adapters/common/session-recovery.ts | 83 +++++++++ adapters/common/ws-bridge.ts | 11 ++ adapters/dingtalk/index.ts | 31 ++-- adapters/feishu/index.ts | 31 ++-- adapters/telegram/index.ts | 31 ++-- adapters/wechat/index.ts | 31 ++-- 10 files changed, 338 insertions(+), 76 deletions(-) create mode 100644 adapters/common/__tests__/session-recovery.test.ts create mode 100644 adapters/common/session-recovery.ts diff --git a/adapters/common/__tests__/http-client.test.ts b/adapters/common/__tests__/http-client.test.ts index e4c83622..4135aa6e 100644 --- a/adapters/common/__tests__/http-client.test.ts +++ b/adapters/common/__tests__/http-client.test.ts @@ -106,6 +106,20 @@ describe('AdapterHttpClient', () => { expect(client.createSession('')).rejects.toThrow() }) + it('sessionExists returns false for deleted sessions', async () => { + globalThis.fetch = mock(() => + Promise.resolve(new Response(JSON.stringify({ error: 'NOT_FOUND' }), { + status: 404, + headers: { 'Content-Type': 'application/json' }, + })) + ) as any + + await expect(client.sessionExists('deleted-session')).resolves.toBe(false) + expect((globalThis.fetch as any).mock.calls[0][0]).toBe( + 'http://127.0.0.1:3456/api/sessions/deleted-session', + ) + }) + it('getGitInfo calls GET /api/sessions/:id/git-info', async () => { globalThis.fetch = mock(() => Promise.resolve(new Response(JSON.stringify({ diff --git a/adapters/common/__tests__/session-recovery.test.ts b/adapters/common/__tests__/session-recovery.test.ts new file mode 100644 index 00000000..d3d0d956 --- /dev/null +++ b/adapters/common/__tests__/session-recovery.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it } from 'bun:test' +import { restoreStoredSessionBinding } from '../session-recovery.js' +import type { SessionEntry } from '../session-store.js' + +function makeStore(initial: SessionEntry | null) { + let entry = initial + return { + get() { + return entry + }, + delete() { + entry = null + }, + current() { + return entry + }, + } +} + +function makeBridge(options?: { currentSessionId?: string | null; open?: boolean; hasSession?: boolean }) { + const calls: string[] = [] + const bridge = { + calls, + connectSession(_chatId: string, sessionId: string) { + calls.push(`connect:${sessionId}`) + return true + }, + getSessionId() { + return options?.currentSessionId ?? null + }, + hasSession() { + return options?.hasSession ?? Boolean(options?.currentSessionId) + }, + isSessionOpen(_chatId: string, sessionId?: string) { + return Boolean(options?.open && (!sessionId || sessionId === options?.currentSessionId)) + }, + onServerMessage() { + calls.push('handler') + }, + resetSession() { + calls.push('reset') + }, + waitForOpen() { + calls.push('wait') + return Promise.resolve(true) + }, + } + return bridge +} + +describe('restoreStoredSessionBinding', () => { + it('resets stale bridge memory when server-side delete removed the stored mapping', async () => { + const store = makeStore(null) + const bridge = makeBridge({ currentSessionId: 'deleted-session', hasSession: true }) + let cleared = 0 + + const restored = await restoreStoredSessionBinding({ + chatId: 'chat-1', + bridge, + sessionStore: store, + httpClient: { sessionExists: async () => true }, + onServerMessage: () => {}, + logPrefix: '[Test]', + clearTransientState: () => { + cleared += 1 + }, + }) + + expect(restored).toBeNull() + expect(bridge.calls).toEqual(['reset']) + expect(cleared).toBe(1) + }) + + it('drops a persisted mapping when the server no longer has that session', async () => { + const entry = { sessionId: 'deleted-session', workDir: '/repo', updatedAt: 1 } + const store = makeStore(entry) + const bridge = makeBridge() + let cleared = 0 + + const restored = await restoreStoredSessionBinding({ + chatId: 'chat-1', + bridge, + sessionStore: store, + httpClient: { sessionExists: async () => false }, + onServerMessage: () => {}, + logPrefix: '[Test]', + clearTransientState: () => { + cleared += 1 + }, + }) + + expect(restored).toBeNull() + expect(store.current()).toBeNull() + expect(bridge.calls).toEqual([]) + expect(cleared).toBe(1) + }) + + it('resets bridge memory when it points at a different session than the store', async () => { + const entry = { sessionId: 'stored-session', workDir: '/repo', updatedAt: 1 } + const store = makeStore(entry) + const bridge = makeBridge({ currentSessionId: 'stale-session', hasSession: true }) + let cleared = 0 + + const restored = await restoreStoredSessionBinding({ + chatId: 'chat-1', + bridge, + sessionStore: store, + httpClient: { sessionExists: async () => true }, + onServerMessage: () => {}, + logPrefix: '[Test]', + clearTransientState: () => { + cleared += 1 + }, + }) + + expect(restored).toEqual(entry) + expect(bridge.calls).toEqual(['reset', 'connect:stored-session', 'handler', 'wait']) + expect(cleared).toBe(1) + }) + + it('reuses an already-open bridge for the stored session', async () => { + const entry = { sessionId: 'live-session', workDir: '/repo', updatedAt: 1 } + const store = makeStore(entry) + const bridge = makeBridge({ currentSessionId: 'live-session', open: true }) + let checked = 0 + + const restored = await restoreStoredSessionBinding({ + chatId: 'chat-1', + bridge, + sessionStore: store, + httpClient: { + sessionExists: async () => { + checked += 1 + return true + }, + }, + onServerMessage: () => {}, + logPrefix: '[Test]', + }) + + expect(restored).toEqual(entry) + expect(checked).toBe(0) + expect(bridge.calls).toEqual([]) + }) + + it('connects to an existing stored session after validating it still exists', async () => { + const entry = { sessionId: 'stored-session', workDir: '/repo', updatedAt: 1 } + const store = makeStore(entry) + const bridge = makeBridge() + + const restored = await restoreStoredSessionBinding({ + chatId: 'chat-1', + bridge, + sessionStore: store, + httpClient: { sessionExists: async () => true }, + onServerMessage: () => {}, + logPrefix: '[Test]', + }) + + expect(restored).toEqual(entry) + expect(bridge.calls).toEqual(['connect:stored-session', 'handler', 'wait']) + }) +}) diff --git a/adapters/common/__tests__/ws-bridge.test.ts b/adapters/common/__tests__/ws-bridge.test.ts index 7e8618b2..787be014 100644 --- a/adapters/common/__tests__/ws-bridge.test.ts +++ b/adapters/common/__tests__/ws-bridge.test.ts @@ -17,6 +17,8 @@ describe('WsBridge', () => { const result = bridge.connectSession('chat-1', 'my-uuid-session-id') expect(result).toBe(true) expect(bridge.hasSession('chat-1')).toBe(true) + expect(bridge.getSessionId('chat-1')).toBe('my-uuid-session-id') + expect(bridge.isSessionOpen('chat-1', 'my-uuid-session-id')).toBe(false) }) it('connectSession for different chatIds creates separate sessions', () => { diff --git a/adapters/common/http-client.ts b/adapters/common/http-client.ts index 61bd0bcb..4c006d96 100644 --- a/adapters/common/http-client.ts +++ b/adapters/common/http-client.ts @@ -72,6 +72,23 @@ export class AdapterHttpClient { } } + async sessionExists(sessionId: string): Promise { + const { controller, timer } = this.createTimeoutController() + try { + const res = await fetch(`${this.httpBaseUrl}/api/sessions/${encodeURIComponent(sessionId)}`, { + signal: controller.signal, + }) + if (res.status === 404) return false + if (!res.ok) { + const err = await res.json().catch(() => ({ message: res.statusText })) + throw new Error(`Failed to check session: ${(err as any).message}`) + } + return true + } finally { + clearTimeout(timer) + } + } + async listRecentProjects(): Promise { const { controller, timer } = this.createTimeoutController() try { diff --git a/adapters/common/session-recovery.ts b/adapters/common/session-recovery.ts new file mode 100644 index 00000000..fa44f6d2 --- /dev/null +++ b/adapters/common/session-recovery.ts @@ -0,0 +1,83 @@ +import type { AdapterHttpClient } from './http-client.js' +import type { SessionEntry, SessionStore } from './session-store.js' +import type { ServerMessage, WsBridge } from './ws-bridge.js' + +type BridgeSessionOps = Pick< + WsBridge, + | 'connectSession' + | 'getSessionId' + | 'hasSession' + | 'isSessionOpen' + | 'onServerMessage' + | 'resetSession' + | 'waitForOpen' +> + +type RestoreStoredSessionBindingOptions = { + chatId: string + bridge: BridgeSessionOps + sessionStore: Pick + httpClient: Pick + onServerMessage: (msg: ServerMessage) => void | Promise + logPrefix: string + clearTransientState?: () => void +} + +function resetStaleBridge( + chatId: string, + bridge: BridgeSessionOps, + clearTransientState?: () => void, +): void { + if (!bridge.hasSession(chatId)) return + bridge.resetSession(chatId) + clearTransientState?.() +} + +export async function restoreStoredSessionBinding({ + chatId, + bridge, + sessionStore, + httpClient, + onServerMessage, + logPrefix, + clearTransientState, +}: RestoreStoredSessionBindingOptions): Promise { + const stored = sessionStore.get(chatId) + if (!stored) { + resetStaleBridge(chatId, bridge, clearTransientState) + return null + } + + const currentSessionId = bridge.getSessionId(chatId) + if (currentSessionId && currentSessionId !== stored.sessionId) { + resetStaleBridge(chatId, bridge, clearTransientState) + } + + if (bridge.isSessionOpen(chatId, stored.sessionId)) { + return stored + } + + let exists = true + try { + exists = await httpClient.sessionExists(stored.sessionId) + } catch (err) { + console.warn( + `${logPrefix} Failed to verify stored session ${stored.sessionId}: ${ + err instanceof Error ? err.message : String(err) + }`, + ) + } + + if (!exists) { + sessionStore.delete(chatId) + const hadBridgeSession = bridge.hasSession(chatId) + resetStaleBridge(chatId, bridge, clearTransientState) + if (!hadBridgeSession) clearTransientState?.() + return null + } + + bridge.connectSession(chatId, stored.sessionId) + bridge.onServerMessage(chatId, onServerMessage) + const opened = await bridge.waitForOpen(chatId) + return opened ? stored : null +} diff --git a/adapters/common/ws-bridge.ts b/adapters/common/ws-bridge.ts index 2ccc6ca5..21e633db 100644 --- a/adapters/common/ws-bridge.ts +++ b/adapters/common/ws-bridge.ts @@ -113,6 +113,17 @@ export class WsBridge { this.handlers.set(chatId, handler) } + getSessionId(chatId: string): string | null { + return this.sessions.get(chatId)?.sessionId ?? null + } + + isSessionOpen(chatId: string, sessionId?: string): boolean { + const session = this.sessions.get(chatId) + if (!session) return false + if (sessionId && session.sessionId !== sessionId) return false + return session.ws.readyState === WebSocket.OPEN + } + /** Reset session for a chatId (e.g. /new command). */ resetSession(chatId: string): void { const session = this.sessions.get(chatId) diff --git a/adapters/dingtalk/index.ts b/adapters/dingtalk/index.ts index 4b2bce32..c74f15ca 100644 --- a/adapters/dingtalk/index.ts +++ b/adapters/dingtalk/index.ts @@ -21,6 +21,7 @@ import { } from '../common/permission.js' import { SessionStore } from '../common/session-store.js' import { AdapterHttpClient, type RecentProject } from '../common/http-client.js' +import { restoreStoredSessionBinding } from '../common/session-recovery.js' import { isAllowedUser, tryPair } from '../common/pairing.js' import { AttachmentStore } from '../common/attachment/attachment-store.js' import { checkAttachmentLimit } from '../common/attachment/attachment-limits.js' @@ -222,17 +223,15 @@ function clearPendingPermissions(chatId: string): void { } async function ensureExistingSession(chatId: string): Promise<{ sessionId: string; workDir: string } | null> { - const stored = sessionStore.get(chatId) - if (!stored) return null - - if (!bridge.hasSession(chatId)) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - const opened = await bridge.waitForOpen(chatId) - if (!opened) return null - } - - return stored + return await restoreStoredSessionBinding({ + chatId, + bridge, + sessionStore, + httpClient, + onServerMessage: (msg) => handleServerMessage(chatId, msg), + logPrefix: '[DingTalk]', + clearTransientState: () => clearTransientChatState(chatId), + }) } async function buildStatusText(chatId: string): Promise { @@ -287,14 +286,8 @@ async function buildStatusText(chatId: string): Promise { } async function ensureSession(chatId: string): Promise { - if (bridge.hasSession(chatId)) return true - - const stored = sessionStore.get(chatId) - if (stored) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - return await bridge.waitForOpen(chatId) - } + const stored = await ensureExistingSession(chatId) + if (stored) return true return await createSessionForChat(chatId, defaultWorkDir) } diff --git a/adapters/feishu/index.ts b/adapters/feishu/index.ts index bea9e226..e50f09d8 100644 --- a/adapters/feishu/index.ts +++ b/adapters/feishu/index.ts @@ -29,6 +29,7 @@ import { } from '../common/permission.js' import { SessionStore } from '../common/session-store.js' import { AdapterHttpClient, type RecentProject } from '../common/http-client.js' +import { restoreStoredSessionBinding } from '../common/session-recovery.js' import { isAllowedUser, tryPair } from '../common/pairing.js' import { optimizeMarkdownForFeishu } from './markdown-style.js' import { extractInboundPayload } from './extract-payload.js' @@ -218,17 +219,15 @@ function clearTransientChatState(chatId: string): void { } async function ensureExistingSession(chatId: string): Promise<{ sessionId: string; workDir: string } | null> { - const stored = sessionStore.get(chatId) - if (!stored) return null - - if (!bridge.hasSession(chatId)) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - const opened = await bridge.waitForOpen(chatId) - if (!opened) return null - } - - return stored + return await restoreStoredSessionBinding({ + chatId, + bridge, + sessionStore, + httpClient, + onServerMessage: (msg) => handleServerMessage(chatId, msg), + logPrefix: '[Feishu]', + clearTransientState: () => clearTransientChatState(chatId), + }) } async function buildStatusText(chatId: string): Promise { @@ -646,14 +645,8 @@ function buildPermissionCard( // ---------- session management ---------- async function ensureSession(chatId: string): Promise { - if (bridge.hasSession(chatId)) return true - - const stored = sessionStore.get(chatId) - if (stored) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - return await bridge.waitForOpen(chatId) - } + const stored = await ensureExistingSession(chatId) + if (stored) return true const workDir = defaultWorkDir if (workDir) { diff --git a/adapters/telegram/index.ts b/adapters/telegram/index.ts index f38fbd46..3da671e2 100644 --- a/adapters/telegram/index.ts +++ b/adapters/telegram/index.ts @@ -33,6 +33,7 @@ import { } from '../common/permission.js' import { SessionStore } from '../common/session-store.js' import { AdapterHttpClient } from '../common/http-client.js' +import { restoreStoredSessionBinding } from '../common/session-recovery.js' import { isAllowedUser, tryPair } from '../common/pairing.js' import { TelegramMediaService } from './media.js' import { AttachmentStore } from '../common/attachment/attachment-store.js' @@ -150,17 +151,15 @@ async function handlePermissionDecision(chatId: string, decision: PermissionDeci } async function ensureExistingSession(chatId: string): Promise<{ sessionId: string; workDir: string } | null> { - const stored = sessionStore.get(chatId) - if (!stored) return null - - if (!bridge.hasSession(chatId)) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - const opened = await bridge.waitForOpen(chatId) - if (!opened) return null - } - - return stored + return await restoreStoredSessionBinding({ + chatId, + bridge, + sessionStore, + httpClient, + onServerMessage: (msg) => handleServerMessage(chatId, msg), + logPrefix: '[Telegram]', + clearTransientState: () => clearTransientChatState(chatId), + }) } async function buildStatusText(chatId: string): Promise { @@ -283,14 +282,8 @@ async function flushToTelegram(chatId: string, newText: string, isComplete: bool // ---------- session management ---------- async function ensureSession(chatId: string): Promise { - if (bridge.hasSession(chatId)) return true - - const stored = sessionStore.get(chatId) - if (stored) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - return await bridge.waitForOpen(chatId) - } + const stored = await ensureExistingSession(chatId) + if (stored) return true const workDir = defaultWorkDir if (workDir) { diff --git a/adapters/wechat/index.ts b/adapters/wechat/index.ts index b48d38ef..16dfe468 100644 --- a/adapters/wechat/index.ts +++ b/adapters/wechat/index.ts @@ -17,6 +17,7 @@ import { } from '../common/permission.js' import { SessionStore } from '../common/session-store.js' import { AdapterHttpClient } from '../common/http-client.js' +import { restoreStoredSessionBinding } from '../common/session-recovery.js' import { isAllowedUser, tryPair } from '../common/pairing.js' import { AttachmentStore } from '../common/attachment/attachment-store.js' import { checkAttachmentLimit } from '../common/attachment/attachment-limits.js' @@ -203,17 +204,15 @@ function enqueueWechat(chatId: string, task: () => Promise): void { } async function ensureExistingSession(chatId: string): Promise<{ sessionId: string; workDir: string } | null> { - const stored = sessionStore.get(chatId) - if (!stored) return null - - if (!bridge.hasSession(chatId)) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - const opened = await bridge.waitForOpen(chatId) - if (!opened) return null - } - - return stored + return await restoreStoredSessionBinding({ + chatId, + bridge, + sessionStore, + httpClient, + onServerMessage: (msg) => handleServerMessage(chatId, msg), + logPrefix: '[WeChat]', + clearTransientState: () => clearTransientChatState(chatId), + }) } async function buildStatusText(chatId: string): Promise { @@ -268,14 +267,8 @@ async function buildStatusText(chatId: string): Promise { } async function ensureSession(chatId: string): Promise { - if (bridge.hasSession(chatId)) return true - - const stored = sessionStore.get(chatId) - if (stored) { - bridge.connectSession(chatId, stored.sessionId) - bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg)) - return await bridge.waitForOpen(chatId) - } + const stored = await ensureExistingSession(chatId) + if (stored) return true const workDir = defaultWorkDir if (workDir) return await createSessionForChat(chatId, workDir)