feat(proxy): complete openai routing scope

tests/integration/proxy/messages-edge-cases.test.ts

@@ -0,0 +1,259 @@
+import { afterEach, beforeEach, describe, expect, it } from 'bun:test';
+import getPort from 'get-port';
+import * as fs from 'fs';
+import * as http from 'http';
+import * as os from 'os';
+import * as path from 'path';
+import { startOpenAICompatProxyServer } from '../../../src/proxy/server/proxy-server';
+import type { OpenAICompatProfileConfig } from '../../../src/proxy/profile-router';
+
+let proxyServer: http.Server;
+let upstreamServer: http.Server;
+let upstreamSockets = new Set<import('net').Socket>();
+let upstreamPort: number;
+let proxyPort: number;
+let tempDir: string;
+let originalTimeoutEnv: string | undefined;
+let originalCcsHome: string | undefined;
+
+async function startUpstream(
+  handler: (req: http.IncomingMessage, res: http.ServerResponse) => Promise<void> | void
+): Promise<void> {
+  upstreamServer = http.createServer((req, res) => {
+    void Promise.resolve(handler(req, res));
+  });
+  upstreamServer.on('connection', (socket) => {
+    upstreamSockets.add(socket);
+    socket.on('close', () => {
+      upstreamSockets.delete(socket);
+    });
+  });
+  await new Promise<void>((resolve) =>
+    upstreamServer.listen(upstreamPort, '127.0.0.1', () => resolve())
+  );
+}
+
+async function requestProxy(payload: unknown, signal?: AbortSignal): Promise<Response> {
+  return fetch(`http://127.0.0.1:${proxyPort}/v1/messages`, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      'x-api-key': 'test-proxy-token',
+    },
+    body: JSON.stringify(payload),
+    signal,
+  });
+}
+
+beforeEach(async () => {
+  upstreamPort = await getPort();
+  proxyPort = await getPort();
+  tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccs-proxy-edge-'));
+  originalTimeoutEnv = process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS;
+  originalCcsHome = process.env.CCS_HOME;
+  process.env.CCS_HOME = tempDir;
+});
+
+afterEach(async () => {
+  if (originalTimeoutEnv !== undefined) {
+    process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS = originalTimeoutEnv;
+  } else {
+    delete process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS;
+  }
+  if (originalCcsHome !== undefined) {
+    process.env.CCS_HOME = originalCcsHome;
+  } else {
+    delete process.env.CCS_HOME;
+  }
+
+  if (proxyServer) {
+    await new Promise<void>((resolve) => proxyServer.close(() => resolve()));
+  }
+  if (upstreamServer) {
+    for (const socket of upstreamSockets) {
+      socket.destroy();
+    }
+    upstreamSockets = new Set();
+    await new Promise<void>((resolve) => upstreamServer.close(() => resolve()));
+  }
+  fs.rmSync(tempDir, { recursive: true, force: true });
+});
+
+describe('openai proxy message edge cases', () => {
+  async function startProxyWithHandler(
+    handler: (req: http.IncomingMessage, res: http.ServerResponse) => Promise<void> | void
+  ) {
+    await startUpstream(handler);
+    const profile: OpenAICompatProfileConfig = {
+      profileName: 'hf',
+      settingsPath: '/tmp/hf.settings.json',
+      baseUrl: `http://127.0.0.1:${upstreamPort}`,
+      apiKey: 'hf_token',
+      provider: 'generic-chat-completion-api',
+      model: 'hf-model',
+    };
+    proxyServer = startOpenAICompatProxyServer({
+      profile,
+      port: proxyPort,
+      authToken: 'test-proxy-token',
+    });
+  }
+
+  it('preserves rate-limit errors from the upstream provider', async () => {
+    await startProxyWithHandler((_req, res) => {
+      res.writeHead(429, {
+        'Content-Type': 'application/json',
+        'Retry-After': '9',
+      });
+      res.end(JSON.stringify({ error: { message: 'rate limited' } }));
+    });
+
+    const response = await requestProxy({
+      model: 'hf-model',
+      messages: [{ role: 'user', content: 'hello' }],
+    });
+
+    expect(response.status).toBe(429);
+    expect(response.headers.get('retry-after')).toBe('9');
+    await expect(response.json()).resolves.toMatchObject({
+      type: 'error',
+      error: {
+        type: 'rate_limit_error',
+        message: 'rate limited',
+      },
+    });
+  });
+
+  it('returns api_error when the upstream JSON response has no usable choices', async () => {
+    await startProxyWithHandler((_req, res) => {
+      res.writeHead(200, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ id: 'chatcmpl_empty', choices: [] }));
+    });
+
+    const response = await requestProxy({
+      model: 'hf-model',
+      messages: [{ role: 'user', content: 'hello' }],
+    });
+
+    expect(response.status).toBe(502);
+    await expect(response.json()).resolves.toMatchObject({
+      type: 'error',
+      error: {
+        type: 'api_error',
+        message: 'Failed to translate Cursor JSON response',
+      },
+    });
+  });
+
+  it('streams thinking deltas and chunked tool-call arguments back as Anthropic SSE', async () => {
+    await startProxyWithHandler((_req, res) => {
+      res.writeHead(200, { 'Content-Type': 'text/event-stream' });
+      res.write(
+        'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{"role":"assistant"}}]}\n\n'
+      );
+      res.write(
+        'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{"reasoning_content":"Need to search first."}}]}\n\n'
+      );
+      res.write(
+        'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"search","arguments":"{\\"q\\":\\""}}]}}]}\n\n'
+      );
+      res.write(
+        'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"docs\\"}"}}]}}]}\n\n'
+      );
+      res.write(
+        'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":8,"completion_tokens":6}}\n\n'
+      );
+      res.end('data: [DONE]\n\n');
+    });
+
+    const response = await requestProxy({
+      model: 'hf-model',
+      stream: true,
+      messages: [{ role: 'user', content: 'search docs' }],
+    });
+
+    const body = await response.text();
+    expect(response.status).toBe(200);
+    expect(body).toContain('"type":"thinking_delta"');
+    expect(body).toContain('"type":"tool_use"');
+    expect(body).toContain('"partial_json":"{\\"q\\":\\""');
+    expect(body).toContain('"partial_json":"docs\\"}"');
+    expect(body).toContain('event: message_stop');
+  });
+
+  it('returns a timeout error when the upstream does not respond in time', async () => {
+    process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS = '50';
+    await startProxyWithHandler(async (req, _res) => {
+      await new Promise<void>((resolve) => req.on('close', () => resolve()));
+    });
+
+    const response = await requestProxy({
+      model: 'hf-model',
+      messages: [{ role: 'user', content: 'hello' }],
+    });
+
+    expect(response.status).toBe(502);
+    await expect(response.json()).resolves.toMatchObject({
+      type: 'error',
+      error: {
+        type: 'api_error',
+        message: 'The upstream provider did not respond within 50ms',
+      },
+    });
+  });
+
+  it('aborts the upstream request when the client disconnects mid-flight', async () => {
+    await startProxyWithHandler(() => {});
+
+    await new Promise<void>((resolve) => {
+      const request = http.request(
+        {
+          hostname: '127.0.0.1',
+          port: proxyPort,
+          path: '/v1/messages',
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+            'x-api-key': 'test-proxy-token',
+          },
+        },
+        () => resolve()
+      );
+
+      request.write(
+        JSON.stringify({
+          model: 'hf-model',
+          messages: [{ role: 'user', content: 'hello' }],
+        })
+      );
+      request.end();
+
+      setTimeout(() => {
+        request.destroy(new Error('client aborted'));
+        resolve();
+      }, 50);
+    });
+
+    const logPath = path.join(tempDir, '.ccs', 'logs', 'current.jsonl');
+    await Promise.race([
+      new Promise<void>((resolve, reject) => {
+        const startedAt = Date.now();
+        const timer = setInterval(() => {
+          if (fs.existsSync(logPath)) {
+            const content = fs.readFileSync(logPath, 'utf8');
+            if (content.includes('"event":"request.disconnect"')) {
+              clearInterval(timer);
+              resolve();
+              return;
+            }
+          }
+
+          if (Date.now() - startedAt > 1500) {
+            clearInterval(timer);
+            reject(new Error('proxy did not log disconnect cleanup'));
+          }
+        }, 50);
+      }),
+    ]);
+  });
+});

