支持responses

2026-08-05 16:43:18 +08:00
parent 15176a3238
commit 7ec6914872
15 changed files with 1560 additions and 101 deletions
@@ -0,0 +1,112 @@
+import { randomUUID } from "node:crypto";
+import type { ServerResponse } from "node:http";
+import type { FastifyReply } from "fastify";
+import { FinalTurnAccumulator } from "./final-turn.js";
+import type { CommandUsage } from "./openai.js";
+
+export class ChatCompletionSseWriter {
+  private readonly id = `chatcmpl-local-${randomUUID().replaceAll("-", "")}`;
+  private readonly created = Math.floor(Date.now() / 1000);
+  private readonly accumulator = new FinalTurnAccumulator();
+  private outputText = "";
+
+  constructor(
+    private readonly response: ServerResponse,
+    private readonly model: string,
+  ) {}
+
+  begin(): void {
+    this.chunk({ role: "assistant", content: "" }, null);
+  }
+
+  commandEvent(event: Record<string, unknown>): void {
+    const chunks = this.accumulator.event(event);
+    if (chunks) this.addText(chunks);
+  }
+
+  finish(finalText: string, usage?: CommandUsage, includeUsage = false): void {
+    this.ensureFinalText(finalText);
+    this.chunk({}, "stop");
+
+    if (includeUsage) {
+      const promptTokens = usage?.inputTokens ?? 0;
+      const completionTokens = usage?.outputTokens ?? 0;
+      this.write({
+        ...this.base(),
+        choices: [],
+        usage: {
+          prompt_tokens: promptTokens,
+          completion_tokens: completionTokens,
+          total_tokens: promptTokens + completionTokens,
+        },
+      });
+    }
+
+    this.writeDone();
+  }
+
+  error(error: object): void {
+    this.write(error);
+    this.writeDone();
+  }
+
+  private addText(chunks: string[]): void {
+    for (const delta of chunks) {
+      if (delta === "") continue;
+      this.outputText += delta;
+      this.chunk({ content: delta }, null);
+    }
+  }
+
+  private ensureFinalText(finalText: string): void {
+    if (this.outputText === finalText) return;
+    if (finalText.startsWith(this.outputText)) {
+      this.addText([finalText.slice(this.outputText.length)]);
+      return;
+    }
+    throw new Error("Command Code finalText does not match the final structured turn text");
+  }
+
+  private chunk(delta: Record<string, unknown>, finishReason: "stop" | null): void {
+    this.write({
+      ...this.base(),
+      choices: [{ index: 0, delta, finish_reason: finishReason }],
+    });
+  }
+
+  private base() {
+    return {
+      id: this.id,
+      object: "chat.completion.chunk",
+      created: this.created,
+      model: this.model,
+    };
+  }
+
+  private write(value: object): void {
+    if (this.response.destroyed || this.response.writableEnded) return;
+    this.response.write(`data: ${JSON.stringify(value)}\n\n`);
+  }
+
+  private writeDone(): void {
+    if (this.response.destroyed || this.response.writableEnded) return;
+    this.response.end("data: [DONE]\n\n");
+  }
+}
+
+export function startChatCompletionSse(
+  reply: FastifyReply,
+  model: string,
+): ChatCompletionSseWriter {
+  reply.hijack();
+  reply.raw.writeHead(200, {
+    "Content-Type": "text/event-stream; charset=utf-8",
+    "Cache-Control": "no-cache, no-transform",
+    Connection: "keep-alive",
+    "X-Accel-Buffering": "no",
+  });
+  reply.raw.flushHeaders();
+  const writer = new ChatCompletionSseWriter(reply.raw, model);
+  writer.begin();
+  return writer;
+}
@@ -19,6 +19,13 @@ export interface CommandResult {
  usage?: CommandUsage;
 }

