Refactor proxy-core surfaces and transformer boundaries (#163)

* refactor: add canonical request contracts * test: harden transformer boundary contracts * refactor: adapt protocol transformers to canonical contract * refactor: add cli profile registry * refactor: extract provider profiles * refactor: add conductor and provider executors * refactor: add models surface * refactor: migrate responses surface * refactor: migrate chat surface * refactor: migrate gemini surface * refactor: migrate file surfaces * test: align runtime executor undici mock * fix: preserve compact SSE closeout payloads * feat: support all-api-hub v2 backup imports * fix: aggregate codex oauth SSE final payloads * fix: harden responses SSE failure handling
2026-03-19 16:07:56 +08:00
parent c0c993f448
commit 24483c28eb
75 changed files with 8333 additions and 3574 deletions
@@ -0,0 +1,241 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { DefaultProxyConductor } from './DefaultProxyConductor.js';
+import { terminalStreamFailure } from './streamTermination.js';
+
+const baseSelectedChannel = {
+  channel: { id: 11, routeId: 22 },
+  site: { id: 44, name: 'demo-site', url: 'https://upstream.example.com', platform: 'openai' },
+  account: { id: 33, username: 'demo-user' },
+  tokenName: 'default',
+  tokenValue: 'sk-demo',
+  actualModel: 'upstream-gpt',
+};
+
+describe('DefaultProxyConductor', () => {
+  it('returns the first selected channel when the first attempt succeeds', async () => {
+    const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
+    const selectNextChannel = vi.fn();
+    const recordSuccess = vi.fn().mockResolvedValue(undefined);
+    const recordFailure = vi.fn().mockResolvedValue(undefined);
+    const conductor = new DefaultProxyConductor({
+      selectChannel,
+      selectNextChannel,
+      recordSuccess,
+      recordFailure,
+    });
+    const attempt = vi.fn().mockResolvedValue({
+      ok: true,
+      response: new Response('ok', { status: 200 }),
+      latencyMs: 12,
+      cost: 0.25,
+    });
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt,
+    });
+
+    expect(result).toMatchObject({
+      ok: true,
+      selected: baseSelectedChannel,
+      attempts: 1,
+    });
+    expect(selectChannel).toHaveBeenCalledWith('gpt-5.4', undefined);
+    expect(selectNextChannel).not.toHaveBeenCalled();
+    expect(recordFailure).not.toHaveBeenCalled();
+    expect(recordSuccess).toHaveBeenCalledWith(11, {
+      latencyMs: 12,
+      cost: 0.25,
+    });
+  });
+
+  it('retries on the same channel when the attempt asks for a same-channel retry', async () => {
+    const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
+    const selectNextChannel = vi.fn();
+    const recordSuccess = vi.fn().mockResolvedValue(undefined);
+    const recordFailure = vi.fn().mockResolvedValue(undefined);
+    const conductor = new DefaultProxyConductor({
+      selectChannel,
+      selectNextChannel,
+      recordSuccess,
+      recordFailure,
+    });
+    const attempt = vi.fn()
+      .mockResolvedValueOnce({
+        ok: false,
+        action: 'retry_same_channel',
+        status: 429,
+        rawErrorText: 'rate limited',
+      })
+      .mockResolvedValueOnce({
+        ok: true,
+        response: new Response('ok', { status: 200 }),
+      });
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt,
+    });
+
+    expect(result).toMatchObject({
+      ok: true,
+      attempts: 2,
+    });
+    expect(selectNextChannel).not.toHaveBeenCalled();
+    expect(attempt).toHaveBeenCalledTimes(2);
+    expect(recordFailure).toHaveBeenCalledWith(11, {
+      status: 429,
+      rawErrorText: 'rate limited',
+    });
+  });
+
+  it('fails over to the next channel when the attempt asks for failover', async () => {
+    const nextSelectedChannel = {
+      ...baseSelectedChannel,
+      channel: { id: 12, routeId: 22 },
+      tokenValue: 'sk-next',
+    };
+    const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
+    const selectNextChannel = vi.fn().mockResolvedValue(nextSelectedChannel);
+    const recordSuccess = vi.fn().mockResolvedValue(undefined);
+    const recordFailure = vi.fn().mockResolvedValue(undefined);
+    const conductor = new DefaultProxyConductor({
+      selectChannel,
+      selectNextChannel,
+      recordSuccess,
+      recordFailure,
+    });
+    const attempt = vi.fn()
+      .mockResolvedValueOnce({
+        ok: false,
+        action: 'failover',
+        status: 503,
+        rawErrorText: 'upstream unavailable',
+      })
+      .mockResolvedValueOnce({
+        ok: true,
+        response: new Response('ok', { status: 200 }),
+      });
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt,
+    });
+
+    expect(result).toMatchObject({
+      ok: true,
+      selected: nextSelectedChannel,
+      attempts: 2,
+    });
+    expect(selectNextChannel).toHaveBeenCalledWith('gpt-5.4', [11], undefined);
+    expect(recordFailure).toHaveBeenCalledWith(11, {
+      status: 503,
+      rawErrorText: 'upstream unavailable',
+    });
+    expect(recordSuccess).toHaveBeenCalledWith(12, {
+      latencyMs: null,
+      cost: null,
+    });
+  });
+
+  it('refreshes auth on 401 and retries the same channel with the refreshed selection', async () => {
+    const refreshedChannel = {
+      ...baseSelectedChannel,
+      tokenValue: 'sk-refreshed',
+    };
+    const refreshAuth = vi.fn().mockResolvedValue(refreshedChannel);
+    const conductor = new DefaultProxyConductor({
+      selectChannel: vi.fn().mockResolvedValue(baseSelectedChannel),
+      selectNextChannel: vi.fn(),
+      recordSuccess: vi.fn().mockResolvedValue(undefined),
+      recordFailure: vi.fn().mockResolvedValue(undefined),
+      refreshAuth,
+    });
+    const attempt = vi.fn()
+      .mockResolvedValueOnce({
+        ok: false,
+        action: 'refresh_auth',
+        status: 401,
+        rawErrorText: 'expired token',
+      })
+      .mockResolvedValueOnce({
+        ok: true,
+        response: new Response('ok', { status: 200 }),
+      });
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt,
+    });
+
+    expect(result).toMatchObject({
+      ok: true,
+      selected: refreshedChannel,
+      attempts: 2,
+    });
+    expect(refreshAuth).toHaveBeenCalledWith(baseSelectedChannel, {
+      status: 401,
+      rawErrorText: 'expired token',
+    });
+  });
+
+  it('returns a no_channel result when no channel is available', async () => {
+    const conductor = new DefaultProxyConductor({
+      selectChannel: vi.fn().mockResolvedValue(null),
+      selectNextChannel: vi.fn(),
+      recordSuccess: vi.fn(),
+      recordFailure: vi.fn(),
+      previewSelectedChannel: vi.fn().mockResolvedValue(null),
+    });
+
+    expect(await conductor.previewSelectedChannel('gpt-5.4')).toBe(null);
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt: vi.fn(),
+    });
+
+    expect(result).toEqual({
+      ok: false,
+      reason: 'no_channel',
+      attempts: 0,
+    });
+  });
+
+  it('propagates terminal stream failures and calls the terminal failure hook', async () => {
+    const onTerminalFailure = vi.fn().mockResolvedValue(undefined);
+    const conductor = new DefaultProxyConductor({
+      selectChannel: vi.fn().mockResolvedValue(baseSelectedChannel),
+      selectNextChannel: vi.fn(),
+      recordSuccess: vi.fn(),
+      recordFailure: vi.fn().mockResolvedValue(undefined),
+    });
+    const attempt = vi.fn().mockResolvedValue({
+      ok: false,
+      ...terminalStreamFailure({
+        status: 502,
+        rawErrorText: 'stream disconnected before completion',
+      }),
+    });
+
+    const result = await conductor.execute({
+      requestedModel: 'gpt-5.4',
+      attempt,
+      onTerminalFailure,
+    });
+
+    expect(result).toEqual({
+      ok: false,
+      reason: 'terminal',
+      selected: baseSelectedChannel,
+      status: 502,
+      rawErrorText: 'stream disconnected before completion',
+      attempts: 1,
+    });
+    expect(onTerminalFailure).toHaveBeenCalledWith(baseSelectedChannel, {
+      status: 502,
+      rawErrorText: 'stream disconnected before completion',
+    });
+  });
+});
@@ -0,0 +1,127 @@
+import {
+  failureActionOf,
+  isTerminalFailure,
+  shouldFailover,
+  shouldRefreshAuth,
+  shouldRetrySameChannel,
+} from './retryPolicy.js';
+import type { ExecuteInput, ExecuteResult, ProxyConductorDependencies, SelectedChannelLike } from './types.js';
+import { recordFailedAttempt, recordSuccessfulAttempt } from './usageHooks.js';
+
+export class DefaultProxyConductor {
+  constructor(private readonly deps: ProxyConductorDependencies) {}
+
+  async previewSelectedChannel(requestedModel: string, downstreamPolicy?: unknown): Promise<SelectedChannelLike | null> {
+    if (this.deps.previewSelectedChannel) {
+      return this.deps.previewSelectedChannel(requestedModel, downstreamPolicy);
+    }
+    return this.deps.selectChannel(requestedModel, downstreamPolicy);
+  }
+
+  async execute(input: ExecuteInput): Promise<ExecuteResult> {
+    const excludeChannelIds: number[] = [];
+    let attempts = 0;
+    let selected = await this.deps.selectChannel(input.requestedModel, input.downstreamPolicy);
+    if (!selected) {
+      return {
+        ok: false,
+        reason: 'no_channel',
+        attempts: 0,
+      };
+    }
+
+    while (selected) {
+      const result = await input.attempt({
+        selected,
+        attemptIndex: attempts,
+        excludeChannelIds: [...excludeChannelIds],
+      });
+      attempts += 1;
+
+      if (result.ok) {
+        await recordSuccessfulAttempt(this.deps, selected.channel.id, {
+          latencyMs: result.latencyMs ?? null,
+          cost: result.cost ?? null,
+        });
+        return {
+          ok: true,
+          selected,
+          response: result.response,
+          attempts,
+        };
+      }
+
+      const action = failureActionOf(result);
+      await recordFailedAttempt(this.deps, selected.channel.id, {
+        status: result.status,
+        rawErrorText: result.rawErrorText,
+      });
+
+      if (isTerminalFailure(action)) {
+        await input.onTerminalFailure?.(selected, {
+          status: result.status,
+          rawErrorText: result.rawErrorText,
+        });
+        return {
+          ok: false,
+          reason: 'terminal',
+          selected,
+          status: result.status,
+          rawErrorText: result.rawErrorText,
+          attempts,
+        };
+      }
+
+      if (shouldRetrySameChannel(action)) {
+        continue;
+      }
+
+      if (shouldRefreshAuth(action) && this.deps.refreshAuth) {
+        const refreshed = await this.deps.refreshAuth(selected, {
+          status: result.status,
+          rawErrorText: result.rawErrorText,
+        });
+        if (refreshed) {
+          selected = refreshed;
+          continue;
+        }
+      }
+
+      if (shouldFailover(action)) {
+        excludeChannelIds.push(selected.channel.id);
+        const next = await this.deps.selectNextChannel(
+          input.requestedModel,
+          excludeChannelIds,
+          input.downstreamPolicy,
+        );
+        if (!next) {
+          return {
+            ok: false,
+            reason: 'failed',
+            selected,
+            status: result.status,
+            rawErrorText: result.rawErrorText,
+            attempts,
+          };
+        }
+        selected = next;
+        continue;
+      }
+
+      return {
+        ok: false,
+        reason: 'failed',
+        selected,
+        status: result.status,
+        rawErrorText: result.rawErrorText,
+        attempts,
+      };
+    }
+
+    return {
+      ok: false,
+      reason: 'failed',
+      attempts,
+    };
+  }
+}
@@ -0,0 +1,21 @@
+import type { AttemptFailure, AttemptFailureAction } from './types.js';
+
+export function failureActionOf(result: AttemptFailure): AttemptFailureAction {
+  return result.action;
+}
+
+export function shouldRetrySameChannel(action: AttemptFailureAction): boolean {
+  return action === 'retry_same_channel';
+}
+
+export function shouldRefreshAuth(action: AttemptFailureAction): boolean {
+  return action === 'refresh_auth';
+}
+
+export function shouldFailover(action: AttemptFailureAction): boolean {
+  return action === 'failover' || action === 'refresh_auth';
+}
+
+export function isTerminalFailure(action: AttemptFailureAction): boolean {
+  return action === 'terminal';
+}
@@ -0,0 +1,10 @@
+export function terminalStreamFailure(input: {
+  status?: number;
+  rawErrorText?: string;
+}) {
+  return {
+    action: 'terminal' as const,
+    ...(input.status !== undefined ? { status: input.status } : {}),
+    ...(input.rawErrorText ? { rawErrorText: input.rawErrorText } : {}),
+  };
+}
@@ -0,0 +1,80 @@
+export type SelectedChannelLike = {
+  channel: { id: number; routeId?: number };
+  site: Record<string, unknown>;
+  account: Record<string, unknown>;
+  tokenName?: string;
+  tokenValue?: string;
+  actualModel?: string;
+};
+
+export type AttemptSuccess = {
+  ok: true;
+  response: Response;
+  latencyMs?: number | null;
+  cost?: number | null;
+};
+
+export type AttemptFailureAction =
+  | 'retry_same_channel'
+  | 'refresh_auth'
+  | 'failover'
+  | 'terminal'
+  | 'stop';
+
+export type AttemptFailure = {
+  ok: false;
+  action: AttemptFailureAction;
+  status?: number;
+  rawErrorText?: string;
+  error?: unknown;
+};
+
+export type AttemptResult = AttemptSuccess | AttemptFailure;
+
+export type ExecuteAttemptContext = {
+  selected: SelectedChannelLike;
+  attemptIndex: number;
+  excludeChannelIds: number[];
+};
+
+export type ProxyConductorDependencies = {
+  selectChannel: (requestedModel: string, downstreamPolicy?: unknown) => Promise<SelectedChannelLike | null>;
+  previewSelectedChannel?: (requestedModel: string, downstreamPolicy?: unknown) => Promise<SelectedChannelLike | null>;
+  selectNextChannel: (
+    requestedModel: string,
+    excludeChannelIds: number[],
+    downstreamPolicy?: unknown,
+  ) => Promise<SelectedChannelLike | null>;
+  recordSuccess?: (channelId: number, metrics: { latencyMs: number | null; cost: number | null }) => Promise<void> | void;
+  recordFailure?: (channelId: number, failure: { status?: number; rawErrorText?: string }) => Promise<void> | void;
+  refreshAuth?: (
+    selected: SelectedChannelLike,
+    failure: { status?: number; rawErrorText?: string },
+  ) => Promise<SelectedChannelLike | null>;
+};
+
+export type ExecuteInput = {
+  requestedModel: string;
+  downstreamPolicy?: unknown;
+  attempt: (context: ExecuteAttemptContext) => Promise<AttemptResult>;
+  onTerminalFailure?: (
+    selected: SelectedChannelLike,
+    failure: { status?: number; rawErrorText?: string },
+  ) => Promise<void> | void;
+};
+
+export type ExecuteResult =
+  | {
+    ok: true;
+    selected: SelectedChannelLike;
+    response: Response;
+    attempts: number;
+  }
+  | {
+    ok: false;
+    reason: 'no_channel' | 'failed' | 'terminal';
+    selected?: SelectedChannelLike;
+    status?: number;
+    rawErrorText?: string;
+    attempts: number;
+  };
@@ -0,0 +1,20 @@
+import type { ProxyConductorDependencies } from './types.js';
+
+export async function recordSuccessfulAttempt(
+  deps: ProxyConductorDependencies,
+  channelId: number,
+  metrics: { latencyMs?: number | null; cost?: number | null },
+): Promise<void> {
+  await deps.recordSuccess?.(channelId, {
+    latencyMs: metrics.latencyMs ?? null,
+    cost: metrics.cost ?? null,
+  });
+}
+
+export async function recordFailedAttempt(
+  deps: ProxyConductorDependencies,
+  channelId: number,
+  failure: { status?: number; rawErrorText?: string },
+): Promise<void> {
+  await deps.recordFailure?.(channelId, failure);
+}