tests/integration/proxy/messages-endpoint.test.ts

@@ -173,16 +173,44 @@ describe('openai proxy messages endpoint', () => {
     expect(body.error?.type).toBe('authentication_error');
   });
 
-  it('returns invalid_request_error for unsupported Anthropic content blocks', async () => {
+  it('translates supported Anthropic image blocks into OpenAI image_url content', async () => {
     const response = await requestProxy({
       model: 'hf-model',
-      messages: [{ role: 'user', content: [{ type: 'image' }] }],
+      messages: [
+        {
+          role: 'user',
+          content: [
+            { type: 'text', text: 'Describe this' },
+            {
+              type: 'image',
+              source: {
+                type: 'base64',
+                media_type: 'image/png',
+                data: 'ZmFrZQ==',
+              },
+            },
+          ],
+        },
+      ],
     });
-    const body = (await response.json()) as { error?: { type?: string; message?: string } };
 
-    expect(response.status).toBe(400);
-    expect(body.error?.type).toBe('invalid_request_error');
-    expect(body.error?.message).toContain('is not supported');
+    expect(response.status).toBe(200);
+    const parsedUpstream = upstreamBody as {
+      messages?: Array<{
+        role: string;
+        content: Array<{ type: string; text?: string; image_url?: { url?: string } }>;
+      }>;
+    };
+    expect(parsedUpstream.messages?.[0]).toEqual({
+      role: 'user',
+      content: [
+        { type: 'text', text: 'Describe this' },
+        {
+          type: 'image_url',
+          image_url: { url: 'data:image/png;base64,ZmFrZQ==' },
+        },
+      ],
+    });
   });
 
   it('accepts query strings and trailing slashes on the messages route', async () => {

tests/integration/proxy/request-routing.test.ts

@@ -0,0 +1,217 @@
+import { afterEach, beforeEach, describe, expect, it } from 'bun:test';
+import getPort from 'get-port';
+import * as fs from 'fs';
+import * as http from 'http';
+import * as os from 'os';
+import * as path from 'path';
+import { startOpenAICompatProxyServer } from '../../../src/proxy/server/proxy-server';
+import type { OpenAICompatProfileConfig } from '../../../src/proxy/profile-router';
+
+let originalCcsHome: string | undefined;
+let tempDir: string;
+let proxyServer: http.Server;
+let upstreamServers: http.Server[] = [];
+let proxyPort: number;
+
+function startMockUpstream(
+  port: number,
+  hitLabel: string,
+  hits: string[],
+  bodies: Array<{ label: string; body: unknown }>
+): Promise<void> {
+  return new Promise((resolve) => {
+    const server = http.createServer(async (req, res) => {
+      let body = '';
+      for await (const chunk of req) {
+        body += chunk.toString();
+      }
+      hits.push(hitLabel);
+      bodies.push({ label: hitLabel, body: JSON.parse(body) });
+
+      res.writeHead(200, { 'Content-Type': 'application/json' });
+      res.end(
+        JSON.stringify({
+          id: `chatcmpl_${hitLabel}`,
+          model: hitLabel,
+          choices: [
+            {
+              index: 0,
+              message: { role: 'assistant', content: `Reply from ${hitLabel}` },
+              finish_reason: 'stop',
+            },
+          ],
+          usage: { prompt_tokens: 2, completion_tokens: 3 },
+        })
+      );
+    });
+    upstreamServers.push(server);
+    server.listen(port, '127.0.0.1', () => resolve());
+  });
+}
+
+function writeSettings(profileName: string, env: Record<string, string>): string {
+  const settingsPath = path.join(tempDir, '.ccs', `${profileName}.settings.json`);
+  fs.writeFileSync(settingsPath, JSON.stringify({ env }, null, 2), 'utf8');
+  return settingsPath;
+}
+
+async function requestProxy(payload: unknown): Promise<Response> {
+  return fetch(`http://127.0.0.1:${proxyPort}/v1/messages`, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      'x-api-key': 'test-proxy-token',
+    },
+    body: JSON.stringify(payload),
+  });
+}
+
+beforeEach(async () => {
+  originalCcsHome = process.env.CCS_HOME;
+  tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccs-proxy-routing-'));
+  fs.mkdirSync(path.join(tempDir, '.ccs'), { recursive: true });
+  process.env.CCS_HOME = tempDir;
+  proxyPort = await getPort();
+});
+
+afterEach(async () => {
+  await Promise.all(
+    upstreamServers.map(
+      (server) =>
+        new Promise<void>((resolve) => {
+          server.close(() => resolve());
+        })
+    )
+  );
+  upstreamServers = [];
+  if (proxyServer) {
+    await new Promise<void>((resolve) => proxyServer.close(() => resolve()));
+  }
+  if (originalCcsHome !== undefined) {
+    process.env.CCS_HOME = originalCcsHome;
+  } else {
+    delete process.env.CCS_HOME;
+  }
+  fs.rmSync(tempDir, { recursive: true, force: true });
+});
+
+describe('openai proxy request routing', () => {
+  it('routes explicit profile:model selectors to the matching upstream profile', async () => {
+    const primaryPort = await getPort();
+    const secondaryPort = await getPort();
+    const hits: string[] = [];
+    const bodies: Array<{ label: string; body: unknown }> = [];
+    await startMockUpstream(primaryPort, 'primary', hits, bodies);
+    await startMockUpstream(secondaryPort, 'secondary', hits, bodies);
+
+    const primarySettings = writeSettings('hf', {
+      ANTHROPIC_BASE_URL: `http://127.0.0.1:${primaryPort}`,
+      ANTHROPIC_AUTH_TOKEN: 'hf_token',
+      ANTHROPIC_MODEL: 'hf-default',
+      CCS_DROID_PROVIDER: 'generic-chat-completion-api',
+    });
+    const secondarySettings = writeSettings('deepseek', {
+      ANTHROPIC_BASE_URL: `http://127.0.0.1:${secondaryPort}`,
+      ANTHROPIC_AUTH_TOKEN: 'deepseek_token',
+      ANTHROPIC_MODEL: 'deepseek-chat',
+      CCS_DROID_PROVIDER: 'generic-chat-completion-api',
+    });
+
+    fs.writeFileSync(
+      path.join(tempDir, '.ccs', 'config.json'),
+      JSON.stringify({ profiles: { hf: primarySettings, deepseek: secondarySettings } }, null, 2),
+      'utf8'
+    );
+
+    const profile: OpenAICompatProfileConfig = {
+      profileName: 'hf',
+      settingsPath: primarySettings,
+      baseUrl: `http://127.0.0.1:${primaryPort}`,
+      apiKey: 'hf_token',
+      provider: 'generic-chat-completion-api',
+      model: 'hf-default',
+    };
+    proxyServer = startOpenAICompatProxyServer({
+      profile,
+      port: proxyPort,
+      authToken: 'test-proxy-token',
+    });
+
+    const response = await requestProxy({
+      model: 'deepseek:deepseek-reasoner',
+      messages: [{ role: 'user', content: 'hello' }],
+    });
+    expect(response.status).toBe(200);
+    expect(await response.json()).toMatchObject({
+      content: [{ type: 'text', text: 'Reply from secondary' }],
+    });
+    expect(hits).toEqual(['secondary']);
+    expect(bodies[0]?.body).toMatchObject({ model: 'deepseek-reasoner' });
+  });
+
+  it('routes thinking requests through the configured think scenario', async () => {
+    const primaryPort = await getPort();
+    const thinkPort = await getPort();
+    const hits: string[] = [];
+    const bodies: Array<{ label: string; body: unknown }> = [];
+    await startMockUpstream(primaryPort, 'primary', hits, bodies);
+    await startMockUpstream(thinkPort, 'thinker', hits, bodies);
+
+    const primarySettings = writeSettings('hf', {
+      ANTHROPIC_BASE_URL: `http://127.0.0.1:${primaryPort}`,
+      ANTHROPIC_AUTH_TOKEN: 'hf_token',
+      ANTHROPIC_MODEL: 'hf-default',
+      CCS_DROID_PROVIDER: 'generic-chat-completion-api',
+    });
+    const thinkSettings = writeSettings('thinker', {
+      ANTHROPIC_BASE_URL: `http://127.0.0.1:${thinkPort}`,
+      ANTHROPIC_AUTH_TOKEN: 'think_token',
+      ANTHROPIC_MODEL: 'deepseek-reasoner',
+      CCS_DROID_PROVIDER: 'generic-chat-completion-api',
+    });
+
+    fs.writeFileSync(
+      path.join(tempDir, '.ccs', 'config.json'),
+      JSON.stringify(
+        {
+          profiles: { hf: primarySettings, thinker: thinkSettings },
+          proxy: {
+            routing: {
+              think: 'thinker:deepseek-reasoner',
+            },
+          },
+        },
+        null,
+        2
+      ),
+      'utf8'
+    );
+
+    const profile: OpenAICompatProfileConfig = {
+      profileName: 'hf',
+      settingsPath: primarySettings,
+      baseUrl: `http://127.0.0.1:${primaryPort}`,
+      apiKey: 'hf_token',
+      provider: 'generic-chat-completion-api',
+      model: 'hf-default',
+    };
+    proxyServer = startOpenAICompatProxyServer({
+      profile,
+      port: proxyPort,
+      authToken: 'test-proxy-token',
+    });
+
+    const response = await requestProxy({
+      model: 'hf-default',
+      thinking: { type: 'enabled', budget_tokens: 9000 },
+      messages: [{ role: 'user', content: 'think hard' }],
+    });
+
+    expect(response.status).toBe(200);
+    expect(await response.json()).toMatchObject({
+      content: [{ type: 'text', text: 'Reply from thinker' }],
+    });
+    expect(hits).toEqual(['thinker']);
+    expect(bodies[0]?.body).toMatchObject({ model: 'deepseek-reasoner' });
+  });
+});