+export interface CommandRunOptions {
+  onEvent?: (event: Record<string, unknown>) => void;
+  timeoutMs?: number;
+}
+
+export type CommandFailureKind = "cancelled" | "max_turns" | "timeout" | "command_error";
+
 export interface InstallationStatus {
  installed: boolean;
  authenticated: boolean;
@@ -29,6 +36,7 @@ export class CommandCodeError extends Error {
  constructor(
    message: string,
    readonly exitCode?: number | null,
+    readonly kind: CommandFailureKind = "command_error",
  ) {
    super(message);
    this.name = "CommandCodeError";
@@ -70,7 +78,13 @@ export async function runCommandCode(
  effort: string,
  prompt: string,
  signal: AbortSignal,
+  options: CommandRunOptions = {},
 ): Promise<CommandResult> {
+  const timeoutMs = options.timeoutMs ?? config.timeout_seconds * 1000;
+  if (timeoutMs <= 0) {
+    throw new CommandCodeError("Command Code request timed out", undefined, "timeout");
+  }
+
  const args = [
    "-p",
    "--output-format", "json",
@@ -105,6 +119,7 @@ export async function runCommandCode(
    }

    let settled = false;
+    let timedOut = false;
    let result: ResultFrame | undefined;
    let parseError: Error | undefined;
    let stderr = "";
@@ -128,9 +143,10 @@ export async function runCommandCode(
    };

    const timeout = setTimeout(() => {
-      renderer.fail(`超过 ${config.timeout_seconds} 秒总超时`);
+      timedOut = true;
+      renderer.fail(`超过 ${(timeoutMs / 1000).toFixed(1)} 秒总超时`);
      cancelEscalation ??= terminateChild(child);
-    }, config.timeout_seconds * 1000);
+    }, timeoutMs);
    timeout.unref();

    signal.addEventListener("abort", onAbort, { once: true });
@@ -180,6 +196,7 @@ export async function runCommandCode(
            lastMessageText = undefined;
          }
          renderer.event(event);
+          options.onEvent?.(event);
        } else if (frame.type === "result") {
          result = frame as unknown as ResultFrame;
        }
@@ -192,16 +209,25 @@ export async function runCommandCode(
    child.once("close", (code, closeSignal) => {
      lines.close();

+      if (timedOut) {
+        finish(new CommandCodeError("Command Code request timed out", code, "timeout"));
+        return;
+      }
+
      if (signal.aborted) {
        renderer.fail("请求已取消");
-        finish(new CommandCodeError("Command Code request cancelled", code));
+        finish(new CommandCodeError("Command Code request cancelled", code, "cancelled"));
        return;
      }

      if (code !== 0) {
        const detail = stderr.trim().split("\n").slice(-3).join(" | ");
        renderer.fail(`退出码 ${String(code)}${closeSignal ? `，信号 ${closeSignal}` : ""}${detail ? `：${detail}` : ""}`);
-        finish(new CommandCodeError("Command Code request failed", code));
+        finish(new CommandCodeError(
+          "Command Code request failed",
+          code,
+          code === 8 ? "max_turns" : "command_error",
+        ));
        return;
      }

@@ -224,7 +250,11 @@ export async function runCommandCode(

      if (result.subtype !== "success") {
        renderer.fail(`结果状态 ${result.subtype}`);
-        finish(new CommandCodeError("Command Code request failed", code));
+        finish(new CommandCodeError(
+          "Command Code request failed",
+          code,
+          result.subtype === "max_turns" ? "max_turns" : "command_error",
+        ));
        return;
      }

@@ -16,6 +16,7 @@ const configSchema = z.object({
  timeout_seconds: z.number().int().positive().default(1800),
  max_request_bytes: z.number().int().positive().default(20 * 1024 * 1024),
  max_turns: z.number().int().positive().default(100),
+  response_store_directory: z.string().min(1).default(".command-code-openai-bridge/responses"),
  permission_mode: z.enum(["default", "standard", "plan", "auto-accept", "dont-ask"]).default("auto-accept"),
  dangerously_skip_permissions: z.boolean().default(false),
  models: z.record(z.string().min(1), modelSchema).refine(
@@ -26,6 +27,7 @@ const configSchema = z.object({

 export type BridgeConfig = z.infer<typeof configSchema> & {
  configDirectory: string;
+  resolvedResponseStoreDirectory: string;
  resolvedWorkingDirectory: string;
 };

@@ -38,6 +40,7 @@ export async function loadConfig(configPath: string): Promise<BridgeConfig> {
  return {
    ...parsed,
    configDirectory,
+    resolvedResponseStoreDirectory: path.resolve(configDirectory, parsed.response_store_directory),
    resolvedWorkingDirectory: path.resolve(configDirectory, parsed.command_code_working_directory),
  };
 }
@@ -0,0 +1,43 @@
+export class FinalTurnAccumulator {
+  private deltas: string[] = [];
+  private messageText: string | undefined;
+  finalChunks: string[] | undefined;
+
+  event(event: Record<string, unknown>): string[] | undefined {
+    if (event.type === "turn_start") {
+      this.deltas = [];
+      this.messageText = undefined;
+      return undefined;
+    }
+    if (event.type === "text_delta" && typeof event.delta === "string") {
+      this.deltas.push(event.delta);
+      return undefined;
+    }
+    if (event.type === "message_end") {
+      this.messageText = extractMessageText(event);
+      return undefined;
+    }
+    if (event.type !== "turn_end" || event.hadToolCalls !== false) return undefined;
+
+    const deltaText = this.deltas.join("");
+    const finalText = this.messageText ?? deltaText;
+    this.finalChunks = deltaText === finalText && this.deltas.length > 0
+      ? [...this.deltas]
+      : [finalText];
+    return this.finalChunks;
+  }
+}
+
+function extractMessageText(event: Record<string, unknown>): string | undefined {
+  if (!Array.isArray(event.content)) return undefined;
+  const text = event.content
+    .filter(isRecord)
+    .filter((part) => part.type === "text" && typeof part.text === "string")
+    .map((part) => String(part.text))
+    .join("");
+  return text === "" ? undefined : text;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
@@ -17,8 +17,8 @@ export const chatCompletionRequestSchema = z.object({
  stream: z.boolean().optional().default(false),
  stream_options: z.object({
    include_usage: z.boolean().optional().default(false),
-  }).passthrough().optional(),
-}).passthrough();
+  }).strict().optional(),
+}).strict();

 export type ChatCompletionRequest = z.infer<typeof chatCompletionRequestSchema>;

@@ -75,59 +75,11 @@ export function completionResponse(model: string, content: string, usage?: Comma
  };
 }

-export function completionStreamResponse(
-  model: string,
-  content: string,
-  usage?: CommandUsage,
-  includeUsage = false,
-): string {
-  const id = `chatcmpl-local-${randomUUID().replaceAll("-", "")}`;
-  const created = Math.floor(Date.now() / 1000);
-  const base = { id, object: "chat.completion.chunk", created, model };
-  const chunks: object[] = [
-    {
-      ...base,
-      choices: [{
-        index: 0,
-        delta: { role: "assistant", content: "" },
-        finish_reason: null,
-      }],
-    },
-    {
-      ...base,
-      choices: [{
-        index: 0,
-        delta: { content },
-        finish_reason: null,
-      }],
-    },
-    {
-      ...base,
-      choices: [{
-        index: 0,
-        delta: {},
-        finish_reason: "stop",
-      }],
-    },
-  ];
-
-  if (includeUsage) {
-    const promptTokens = usage?.inputTokens ?? 0;
-    const completionTokens = usage?.outputTokens ?? 0;
-    chunks.push({
-      ...base,
-      choices: [],
-      usage: {
-        prompt_tokens: promptTokens,
-        completion_tokens: completionTokens,
-        total_tokens: promptTokens + completionTokens,
-      },
-    });
-  }
-
-  return `${chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join("")}data: [DONE]\n\n`;
-}
-
-export function openAIError(message: string, code: string, type = "invalid_request_error") {
-  return { error: { message, type, code } };
+export function openAIError(
+  message: string,
+  code: string,
+  type = "invalid_request_error",
+  param: string | null = null,
+) {
+  return { error: { message, type, param, code } };
 }
@@ -0,0 +1,23 @@
+export class RequestCoordinator {
+  private activeController: AbortController | undefined;
+
+  get busy(): boolean {
+    return this.activeController !== undefined;
+  }
+
+  begin(): AbortController | undefined {
+    if (this.activeController) return undefined;
+    this.activeController = new AbortController();
+    return this.activeController;
+  }
+
+  finish(controller: AbortController): void {
+    if (this.activeController === controller) this.activeController = undefined;
+  }
+
+  abortActive(reason = "bridge interrupted"): boolean {
+    if (!this.activeController) return false;
+    this.activeController.abort(reason);
+    return true;
+  }
+}
@@ -1,5 +1,6 @@
 import Fastify, { type FastifyInstance } from "fastify";
 import { ZodError } from "zod";
+import { startChatCompletionSse, type ChatCompletionSseWriter } from "./chat-stream.js";
 import type { BridgeConfig } from "./config.js";
 import {
  CommandCodeError,
@@ -11,13 +12,10 @@ import {
  buildCommandPrompt,
  chatCompletionRequestSchema,
  completionResponse,
-  completionStreamResponse,
  openAIError,
 } from "./openai.js";
-
-interface ActiveRequest {
-  abortController: AbortController;
-}
+import { RequestCoordinator } from "./request-coordinator.js";
+import { registerResponseRoutes } from "./responses.js";

 export interface BridgeServer {
  app: FastifyInstance;
@@ -32,7 +30,7 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
    bodyLimit: config.max_request_bytes,
    requestTimeout: 0,
  });
-  let active: ActiveRequest | undefined;
+  const coordinator = new RequestCoordinator();

  app.setErrorHandler((error, _request, reply) => {
    const fastifyError = error as Error & { code?: string; statusCode?: number };
@@ -56,7 +54,7 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
  app.get("/health", async () => ({
    status: installation.installed && installation.authenticated ? "ok" : "degraded",
    command_code: installation,
-    busy: active !== undefined,
+    busy: coordinator.busy,
  }));

  app.get("/v1/models", async () => ({
@@ -70,7 +68,7 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
  }));

  app.post("/v1/chat/completions", async (request, reply) => {
-    if (active) {
+    if (coordinator.busy) {
      return reply.status(429).send(openAIError(
        "Another Command Code request is already running",
        "busy",
@@ -98,6 +96,19 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
    try {
      parsed = chatCompletionRequestSchema.parse(request.body);
    } catch (error) {
+      const unknown = error instanceof ZodError
+        ? error.issues.find((issue) => issue.code === "unrecognized_keys")
+        : undefined;
+      if (unknown?.code === "unrecognized_keys") {
+        const key = unknown.keys[0] ?? "unknown";
+        const param = [...unknown.path, key].join(".");
+        return reply.status(400).send(openAIError(
+          `Unsupported parameter: '${param}'`,
+          "unsupported_parameter",
+          "invalid_request_error",
+          param,
+        ));
+      }
      const message = error instanceof ZodError
        ? error.issues.map((issue) => `${issue.path.join(".")}: ${issue.message}`).join("; ")
        : "Invalid request body";
@@ -112,8 +123,14 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
      ));
    }

-    const abortController = new AbortController();
-    active = { abortController };
+    const abortController = coordinator.begin();
+    if (!abortController) {
+      return reply.status(429).send(openAIError(
+        "Another Command Code request is already running",
+        "busy",
+        "server_error",
+      ));
+    }
    let responseCompleted = false;

    const cancelOnDisconnect = () => {
@@ -122,31 +139,38 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
    request.raw.once("aborted", cancelOnDisconnect);
    reply.raw.once("close", cancelOnDisconnect);

+    let writer: ChatCompletionSseWriter | undefined;
    try {
+      if (parsed.stream) writer = startChatCompletionSse(reply, parsed.model);
      const result = await runCommandCode(
        config,
        model.cli_model,
        model.effort,
        buildCommandPrompt(parsed),
        abortController.signal,
+        writer ? { onEvent: (event) => writer?.commandEvent(event) } : {},
      );
      responseCompleted = true;
-      if (parsed.stream) {
-        return reply
-          .type("text/event-stream; charset=utf-8")
-          .header("Cache-Control", "no-cache")
-          .header("Connection", "keep-alive")
-          .send(completionStreamResponse(
-            parsed.model,
-            result.finalText,
-            result.usage,
-            parsed.stream_options?.include_usage,
-          ));
+      if (writer) {
+        writer.finish(
+          result.finalText,
+          result.usage,
+          parsed.stream_options?.include_usage,
+        );
+        return reply;
      }
      return reply.send(completionResponse(parsed.model, result.finalText, result.usage));
    } catch (error) {
      responseCompleted = true;
      if (abortController.signal.aborted) {
+        if (writer) {
+          writer.error(openAIError(
+            "Command Code request cancelled",
+            "cancelled",
+            "server_error",
+          ));
+          return reply;
+        }
        if (!reply.raw.destroyed) {
          return reply.status(499).send(openAIError(
            "Command Code request cancelled",
@@ -164,6 +188,14 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
        : code === 10
          ? "insufficient_credits"
          : "command_code_error";
+      if (writer) {
+        writer.error(openAIError(
+          "Command Code request failed",
+          errorCode,
+          "server_error",
+        ));
+        return reply;
+      }
      return reply.status(status).send(openAIError(
        "Command Code request failed",
        errorCode,
@@ -172,17 +204,15 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
    } finally {
      request.raw.off("aborted", cancelOnDisconnect);
      reply.raw.off("close", cancelOnDisconnect);
-      active = undefined;
+      coordinator.finish(abortController);
    }
  });

+  await registerResponseRoutes(app, { config, installation, coordinator });
+
  return {
    app,
    installation,
-    abortActive: () => {
-      if (!active) return false;
-      active.abortController.abort("bridge interrupted");
-      return true;
-    },
+    abortActive: () => coordinator.abortActive(),
  };
 }