fix: project id fallback and stream error regression (#94)

* fix(proxy): harden project id handling for upgraded accounts

* fix(proxy): parse stream error payloads to preserve upstream project errors

src/server/modules/proxy/clients/gemini.client.ts

@@ -198,7 +198,7 @@ export class GeminiClient {
         const hasNextEndpoint = index < baseUrls.length - 1;
 
         if (!hasNextEndpoint || !this.shouldSwitchEndpointOnError(error)) {
-          this.handleAxiosError(error, operation);
+          await this.handleAxiosError(error, operation);
         }
 
         this.logger.warn(
@@ -209,33 +209,141 @@ export class GeminiClient {
       }
     }
 
-    this.handleAxiosError(lastError, operation);
+    await this.handleAxiosError(lastError, operation);
+    throw new Error(`[${operation}] unexpected control flow after upstream error handling`);
   }
 
-  private handleAxiosError(error: unknown, operation: string): never {
+  private async handleAxiosError(error: unknown, operation: string): Promise<never> {
     if (axios.isAxiosError(error)) {
       const responseData = error.response?.data;
-      const upstreamMessage = this.extractAxiosErrorMessage(responseData);
+      const upstreamMessage = await this.extractAxiosErrorMessage(responseData);
       this.logger.error(
-        `[${operation}] upstream request error: ${error.message} - ${this.safeStringify(responseData)}`,
+        `[${operation}] upstream request error: ${error.message} - ${this.describeAxiosErrorData(
+          responseData,
+        )}`,
       );
       throw new Error(upstreamMessage || error.message);
     }
     this.throwAsCleanError(error);
   }
 
-  private extractAxiosErrorMessage(responseData: unknown): string | null {
-    if (!responseData || typeof responseData !== 'object') {
+  private async extractAxiosErrorMessage(responseData: unknown): Promise<string | null> {
+    const fromObject = this.extractAxiosErrorMessageFromObject(responseData);
+    if (fromObject) {
+      return fromObject;
+    }
+
+    if (typeof responseData === 'string') {
+      return this.extractAxiosErrorMessageFromText(responseData);
+    }
+
+    if (Buffer.isBuffer(responseData)) {
+      return this.extractAxiosErrorMessageFromText(responseData.toString('utf-8'));
+    }
+
+    if (this.isReadableStream(responseData)) {
+      const streamText = await this.readStreamAsText(responseData);
+      return streamText ? this.extractAxiosErrorMessageFromText(streamText) : null;
+    }
+
+    return null;
+  }
+
+  private extractAxiosErrorMessageFromObject(responseData: unknown): string | null {
+    if (!responseData || typeof responseData !== 'object' || this.isReadableStream(responseData)) {
       return null;
     }
 
     const errorRecord = (responseData as { error?: unknown }).error;
-    if (!errorRecord || typeof errorRecord !== 'object') {
+    if (errorRecord && typeof errorRecord === 'object') {
+      const message = (errorRecord as { message?: unknown }).message;
+      if (typeof message === 'string' && message.trim() !== '') {
+        return message.trim();
+      }
+    }
+
+    const message = (responseData as { message?: unknown }).message;
+    if (typeof message === 'string' && message.trim() !== '') {
+      return message.trim();
+    }
+
+    return null;
+  }
+
+  private extractAxiosErrorMessageFromText(rawText: string): string | null {
+    const text = rawText.trim();
+    if (!text) {
       return null;
     }
 
-    const message = (errorRecord as { message?: unknown }).message;
-    return typeof message === 'string' && message.trim() !== '' ? message : null;
+    const sseLines = text
+      .split('\n')
+      .map((line) => line.trim())
+      .filter((line) => line.startsWith('data:'));
+
+    for (const line of sseLines) {
+      const payload = line.slice(5).trim();
+      if (!payload || payload === '[DONE]') {
+        continue;
+      }
+      const parsed = this.tryParseJson(payload);
+      const message = this.extractAxiosErrorMessageFromObject(parsed);
+      if (message) {
+        return message;
+      }
+    }
+
+    const parsed = this.tryParseJson(text);
+    const fromJson = this.extractAxiosErrorMessageFromObject(parsed);
+    if (fromJson) {
+      return fromJson;
+    }
+
+    return null;
+  }
+
+  private tryParseJson(value: string): unknown {
+    try {
+      return JSON.parse(value);
+    } catch {
+      return null;
+    }
+  }
+
+  private isReadableStream(value: unknown): value is NodeJS.ReadableStream {
+    return (
+      Boolean(value) &&
+      typeof value === 'object' &&
+      typeof (value as { pipe?: unknown }).pipe === 'function'
+    );
+  }
+
+  private async readStreamAsText(stream: NodeJS.ReadableStream): Promise<string | null> {
+    return new Promise((resolve) => {
+      let buffer = '';
+      const maxChars = 512 * 1024;
+
+      stream.on('data', (chunk: Buffer | string) => {
+        const text = Buffer.isBuffer(chunk) ? chunk.toString('utf-8') : String(chunk);
+        if (buffer.length >= maxChars) {
+          return;
+        }
+        buffer += text;
+        if (buffer.length > maxChars) {
+          buffer = buffer.slice(0, maxChars);
+        }
+      });
+
+      stream.on('end', () => resolve(buffer));
+      stream.on('error', () => resolve(null));
+    });
+  }
+
+  private describeAxiosErrorData(responseData: unknown): string {
+    if (this.isReadableStream(responseData)) {
+      return '[stream]';
+    }
+    return this.safeStringify(responseData);
   }
 
   private resolveAxiosProxy(upstreamProxyUrl?: string): AxiosProxyConfig | false | undefined {

src/server/modules/proxy/token-manager.service.ts

@@ -26,12 +26,19 @@ function normalizeProjectId(projectId: string | null | undefined): string | unde
     return undefined;
   }
 
+  // Legacy upgrades may carry malformed resource names (e.g. "projects/").
+  // The internal gateway expects a plain project id, not a resource path.
+  if (/^projects(?:\/.*)?$/i.test(trimmedProjectId)) {
+    return undefined;
+  }
+
   return trimmedProjectId;
 }
 
 @Injectable()
 export class TokenManagerService implements OnModuleInit {
   private readonly logger = new Logger(TokenManagerService.name);
+  private readonly defaultFallbackProjectId = 'silver-orbit-5m7qc';
   private currentIndex = 0;
   private readonly stickySessionTtlMs = 10 * 60 * 1000;
   private readonly rateLimitCooldownMs = 5 * 60 * 1000;
@@ -178,6 +185,8 @@ export class TokenManagerService implements OnModuleInit {
     sessionKey?: string,
   ): Promise<CloudAccount | null> {
     try {
+      let effectiveProjectId: string | undefined;
+
       // Check if token needs refresh (expires in < 5 minutes)
       if (nowSeconds >= tokenData.expiry_timestamp - 300) {
         this.logger.log(`Token for ${tokenData.email} is close to expiry; refreshing...`);
@@ -199,19 +208,19 @@ export class TokenManagerService implements OnModuleInit {
         }
       }
 
-      if (
-        normalizeProjectId(tokenData.project_id) === undefined
-      ) {
+      if (normalizeProjectId(tokenData.project_id) === undefined) {
         tokenData.project_id = undefined;
       }
+      effectiveProjectId = tokenData.project_id;
 
-      if (!tokenData.project_id) {
+      if (!effectiveProjectId) {
         try {
           const fetchedProjectId = await GoogleAPIService.fetchProjectId(tokenData.access_token);
           const normalizedProjectId = normalizeProjectId(fetchedProjectId);
 
           if (normalizedProjectId) {
             tokenData.project_id = normalizedProjectId;
+            effectiveProjectId = normalizedProjectId;
             await this.saveTokenState(accountId, tokenData);
             this.tokens.set(accountId, tokenData);
             this.logger.log(`Resolved project ID for ${tokenData.email}: ${normalizedProjectId}`);
@@ -225,6 +234,14 @@ export class TokenManagerService implements OnModuleInit {
         }
       }
 
+      if (!effectiveProjectId) {
+        const fallbackProjectId = this.getFallbackProjectId();
+        effectiveProjectId = fallbackProjectId;
+        this.logger.warn(
+          `Using fallback project ID for ${tokenData.email} without persistence: ${fallbackProjectId}`,
+        );
+      }
+
       this.logger.log(`Using account: ${tokenData.email}`);
       if (sessionKey) {
         this.sessionBindings.set(sessionKey, {
@@ -245,7 +262,7 @@ export class TokenManagerService implements OnModuleInit {
           token_type: tokenData.token_type,
           expires_in: tokenData.expires_in,
           expiry_timestamp: tokenData.expiry_timestamp,
-          project_id: tokenData.project_id,
+          project_id: effectiveProjectId,
           session_id: tokenData.session_id,
           upstream_proxy_url: tokenData.upstream_proxy_url,
         },
@@ -329,4 +346,13 @@ export class TokenManagerService implements OnModuleInit {
   getAccountCount(): number {
     return this.tokens.size;
   }
+
+  private getFallbackProjectId(): string {
+    const fromEnv = process.env.PROXY_FALLBACK_PROJECT_ID?.trim();
+    const normalizedFromEnv = normalizeProjectId(fromEnv);
+    if (normalizedFromEnv) {
+      return normalizedFromEnv;
+    }
+    return this.defaultFallbackProjectId;
+  }
 }

src/tests/scripts/project-id-request-validation.ts

@@ -3,8 +3,6 @@ import assert from 'node:assert/strict';
 import { transformClaudeRequestIn } from '../../lib/antigravity/ClaudeRequestMapper';
 import { ProxyService } from '../../server/modules/proxy/proxy.service';
 import { TokenManagerService } from '../../server/modules/proxy/token-manager.service';
-import { CloudAccountRepo } from '../../ipc/database/cloudHandler';
-import { GoogleAPIService } from '../../services/GoogleAPIService';
 
 const mockTokenManager: any = {
   getNextToken: async () => null,
@@ -199,43 +197,12 @@ async function validateRuntimeGeminiRequestPath(): Promise<void> {
   assert.ok(!Object.prototype.hasOwnProperty.call(capturedBody, 'project'));
 }
 
-async function validateRuntimeGeminiRequestFromAccountProject(): Promise<void> {
-  const sourceProjectId = `account-project-${Date.now()}`;
-  const account = {
-    id: 'account-source-1',
-    provider: 'google',
-    email: '[email protected]',
-    token: {
-      access_token: 'source-access-token',
-      refresh_token: 'source-refresh-token',
-      token_type: 'Bearer',
-      expires_in: 3600,
-      expiry_timestamp: Math.floor(Date.now() / 1000) + 3600,
-      project_id: sourceProjectId,
-    },
-    created_at: 1,
-    last_used: 1,
-  };
-
-  const cloudAccountRepo = CloudAccountRepo as any;
-  const googleApiService = GoogleAPIService as any;
-
-  const originalGetAccounts = cloudAccountRepo.getAccounts;
-  const originalGetAccount = cloudAccountRepo.getAccount;
-  const originalUpdateToken = cloudAccountRepo.updateToken;
-  const originalFetchProjectId = googleApiService.fetchProjectId;
+async function validateRuntimeAnthropicRequestFromRealTokenManager(): Promise<void> {
   const originalGenerateInternal = mockGeminiClient.generateInternal;
 
   let capturedBody: Record<string, unknown> | null = null;
-  let fetchProjectIdCalled = false;
-
-  cloudAccountRepo.getAccounts = async () => [account];
-  cloudAccountRepo.getAccount = async (id: string) => (id === account.id ? account : null);
-  cloudAccountRepo.updateToken = async () => undefined;
-  googleApiService.fetchProjectId = async () => {
-    fetchProjectIdCalled = true;
-    return 'unexpected-project-id';
-  };
+  let selectedToken: any = null;
+  let observedGetNextTokenOptions: any = null;
 
   mockGeminiClient.generateInternal = async (body: Record<string, unknown>) => {
     capturedBody = body;
@@ -254,36 +221,63 @@ async function validateRuntimeGeminiRequestFromAccountProject(): Promise<void> {
     };
   };
 
+  const realTokenManager = new TokenManagerService();
+  await realTokenManager.onModuleInit();
+
+  const tokenManagerProxy = {
+    getNextToken: async (options?: { sessionKey?: string; excludeAccountIds?: string[] }) => {
+      observedGetNextTokenOptions = options ?? null;
+      selectedToken = await realTokenManager.getNextToken(options);
+      return selectedToken;
+    },
+    markAsRateLimited: (accountIdOrEmail: string) =>
+      realTokenManager.markAsRateLimited(accountIdOrEmail),
+    markAsForbidden: (accountIdOrEmail: string) => realTokenManager.markAsForbidden(accountIdOrEmail),
+  };
+
   try {
-    const tokenManager = new TokenManagerService();
-    const service = new ProxyService(tokenManager as any, mockGeminiClient as any);
-    await service.handleGeminiGenerateContent('models/gemini-2.5-flash', {
-      contents: [{ role: 'user', parts: [{ text: 'hello from account source' }] }],
+    const service = new ProxyService(tokenManagerProxy as any, mockGeminiClient as any);
+    await service.handleAnthropicMessages({
+      model: 'claude-sonnet-4-5',
+      stream: false,
+      max_tokens: 128,
+      metadata: {
+        session_id: 'project-chain-debug-session',
+      },
+      messages: [{ role: 'user', content: 'hello from real token manager chain' }],
     } as any);
   } finally {
-    cloudAccountRepo.getAccounts = originalGetAccounts;
-    cloudAccountRepo.getAccount = originalGetAccount;
-    cloudAccountRepo.updateToken = originalUpdateToken;
-    googleApiService.fetchProjectId = originalFetchProjectId;
     mockGeminiClient.generateInternal = originalGenerateInternal;
   }
 
+  assert.ok(selectedToken, 'Expected TokenManagerService.getNextToken to return a token');
   assert.ok(capturedBody, 'Expected captured Gemini internal request');
-  const captured = capturedBody as Record<string, unknown>;
-  assert.equal(
-    captured['project'],
-    sourceProjectId,
-    'Expected project field to come from account.token.project_id',
-  );
-  assert.equal(
-    fetchProjectIdCalled,
-    false,
-    'fetchProjectId should not be called when account already has project_id',
-  );
 
-  console.log('[DEBUG] Account source project_id:', sourceProjectId);
+  const captured = capturedBody as Record<string, unknown>;
+  const tokenProjectId = selectedToken?.token?.project_id;
+  const normalizedTokenProjectId =
+    typeof tokenProjectId === 'string' && tokenProjectId.trim() !== '' ? tokenProjectId.trim() : undefined;
+
+  if (normalizedTokenProjectId) {
+    assert.equal(
+      captured['project'],
+      normalizedTokenProjectId,
+      'Expected project field to come from tokenManager.getNextToken().token.project_id',
+    );
+  } else {
+    assert.ok(
+      !Object.prototype.hasOwnProperty.call(captured, 'project'),
+      'Expected project field to be omitted when token project_id is unavailable',
+    );
+  }
+
   console.log(
-    '[DEBUG] Runtime captured internal request (project from account):',
+    '[DEBUG] Observed getNextToken options from ProxyService:',
+    JSON.stringify(observedGetNextTokenOptions, null, 2),
+  );
+  console.log('[DEBUG] Selected token project_id from TokenManagerService:', tokenProjectId ?? null);
+  console.log(
+    '[DEBUG] Runtime captured internal request (from handleAnthropicMessages):',
     JSON.stringify(captured, null, 2),
   );
 }
@@ -340,8 +334,8 @@ async function main(): Promise<void> {
   await validateRuntimeGeminiRequestPath();
   console.log('[PASS] Runtime Gemini request path omits empty project');
 
-  await validateRuntimeGeminiRequestFromAccountProject();
-  console.log('[PASS] Runtime Gemini request uses project_id from account source');
+  await validateRuntimeAnthropicRequestFromRealTokenManager();
+  console.log('[PASS] Runtime Anthropic request uses project_id from TokenManagerService');
 
   if (process.argv.includes('--live')) {
     await validateLiveRequests();

src/tests/unit/cloud-monitor.test.ts

@@ -261,7 +261,7 @@ describe('TokenManagerService project-id hydration', () => {
     expect(selectedToken?.token.project_id).toBe('existing-project');
   });
 
-  it('keeps request usable when project_id cannot be resolved', async () => {
+  it('uses fallback project_id when project_id cannot be resolved', async () => {
     const account = {
       id: 'acc-3',
       provider: 'google',
@@ -285,6 +285,42 @@ describe('TokenManagerService project-id hydration', () => {
     const selectedToken = await service.getNextToken();
 
     expect(GoogleAPIService.fetchProjectId).toHaveBeenCalledWith('access-token-3');
-    expect(selectedToken?.token.project_id).toBeUndefined();
+    expect(CloudAccountRepo.updateToken).not.toHaveBeenCalled();
+    expect(selectedToken?.token.project_id).toBe('silver-orbit-5m7qc');
+  });
+
+  it('rehydrates malformed legacy resource-style project_id values', async () => {
+    const account = {
+      id: 'acc-4',
+      provider: 'google',
+      email: '[email protected]',
+      token: {
+        access_token: 'access-token-4',
+        refresh_token: 'refresh-token-4',
+        token_type: 'Bearer',
+        expires_in: 3600,
+        expiry_timestamp: Math.floor(Date.now() / 1000) + 3600,
+        project_id: 'projects/',
+      },
+      created_at: 1,
+      last_used: 1,
+    };
+
+    vi.mocked(CloudAccountRepo.getAccounts).mockResolvedValue([account] as never);
+    vi.mocked(CloudAccountRepo.getAccount).mockResolvedValue(account as never);
+    vi.mocked(CloudAccountRepo.updateToken).mockResolvedValue(undefined as never);
+    vi.mocked(GoogleAPIService.fetchProjectId).mockResolvedValue('resolved-project-4' as never);
+
+    const service = new TokenManagerService();
+    const selectedToken = await service.getNextToken();
+
+    expect(GoogleAPIService.fetchProjectId).toHaveBeenCalledWith('access-token-4');
+    expect(CloudAccountRepo.updateToken).toHaveBeenCalledWith(
+      'acc-4',
+      expect.objectContaining({
+        project_id: 'resolved-project-4',
+      }),
+    );
+    expect(selectedToken?.token.project_id).toBe('resolved-project-4');
   });
 });