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

@@ -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')
+  })
+})

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(', ')}`,
         )

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 })

src/server/services/conversationService.ts

@@ -77,6 +77,8 @@ type SessionProcess = {
   outputDrain: Promise<void>
   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<typeof Bun.spawn>
     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<boolean> {
+    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<void> {
+    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

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()
+  })
+})

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<T>(
       // - 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',
         )