From aa3cba333496618f60d1c668b95b4e994719d58a 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: Mon, 1 Jun 2026 16:36:33 +0800 Subject: [PATCH] fix: recover stale auth and sockets in desktop chats (#658, #651) Desktop official OAuth sessions inject a token into a long-lived CLI process to avoid OS credential prompts. After Windows sleep/wake that env token can become stale, so the server now pushes a refreshed token before each resumed user turn and the CLI clears its OAuth cache when the env token changes. The retry path also disables keep-alive after ECONNRESET/EPIPE by default, matching the stale pooled-socket failure mode reported on #651. Constraint: Real Win11 sleep/wake cannot be physically reproduced in this macOS worktree Rejected: Restart the CLI before every user turn | would add latency and discard useful process continuity Confidence: medium Scope-risk: moderate Directive: Keep official OAuth env-token updates paired with CLI-side OAuth cache invalidation Tested: bun test src/server/__tests__/conversation-service.test.ts --timeout 30000 Tested: bun test src/cli/__tests__/structuredIO.test.ts --timeout 30000 Tested: bun test src/services/api/withRetry.test.ts --timeout 30000 Tested: bun run check:server Tested: git diff --check Not-tested: Real Win11 sleep/wake with Claude official OAuth Not-tested: External ccswitch reqwest pool reset path when the reset is hidden behind the proxy Related: #658 Related: #651 --- src/cli/__tests__/structuredIO.test.ts | 40 +++++++++++++++ src/cli/structuredIO.ts | 4 ++ .../__tests__/conversation-service.test.ts | 45 +++++++++++++++++ src/server/services/conversationService.ts | 36 ++++++++++++++ src/services/api/withRetry.test.ts | 49 +++++++++++++++++++ src/services/api/withRetry.ts | 9 +--- 6 files changed, 175 insertions(+), 8 deletions(-) create mode 100644 src/cli/__tests__/structuredIO.test.ts create mode 100644 src/services/api/withRetry.test.ts diff --git a/src/cli/__tests__/structuredIO.test.ts b/src/cli/__tests__/structuredIO.test.ts new file mode 100644 index 00000000..43f7c1b3 --- /dev/null +++ b/src/cli/__tests__/structuredIO.test.ts @@ -0,0 +1,40 @@ +import { afterEach, describe, expect, test } from 'bun:test' +import { StructuredIO } from '../structuredIO.js' +import { + clearOAuthTokenCache, + getClaudeAIOAuthTokens, +} from '../../utils/auth.js' + +describe('StructuredIO environment updates', () => { + const originalOAuthToken = process.env.CLAUDE_CODE_OAUTH_TOKEN + + afterEach(() => { + if (originalOAuthToken === undefined) { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN + } else { + process.env.CLAUDE_CODE_OAUTH_TOKEN = originalOAuthToken + } + clearOAuthTokenCache() + }) + + test('clears OAuth token cache when CLAUDE_CODE_OAUTH_TOKEN changes at runtime', async () => { + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'stale-env-token' + clearOAuthTokenCache() + expect(getClaudeAIOAuthTokens()?.accessToken).toBe('stale-env-token') + + async function* input() { + yield `${JSON.stringify({ + type: 'update_environment_variables', + variables: { CLAUDE_CODE_OAUTH_TOKEN: 'fresh-env-token' }, + })}\n` + } + + const io = new StructuredIO(input()) + for await (const _message of io.structuredInput) { + // update_environment_variables messages are consumed internally. + } + + expect(process.env.CLAUDE_CODE_OAUTH_TOKEN).toBe('fresh-env-token') + expect(getClaudeAIOAuthTokens()?.accessToken).toBe('fresh-env-token') + }) +}) diff --git a/src/cli/structuredIO.ts b/src/cli/structuredIO.ts index 366b56f5..c70b3f70 100644 --- a/src/cli/structuredIO.ts +++ b/src/cli/structuredIO.ts @@ -354,6 +354,10 @@ export class StructuredIO { for (const [key, value] of Object.entries(message.variables)) { process.env[key] = value } + if (Object.hasOwn(message.variables, 'CLAUDE_CODE_OAUTH_TOKEN')) { + const { clearOAuthTokenCache } = await import('../utils/auth.js') + clearOAuthTokenCache() + } logForDebugging( `[structuredIO] applied update_environment_variables: ${keys.join(', ')}`, ) diff --git a/src/server/__tests__/conversation-service.test.ts b/src/server/__tests__/conversation-service.test.ts index 064bce63..7c7f6d3b 100644 --- a/src/server/__tests__/conversation-service.test.ts +++ b/src/server/__tests__/conversation-service.test.ts @@ -260,6 +260,51 @@ describe('ConversationService', () => { expect(env.CLAUDE_CODE_OAUTH_TOKEN).toBe('haha-fresh-token') }) + test('sendMessage updates a running official OAuth CLI token before the user turn', async () => { + const { hahaOAuthService } = await import('../services/hahaOAuthService.js') + await hahaOAuthService.saveTokens({ + accessToken: 'fresh-after-wake-token', + refreshToken: 'refresh-xxx', + expiresAt: Date.now() + 30 * 60_000, + scopes: ['user:inference'], + subscriptionType: 'max', + }) + + const service = new ConversationService() as any + const sent: string[] = [] + service.sessions.set('sleep-wake-session', { + proc: {}, + outputCallbacks: [], + workDir: tmpDir, + permissionMode: 'default', + sdkToken: 'sdk-token', + sdkSocket: { + send(line: string) { + sent.push(line) + }, + }, + pendingOutbound: [], + startupPending: false, + startupExitCode: null, + stdoutLines: [], + stderrLines: [], + outputDrain: Promise.resolve(), + sdkMessages: [], + initMessage: null, + pendingPermissionRequests: new Map(), + usesOfficialOAuth: true, + officialOAuthToken: 'stale-before-sleep-token', + }) + + const ok = await service.sendMessage('sleep-wake-session', 'hello after wake') + + expect(ok).toBe(true) + expect(sent).toHaveLength(2) + expect(JSON.parse(sent[0]!).type).toBe('update_environment_variables') + expect(JSON.parse(sent[0]!).variables.CLAUDE_CODE_OAUTH_TOKEN).toBe('fresh-after-wake-token') + expect(JSON.parse(sent[1]!).type).toBe('user') + }) + test('buildChildEnv does NOT inject CLAUDE_CODE_OAUTH_TOKEN when not official mode', async () => { const ccHahaDir = path.join(tmpDir, 'cc-haha') await fs.mkdir(ccHahaDir, { recursive: true }) diff --git a/src/server/services/conversationService.ts b/src/server/services/conversationService.ts index d80bf545..b12d7a9c 100644 --- a/src/server/services/conversationService.ts +++ b/src/server/services/conversationService.ts @@ -77,6 +77,8 @@ type SessionProcess = { outputDrain: Promise sdkMessages: any[] initMessage: any | null + usesOfficialOAuth: boolean + officialOAuthToken: string | null pendingPermissionRequests: Map< string, { @@ -261,6 +263,7 @@ export class ConversationService { // chdir 后落到正确目录。 // const childEnv = await this.buildChildEnv(launchWorkDir, sdkUrl, options) + const usesOfficialOAuth = this.shouldMarkManagedOAuth(options?.providerId) let proc: ReturnType try { @@ -308,6 +311,8 @@ export class ConversationService { outputDrain: Promise.resolve(), sdkMessages: [], initMessage: null, + usesOfficialOAuth, + officialOAuthToken: childEnv.CLAUDE_CODE_OAUTH_TOKEN ?? null, pendingPermissionRequests: new Map(), } this.sessions.set(sessionId, session) @@ -412,6 +417,10 @@ export class ConversationService { content: string, attachments?: AttachmentRef[], ): Promise { + const session = this.sessions.get(sessionId) + if (session) { + await this.refreshOfficialOAuthTokenBeforeTurn(sessionId, session) + } const userContent = await this.buildUserContent(content, sessionId, attachments) return this.sendSdkMessage(sessionId, { type: 'user', @@ -1139,6 +1148,33 @@ export class ConversationService { return env } + private async refreshOfficialOAuthTokenBeforeTurn( + sessionId: string, + session: SessionProcess, + ): Promise { + if (!session.usesOfficialOAuth) return + + let token: string | null = null + try { + const { hahaOAuthService } = await import('./hahaOAuthService.js') + token = await hahaOAuthService.ensureFreshAccessToken() + } catch (err) { + console.error( + '[conversationService] refresh official OAuth token before turn failed:', + err instanceof Error ? err.message : err, + ) + return + } + + if (!token || token === session.officialOAuthToken) return + + session.officialOAuthToken = token + this.sendSdkMessage(sessionId, { + type: 'update_environment_variables', + variables: { CLAUDE_CODE_OAUTH_TOKEN: token }, + }) + } + private shouldStripInheritedProviderEnv(providerId?: string | null): boolean { if (providerId !== undefined) { return true diff --git a/src/services/api/withRetry.test.ts b/src/services/api/withRetry.test.ts new file mode 100644 index 00000000..a4b12132 --- /dev/null +++ b/src/services/api/withRetry.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, test } from 'bun:test' +import type Anthropic from '@anthropic-ai/sdk' +import { APIConnectionError } from '@anthropic-ai/sdk' +import { _resetKeepAliveForTesting, getProxyFetchOptions } from '../../utils/proxy.js' +import { withRetry } from './withRetry.js' + +describe('withRetry stale connections', () => { + test('disables keep-alive before retrying ECONNRESET connection failures', async () => { + _resetKeepAliveForTesting() + let attempts = 0 + const cause = Object.assign(new Error('socket hang up'), { + code: 'ECONNRESET', + }) + const staleConnection = new APIConnectionError({ + message: 'Connection error.', + cause, + }) + + const generator = withRetry( + async () => ({} as Anthropic), + async () => { + attempts += 1 + if (attempts === 1) { + throw staleConnection + } + return 'ok' + }, + { + model: 'claude-opus-4-7', + thinkingConfig: { type: 'disabled' }, + maxRetries: 1, + }, + ) + + let finalValue: string | undefined + for (;;) { + const next = await generator.next() + if (next.done) { + finalValue = next.value + break + } + } + + expect(finalValue).toBe('ok') + expect(attempts).toBe(2) + expect(getProxyFetchOptions().keepalive).toBe(false) + _resetKeepAliveForTesting() + }) +}) diff --git a/src/services/api/withRetry.ts b/src/services/api/withRetry.ts index 5ec9ad08..f63b6f77 100644 --- a/src/services/api/withRetry.ts +++ b/src/services/api/withRetry.ts @@ -35,7 +35,6 @@ import { isNonCustomOpusModel } from '../../utils/model/model.js' import { disableKeepAlive } from '../../utils/proxy.js' import { sleep } from '../../utils/sleep.js' import type { ThinkingConfig } from '../../utils/thinking.js' -import { getFeatureValue_CACHED_MAY_BE_STALE } from '../analytics/growthbook.js' import { type AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, logEvent, @@ -216,13 +215,7 @@ export async function* withRetry( // - Vertex-specific auth errors (credential refresh failures, 401) // - ECONNRESET/EPIPE: stale keep-alive socket; disable pooling and reconnect const isStaleConnection = isStaleConnectionError(lastError) - if ( - isStaleConnection && - getFeatureValue_CACHED_MAY_BE_STALE( - 'tengu_disable_keepalive_on_econnreset', - false, - ) - ) { + if (isStaleConnection) { logForDebugging( 'Stale connection (ECONNRESET/EPIPE) — disabling keep-alive for retry', )