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)